services/cv_converter.py

"""
cv_converter.py
===============
Converts CV / résumé files (PDF, DOCX, DOC) to clean Markdown.

Pipeline
--------
    PDF   → pdfplumber (scan detection) → Marker  (with or without OCR)
    DOCX  → pypandoc   → GitHub-Flavoured Markdown
    DOC   → LibreOffice (→ DOCX) → pypandoc → GFM

Why Markdown as the intermediate format?
  • LLMs understand it natively  → better job-match prompts
  • Sentence-transformers get cleaner text  → better embeddings
  • Renders directly in the browser with zero extra work

Install
-------
    pip install marker-pdf pdfplumber pypandoc
    python -c "import pypandoc; pypandoc.download_pandoc()"

    # For legacy .doc support:
    # Ubuntu/Debian : sudo apt-get install libreoffice
    # macOS         : brew install --cask libreoffice

Usage
-----
    from cv_converter import CVConverter

    converter = CVConverter()
    result    = converter.convert("john_doe_cv.pdf")

    if result:                        # bool(result) == result.success
        print(result.markdown)
        converter.save(result, "john_doe_cv.md")
    else:
        print("Failed:", result.error)
        for w in result.warnings:
            print("Warning:", w)
"""

from __future__ import annotations

import logging
import re
import subprocess
import tempfile
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Optional

logger = logging.getLogger(__name__)


# ─────────────────────────────────────────────────────────────────────────────
# Value types
# ─────────────────────────────────────────────────────────────────────────────

class ConversionMethod(str, Enum):
    MARKER        = "marker"         # Marker – text-based PDF
    MARKER_OCR    = "marker_ocr"     # Marker – forced OCR (scanned PDF)
    PANDOC        = "pandoc"         # Pandoc – DOCX
    PANDOC_VIA_LO = "pandoc_via_lo"  # LibreOffice → Pandoc – legacy DOC


class FileType(str, Enum):
    PDF     = "pdf"
    DOCX    = "docx"
    DOC     = "doc"
    UNKNOWN = "unknown"


@dataclass
class ConversionResult:
    """
    Returned by :py:meth:`CVConverter.convert`.

    Always check ``.success`` (or ``bool(result)``) before reading
    ``.markdown``.

    Attributes
    ----------
    success     : bool            – True when conversion produced output.
    markdown    : str | None      – The Markdown text (None on failure).
    method_used : str | None      – Which pipeline was used.
    file_type   : str | None      – Detected file type ("pdf", "docx", …).
    is_scanned  : bool            – True when OCR was required.
    page_count  : int             – Page count (0 when unknown).
    warnings    : list[str]       – Non-fatal notes (mixed PDF, OCR quality…).
    error       : str | None      – Human-readable error message on failure.
    """
    success:     bool
    markdown:    Optional[str]  = None
    method_used: Optional[str]  = None
    file_type:   Optional[str]  = None
    is_scanned:  bool           = False
    page_count:  int            = 0
    warnings:    list[str]      = field(default_factory=list)
    error:       Optional[str]  = None

    def __bool__(self) -> bool:
        return self.success


# ─────────────────────────────────────────────────────────────────────────────
# Main class
# ─────────────────────────────────────────────────────────────────────────────

class CVConverter:
    """
    Converts CV files (PDF / DOCX / DOC) to clean Markdown.

    Marker ML models are loaded lazily on the first PDF call and then
    cached, so all subsequent conversions in the same process reuse them.

    Parameters
    ----------
    temp_dir : str | None
        Where to place intermediate files (e.g. the .docx produced when
        converting a legacy .doc).  Defaults to the OS temp directory.
    marker_device : str
        ``"cpu"`` or ``"cuda"``.  Passed to Marker when loading models.

    Examples
    --------
    >>> converter = CVConverter()
    >>> result = converter.convert("resume.pdf")
    >>> if result:
    ...     converter.save(result, "resume.md")
    """

    # ── tuneable thresholds ───────────────────────────────────────────────

    # Characters needed on a page before we call it "text bearing"
    MIN_TEXT_CHARS_PER_PAGE: int = 50
    # Fraction of pages that must carry text to skip OCR
    SCANNED_PAGE_RATIO: float = 0.30
    # Hard timeout for the LibreOffice subprocess (seconds)
    LIBREOFFICE_TIMEOUT: int = 60

    # ─────────────────────────────────────────────────────────────────────
    def __init__(
        self,
        temp_dir: Optional[str] = None,
        marker_device: str = "cpu",
    ) -> None:
        self.temp_dir       = Path(temp_dir or tempfile.gettempdir())
        self.marker_device  = marker_device
        self._marker_models = None   # populated on first use

    # ─────────────────────────────────────────────────────────────────────
    # Public API
    # ─────────────────────────────────────────────────────────────────────

    def convert(self, file_path: str | Path) -> ConversionResult:
        """
        Convert a CV file to Markdown.

        Automatically detects the file type and delegates to the correct
        sub-pipeline.  Never raises; all errors are captured in the
        returned :class:`ConversionResult`.

        Parameters
        ----------
        file_path : str | Path
            Path to the CV file (.pdf, .docx, or .doc).

        Returns
        -------
        ConversionResult
        """
        path = Path(file_path)

        # ── pre-flight checks ─────────────────────────────────────────────
        if not path.exists():
            return ConversionResult(
                success=False, error=f"File not found: {path}"
            )
        if not path.is_file():
            return ConversionResult(
                success=False, error=f"Path is not a regular file: {path}"
            )
        if path.stat().st_size == 0:
            return ConversionResult(
                success=False, error=f"File is empty: {path.name}"
            )

        file_type = self._detect_file_type(path)

        if file_type == FileType.PDF:
            return self._convert_pdf(path)

        if file_type in (FileType.DOCX, FileType.DOC):
            return self._convert_word(path, file_type)

        return ConversionResult(
            success=False,
            error=(
                f"Unsupported file type '{path.suffix}'. "
                "Accepted: .pdf, .docx, .doc"
            ),
        )

    def save(
        self,
        result: ConversionResult,
        output_path: str | Path,
    ) -> None:
        """
        Write ``result.markdown`` to *output_path* (UTF-8).

        Raises
        ------
        ValueError
            When ``result.success`` is False.
        """
        if not result.success:
            raise ValueError("Cannot save a failed ConversionResult.")
        out = Path(output_path)
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_text(result.markdown, encoding="utf-8")
        logger.info("Markdown written → %s", out)

    def convert_and_save(
        self,
        file_path: str | Path,
        output_path: str | Path,
    ) -> ConversionResult:
        """Convenience wrapper: convert then save if successful."""
        result = self.convert(file_path)
        if result.success:
            self.save(result, output_path)
        return result

    # ─────────────────────────────────────────────────────────────────────
    # File-type detection
    # ─────────────────────────────────────────────────────────────────────

    @staticmethod
    def _detect_file_type(path: Path) -> FileType:
        return {
            ".pdf":  FileType.PDF,
            ".docx": FileType.DOCX,
            ".doc":  FileType.DOC,
        }.get(path.suffix.lower(), FileType.UNKNOWN)

    # ─────────────────────────────────────────────────────────────────────
    # PDF pipeline
    # ─────────────────────────────────────────────────────────────────────

    def _check_pdf_text_layer(
        self, path: Path
    ) -> tuple[bool, int, list[bool]]:
        """
        Use pdfplumber to probe each page for an embedded text layer.

        Returns
        -------
        (has_text_layer, page_count, per_page_flags)
            * has_text_layer  – True when enough pages carry real text.
            * page_count      – Total pages.
            * per_page_flags  – Per-page bool list (True = text found).

        Raises
        ------
        ValueError
            For password-protected or zero-page PDFs.
        """
        try:
            import pdfplumber
        except ImportError:
            logger.warning(
                "pdfplumber not installed — OCR detection skipped. "
                "Run: pip install pdfplumber"
            )
            return True, 0, []

        try:
            with pdfplumber.open(str(path)) as pdf:
                page_count = len(pdf.pages)

                if page_count == 0:
                    raise ValueError(
                        f"'{path.name}' has zero pages and cannot be converted."
                    )

                per_page: list[bool] = []
                for page in pdf.pages:
                    raw_text = page.extract_text() or ""
                    per_page.append(
                        len(raw_text.strip()) >= self.MIN_TEXT_CHARS_PER_PAGE
                    )

                pages_with_text = sum(per_page)
                ratio           = pages_with_text / page_count
                has_text_layer  = ratio >= self.SCANNED_PAGE_RATIO

                logger.debug(
                    "Text-layer check: %d/%d pages have text (%.0f%%)",
                    pages_with_text, page_count, ratio * 100,
                )
                return has_text_layer, page_count, per_page

        except Exception as exc:
            msg = str(exc).lower()
            if any(k in msg for k in ("password", "encrypted", "decrypt")):
                raise ValueError(
                    f"'{path.name}' is password-protected. "
                    "Please provide an unlocked copy."
                ) from exc
            # Unknown pdfplumber error → assume text-based; Marker will cope.
            logger.warning(
                "pdfplumber probe failed (%s) — assuming text-based PDF.", exc
            )
            return True, 0, []

    def _load_marker_models(self):
        """Lazy-load and cache Marker model dict (once per process).

        PyTorch lazy / meta-device initialization can cause:
          'Cannot copy out of meta tensor; no data!'
        when Marker calls model.to(device) on a model that was built on the
        meta device.  We patch torch.nn.Module.to to fall back to to_empty()
        in that case, then restore the original after loading completes.
        """
        if self._marker_models is None:
            logger.info(
                "Loading Marker models for the first time "
                "(this may take ~10–30 s)…"
            )
            try:
                import torch

                _original_to = torch.nn.Module.to

                def _safe_to(module, *args, **kwargs):
                    try:
                        return _original_to(module, *args, **kwargs)
                    except RuntimeError as exc:
                        if "Cannot copy out of meta tensor" in str(exc):
                            device = args[0] if args else kwargs.get("device", "cpu")
                            logger.debug(
                                "Meta-tensor detected — using to_empty(device=%s)", device
                            )
                            return module.to_empty(device=device)
                        raise

                torch.nn.Module.to = _safe_to
                try:
                    from marker.models import create_model_dict
                    self._marker_models = create_model_dict()
                    logger.info("Marker models loaded and cached.")
                finally:
                    # Always restore the original .to even if loading fails
                    torch.nn.Module.to = _original_to

            except ImportError as exc:
                raise ImportError(
                    "Marker is not installed. Fix: pip install marker-pdf"
                ) from exc
        return self._marker_models


    def _run_marker(self, path: Path, force_ocr: bool = False) -> str:
        """
        Execute Marker and return the raw Markdown string.

        Parameters
        ----------
        force_ocr : bool
            When True, Marker is told to OCR every page regardless of
            whether it detects an embedded text layer.

        Raises
        ------
        RuntimeError
            When Marker raises or returns empty output.
        """
        model_dict = self._load_marker_models()
        try:
            from marker.converters.pdf import PdfConverter           # v1.x API
            from marker.output import text_from_rendered             # v1.x API

            config = {"force_ocr": force_ocr} if force_ocr else {}

            converter  = PdfConverter(model_dict, config=config)
            rendered   = converter(str(path))
            full_text, _images, _meta = text_from_rendered(rendered)

        except Exception as exc:
            raise RuntimeError(
                f"Marker raised an exception on '{path.name}': {exc}"
            ) from exc

        if not full_text or not full_text.strip():
            raise RuntimeError(
                f"Marker produced empty output for '{path.name}'. "
                "The file may be image-only or severely corrupted."
            )
        return full_text

    def _convert_pdf(self, path: Path) -> ConversionResult:
        """
        Full PDF → Markdown pipeline.

        Decision tree
        -------------
        1. pdfplumber inspects every page for a text layer.
        2a. Fully scanned  (< SCANNED_PAGE_RATIO text pages)
               → Marker + OCR.
        2b. Mixed pages    (some pages lack text)
               → Marker + OCR for consistency across all pages.
        2c. Fully digital  → Marker without OCR.
        3. If step 2c fails, automatically retry with OCR as a fallback
           (handles PDFs that report a text layer but are actually images).
        """
        warnings: list[str] = []

        # ── 1. detect text layer ─────────────────────────────────────────
        try:
            has_text, page_count, per_page = self._check_pdf_text_layer(path)
        except ValueError as exc:
            return ConversionResult(success=False, error=str(exc))

        # ── 2. decide OCR strategy ───────────────────────────────────────
        scanned_page_nums: list[int] = [
            i + 1 for i, flag in enumerate(per_page) if not flag
        ]
        is_mixed   = bool(scanned_page_nums) and has_text
        is_scanned = not has_text

        if is_scanned:
            force_ocr  = True
            method     = ConversionMethod.MARKER_OCR
            warnings.append(
                "Document appears to be fully scanned. OCR was applied — "
                "accuracy depends on scan quality."
            )
        elif is_mixed:
            force_ocr  = True
            is_scanned = True
            method     = ConversionMethod.MARKER_OCR
            warnings.append(
                f"Mixed PDF: pages {scanned_page_nums} appear scanned. "
                "OCR applied to the entire document for consistency."
            )
        else:
            force_ocr  = False
            method     = ConversionMethod.MARKER

        # ── 3. convert ───────────────────────────────────────────────────
        try:
            markdown = self._run_marker(path, force_ocr=force_ocr)

        except RuntimeError as exc:
            if not force_ocr:
                # Rare edge case: PDF claims a text layer but extraction
                # fails (e.g. custom fonts, badly embedded text).
                logger.warning(
                    "Marker (no OCR) failed on '%s': %s — retrying with OCR…",
                    path.name, exc,
                )
                try:
                    markdown   = self._run_marker(path, force_ocr=True)
                    method     = ConversionMethod.MARKER_OCR
                    is_scanned = True
                    warnings.append(
                        "Standard text extraction failed; OCR fallback was used."
                    )
                except RuntimeError as fallback_exc:
                    return ConversionResult(
                        success=False,
                        error=(
                            f"All extraction methods failed for '{path.name}'. "
                            f"Last error: {fallback_exc}"
                        ),
                    )
            else:
                return ConversionResult(success=False, error=str(exc))

        return ConversionResult(
            success     = True,
            markdown    = self._clean_markdown(markdown),
            method_used = method.value,
            file_type   = FileType.PDF.value,
            is_scanned  = is_scanned,
            page_count  = page_count,
            warnings    = warnings,
        )

    # ─────────────────────────────────────────────────────────────────────
    # Word pipeline
    # ─────────────────────────────────────────────────────────────────────

    def _convert_word(
        self, path: Path, file_type: FileType
    ) -> ConversionResult:
        """
        Convert DOCX (or legacy DOC) to GitHub-Flavoured Markdown via Pandoc.

        For ``.doc`` files LibreOffice is used first to produce a ``.docx``
        intermediate in ``self.temp_dir``, which Pandoc then converts.

        Pandoc flags used
        -----------------
        ``--wrap=none``              No hard line-wrapping.
        ``--strip-comments``         Drop Word tracked-change comments.
        ``--markdown-headings=atx``  Use ``#`` style, not underline style.
        """
        try:
            import pypandoc  # noqa: F401 – just check it is installed
        except ImportError:
            return ConversionResult(
                success=False,
                error=(
                    "pypandoc is not installed.\n"
                    "  pip install pypandoc\n"
                    "  python -c \"import pypandoc; pypandoc.download_pandoc()\""
                ),
            )

        warnings: list[str] = []
        method = ConversionMethod.PANDOC

        # ── .doc → .docx ─────────────────────────────────────────────────
        if file_type == FileType.DOC:
            try:
                path, warnings = self._doc_to_docx(path, warnings)
                method = ConversionMethod.PANDOC_VIA_LO
            except RuntimeError as exc:
                return ConversionResult(success=False, error=str(exc))

        # ── DOCX → GFM Markdown ──────────────────────────────────────────
        try:
            import pypandoc

            markdown = pypandoc.convert_file(
                str(path),
                to="gfm",
                extra_args=[
                    "--wrap=none",
                    "--strip-comments",
                    "--markdown-headings=atx",
                ],
            )
        except Exception as exc:
            return ConversionResult(
                success=False,
                error=f"Pandoc conversion failed: {exc}",
            )

        if not markdown or not markdown.strip():
            return ConversionResult(
                success=False,
                error=(
                    f"Pandoc returned empty output for '{path.name}'. "
                    "The document may be blank."
                ),
            )

        return ConversionResult(
            success     = True,
            markdown    = self._clean_markdown(markdown),
            method_used = method.value,
            file_type   = file_type.value,
            is_scanned  = False,
            warnings    = warnings,
        )

    def _doc_to_docx(
        self, path: Path, warnings: list[str]
    ) -> tuple[Path, list[str]]:
        """
        Convert a legacy ``.doc`` file to ``.docx`` via LibreOffice (headless).

        Returns
        -------
        (docx_path, updated_warnings)

        Raises
        ------
        RuntimeError
            If LibreOffice is missing, exits non-zero, times out, or
            produces no output file.
        """
        docx_path = self.temp_dir / (path.stem + ".docx")

        try:
            proc = subprocess.run(
                [
                    "libreoffice",
                    "--headless",
                    "--convert-to", "docx",
                    "--outdir",     str(self.temp_dir),
                    str(path),
                ],
                capture_output=True,
                text=True,
                timeout=self.LIBREOFFICE_TIMEOUT,
            )
        except FileNotFoundError:
            raise RuntimeError(
                "LibreOffice is required to convert legacy .doc files but "
                "was not found on PATH.\n"
                "  Ubuntu/Debian : sudo apt-get install libreoffice\n"
                "  macOS         : brew install --cask libreoffice"
            )
        except subprocess.TimeoutExpired:
            raise RuntimeError(
                f"LibreOffice timed out after {self.LIBREOFFICE_TIMEOUT} s "
                f"converting '{path.name}'."
            )

        if proc.returncode != 0:
            raise RuntimeError(
                f"LibreOffice exited with code {proc.returncode}:\n"
                f"{proc.stderr.strip()}"
            )

        if not docx_path.exists():
            raise RuntimeError(
                f"LibreOffice ran successfully but no .docx was produced. "
                f"Expected: {docx_path}"
            )

        warnings.append(
            "Legacy .doc file was automatically converted to .docx via "
            "LibreOffice before Markdown extraction."
        )
        return docx_path, warnings

    # ─────────────────────────────────────────────────────────────────────
    # Post-processing helpers
    # ─────────────────────────────────────────────────────────────────────

    @staticmethod
    def _clean_markdown(text: str) -> str:
        """
        Remove converter artefacts and normalise whitespace.

        Handles
        -------
        * Windows / mixed line endings (CRLF → LF)
        * Null bytes and Word private-use bullet characters
        * Trailing whitespace per line
        * Runs of 3+ blank lines → single blank line
        * Marker page-break separators (standalone ``---`` / ``===`` lines)
        * Pandoc footnote-like wrapping artefacts
        """
        # 1. Normalise line endings
        text = text.replace("\r\n", "\n").replace("\r", "\n")

        # 2. Remove null bytes and common Word private-use characters
        replacements = {
            "\x00":  "",   # null byte
            "\uf0b7": "-", # Word solid bullet  •
            "\uf0a7": "-", # Word hollow bullet  ◦
            "\uf020": " ", # Word private space
            "\uf0fc": "-", # Word checkmark bullet
        }
        for bad, good in replacements.items():
            text = text.replace(bad, good)

        # 3. Strip trailing whitespace on every line
        text = "\n".join(line.rstrip() for line in text.split("\n"))

        # 4. Remove Marker's standalone page-break / rule lines
        text = re.sub(r"(?m)^[-=]{3,}\s*$", "", text)

        # 5. Collapse 3+ consecutive blank lines to one blank line
        text = re.sub(r"\n{3,}", "\n\n", text)

        return text.strip()


# ─────────────────────────────────────────────────────────────────────────────
# CLI convenience  (python cv_converter.py <input> [output])
# ─────────────────────────────────────────────────────────────────────────────

def _cli() -> None:
    import sys

    logging.basicConfig(
        level=logging.INFO,
        format="%(levelname)s  %(message)s",
    )

    args = sys.argv[1:]
    if not args:
        print(
            "Usage: python cv_converter.py <input.(pdf|docx|doc)> [output.md]"
        )
        sys.exit(1)

    input_path  = Path(args[0])
    output_path = Path(args[1]) if len(args) > 1 else input_path.with_suffix(".md")

    converter = CVConverter()

    print(f"Converting '{input_path}' …")
    result = converter.convert(input_path)

    if result.warnings:
        for w in result.warnings:
            print(f"  ⚠  {w}")

    if not result:
        print(f"✗  Conversion failed: {result.error}")
        sys.exit(1)

    converter.save(result, output_path)
    print(
        f"✓  Done  [{result.method_used}]  "
        f"{'(scanned) ' if result.is_scanned else ''}"
        f"→ {output_path}"
    )
    if result.page_count:
        print(f"   Pages : {result.page_count}")
    print(f"   Size  : {len(result.markdown):,} characters")


if __name__ == "__main__":
    _cli()