1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import urllib.request
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:
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=10) as response:
html = response.read().decode("utf-8")
markdown = html_to_markdown(html)
res = FetchSuccess(content=markdown)
_cache[url] = res
return res
except Exception as e:
return FetchError(error=str(e))
|
