import shutil import tempfile import subprocess from pathlib import Path from bs4 import BeautifulSoup try: from markdownify import markdownify as _markdownify except ImportError: # pragma: no cover - optional dep gate _markdownify = None from utils.logger import logger class MarkdownConverter: """ HTML -> Markdown converter. Two engines are available: - ``markdownify`` (default): pure Python, no subprocess, ~10x faster per page. Fidelity is good for documentation HTML. - ``pandoc``: spawns the pandoc binary; higher fidelity for pathological HTML but pays a process-start cost per page. """ def __init__( self, engine="markdownify", pandoc_path="pandoc", timeout=120, media_dir="output/assets/pandoc" ): self.engine = engine self.pandoc_path = pandoc_path self.timeout = timeout self.media_dir = Path(media_dir) self.media_dir.mkdir(parents=True, exist_ok=True) if self.engine == "pandoc": self.validate_pandoc() elif self.engine == "markdownify": if _markdownify is None: raise RuntimeError( "markdownify is not installed. " "pip install markdownify, or pass " "engine='pandoc'." ) else: raise ValueError( f"Unknown markdown engine: {self.engine!r}" ) def validate_pandoc(self): """ Ensure pandoc exists. """ if not shutil.which( self.pandoc_path ): raise RuntimeError( "Pandoc not found. " "Please install pandoc." ) def html_to_markdown( self, html: str, title=None, source_url=None, media_dir=None ): """ Convert HTML -> Markdown using the configured engine. """ html = self.preprocess_html(html) if self.engine == "markdownify": markdown = self._convert_markdownify(html) else: markdown = self._convert_pandoc(html, media_dir) markdown = self.postprocess_markdown(markdown) markdown = self.add_frontmatter( markdown, title=title, source_url=source_url, ) logger.debug("Markdown conversion completed.") return markdown def _convert_markdownify(self, html: str) -> str: """ In-process HTML -> Markdown via the markdownify library. Roughly 10x faster per page than spawning pandoc. """ def code_lang_cb(el): # HTMLCleaner.normalize_code_blocks already sets # data-language on
when it can detect one.
return el.get("data-language") or ""
try:
return _markdownify(
html,
heading_style="ATX",
code_language_callback=code_lang_cb,
bullets="-",
strip=["script", "style"],
)
except Exception as e:
logger.exception(f"markdownify failed: {e}")
raise
def _convert_pandoc(self, html: str, media_dir) -> str:
"""
Fallback engine: shell out to pandoc.
"""
temp_html_path = None
temp_md_path = None
try:
with tempfile.NamedTemporaryFile(
suffix=".html",
mode="w",
encoding="utf-8",
delete=False,
) as temp_html:
temp_html.write(html)
temp_html_path = temp_html.name
temp_md_path = temp_html_path + ".md"
extract_media_dir = (
Path(media_dir) if media_dir else self.media_dir
)
extract_media_dir.mkdir(parents=True, exist_ok=True)
command = [
self.pandoc_path,
temp_html_path,
"-f", "html",
"-t", "gfm",
"--wrap=none",
"--markdown-headings=atx",
f"--extract-media={extract_media_dir}",
"-o", temp_md_path,
]
result = subprocess.run(
command,
capture_output=True,
text=True,
timeout=self.timeout,
)
if result.returncode != 0:
logger.error(f"Pandoc failed: {result.stderr}")
raise RuntimeError(result.stderr)
with open(
temp_md_path,
"r",
encoding="utf-8",
) as f:
return f.read()
except subprocess.TimeoutExpired:
logger.error("Pandoc conversion timeout.")
raise
finally:
self.cleanup_temp_file(temp_html_path)
self.cleanup_temp_file(temp_md_path)
def preprocess_html(
self,
html: str
):
"""
Clean HTML before conversion.
"""
soup = BeautifulSoup(
html,
"lxml"
)
# Remove scripts/styles
for tag in soup.find_all(
[
"script",
"style",
"noscript"
]
):
tag.decompose()
# Normalize code blocks
for pre in soup.find_all("pre"):
code = pre.find("code")
if not code:
continue
classes = code.get(
"class",
[]
)
for cls in classes:
if cls.startswith(
"language-"
):
lang = cls.replace(
"language-",
""
)
pre["data-language"] = lang
return str(soup)
def postprocess_markdown(
self,
markdown: str
):
"""
Normalize markdown output.
"""
lines = markdown.splitlines()
cleaned = []
previous_empty = False
for line in lines:
stripped = line.rstrip()
# Collapse excessive empty lines
if not stripped:
if previous_empty:
continue
previous_empty = True
else:
previous_empty = False
cleaned.append(
stripped
)
markdown = "\n".join(cleaned)
markdown = self.fix_code_fences(
markdown
)
markdown = self.fix_tables(
markdown
)
return markdown.strip()
def fix_code_fences(
self,
markdown: str
):
"""
Improve fenced code blocks.
"""
lines = markdown.splitlines()
output = []
in_code = False
for line in lines:
if line.startswith("```"):
in_code = not in_code
output.append(line)
# Close unclosed code block
if in_code:
output.append("```")
return "\n".join(output)
def fix_tables(
self,
markdown: str
):
"""
Fix malformed markdown tables.
"""
# Future enhancement hook
return markdown
def add_frontmatter(
self,
markdown,
title=None,
source_url=None
):
"""
Add YAML frontmatter.
Values are escaped so that titles or URLs containing
quotes / backslashes / newlines do not produce
broken YAML.
"""
metadata = []
if title:
metadata.append(
f'title: {self._yaml_quote(title)}'
)
if source_url:
metadata.append(
f'source_url: {self._yaml_quote(source_url)}'
)
if not metadata:
return markdown
frontmatter = "---\n"
frontmatter += "\n".join(
metadata
)
frontmatter += "\n---\n\n"
return frontmatter + markdown
def _yaml_quote(
self,
value
):
"""
Safely double-quote a value for YAML frontmatter.
Escapes backslashes, double quotes and control chars
(newline / carriage return / tab) which are otherwise
illegal inside a double-quoted YAML scalar.
"""
text = str(value)
text = (
text
.replace("\\", "\\\\")
.replace('"', '\\"')
.replace("\n", "\\n")
.replace("\r", "\\r")
.replace("\t", "\\t")
)
return f'"{text}"'
def cleanup_temp_file(
self,
path
):
"""
Safely remove temp file.
"""
if not path:
return
try:
Path(path).unlink(
missing_ok=True
)
except Exception as e:
logger.warning(
f"Temp cleanup failed "
f"{path}: {e}"
)
def batch_convert(
self,
html_documents
):
"""
Batch conversion helper.
html_documents:
[
{
"html": "...",
"title": "...",
"url": "..."
}
]
"""
results = []
for doc in html_documents:
try:
markdown = self.html_to_markdown(
html=doc["html"],
title=doc.get("title"),
source_url=doc.get("url")
)
results.append({
"success": True,
"markdown": markdown
})
except Exception as e:
results.append({
"success": False,
"error": str(e)
})
return results