import httpx2 import subprocess import re from typing import Dict, Union from dataclasses import dataclass @dataclass class FetchSuccess: """Successful fetch result containing the markdown content.""" content: str @dataclass class FetchError: """Failed fetch result containing the error message.""" error: str FetchResult = Union[FetchSuccess, FetchError] # Memoization cache for web pages _cache: Dict[str, FetchResult] = {} def html_to_markdown(html: str) -> str: """Converts HTML to markdown using pandoc with GFM format.""" try: # Use gfm (GitHub Flavored Markdown) for better compatibility and robust conversion process = subprocess.run( ["pandoc", "-f", "html", "-t", "gfm", "-o", "-"], input=html, capture_output=True, text=True, check=True, encoding="utf-8", ) return process.stdout.strip() except subprocess.CalledProcessError as e: raise RuntimeError(f"Pandoc error: {e}") except Exception as e: raise RuntimeError(f"Unexpected error during markdown conversion: {e}") def fetch_and_markdownify(url: str) -> FetchResult: """Fetches a web page and transforms it to markdown, with memoization.""" if url in _cache: return _cache[url] try: with httpx2.Client(headers={"User-Agent": "Mozilla/5.0"}, timeout=10.0) as client: response = client.get(url) response.raise_for_status() html = response.text markdown = html_to_markdown(html) res = FetchSuccess(content=markdown) _cache[url] = res return res except Exception as e: return FetchError(error=str(e))