crawler-intra-mart/converter/markdown_converter.py

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 <pre> 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