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