convert_docs.py

"""Verify the curated policy Markdown (``Contextmd/``) against the source PDFs.

Design note
-----------
One of the source PDFs (``Placement Policy.pdf``) is a *scanned / image-only*
document with no text layer, so automated extractors (pypdf, pymupdf4llm) return
nothing usable for it. The project therefore indexes the hand-curated,
pre-verified Markdown in ``Contextmd/`` as its source of truth instead of a
machine conversion.

This module is the build-time **quality gate** for that curated Markdown. For
every PDF in ``Reference Documents/`` it:

  1. Confirms a matching ``Contextmd/<name>.md`` exists and is non-empty.
  2. If the PDF has a real text layer, independently re-extracts it with PyMuPDF
     and checks that the curated Markdown covers the bulk of the PDF's wording.
  3. Checks that a set of critical policy anchors are present.

A hard failure (missing/empty curated file, or coverage below the floor for a
text-layer PDF) exits non-zero so a bad corpus never reaches the index.
Originals in ``Reference Documents/`` are never modified.
"""

from __future__ import annotations

import logging
import re
import sys
from pathlib import Path

import fitz  # PyMuPDF (installed via pymupdf4llm)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-7s | convert_docs | %(message)s",
)
log = logging.getLogger("convert_docs")

# --- Structural paths (not secrets/config) -----------------------------------
REFERENCE_DIR = Path("Reference Documents")
CONTEXTMD_DIR = Path("Contextmd")

# For text-layer PDFs: fraction of PDF word-tokens that must appear in the
# curated Markdown. Boilerplate (form fields, signatures) legitimately drops a
# little, so this is a floor, not a target.
COVERAGE_WARN_THRESHOLD = 0.85
COVERAGE_FAIL_THRESHOLD = 0.50

# Anchors we expect the curated policy corpus to contain; missing ones almost
# always mean a truncated or wrong file.
CRITICAL_ANCHORS = [
    "placement",
    "eligibility",
    "attendance",
    "offer",
    "dream offer",
    "stdc",
    "car",
]

_TOKEN_RE = re.compile(r"[a-z0-9₹%×]+")
_SLUG_RE = re.compile(r"[^A-Za-z0-9]+")


def slugify(name: str) -> str:
    return _SLUG_RE.sub("-", name).strip("-")


def tokens(text: str) -> list[str]:
    return _TOKEN_RE.findall(text.lower())


def pdf_reference_text(pdf_path: Path) -> str:
    """Independent raw-text extraction (empty for scanned/image-only PDFs)."""
    parts = []
    with fitz.open(pdf_path) as doc:
        for page in doc:
            parts.append(page.get_text("text"))
    return "\n".join(parts)


def find_curated(pdf_stem: str) -> Path | None:
    """Locate the curated Markdown matching a PDF (case-insensitive slug match)."""
    if not CONTEXTMD_DIR.is_dir():
        return None
    slug = slugify(pdf_stem).lower()
    for md in sorted(CONTEXTMD_DIR.glob("*.md")):
        if md.stem.lower() == slug:
            return md
    return None


def verify(pdf_path: Path, md_path: Path) -> bool:
    """Verify one curated Markdown file against its source PDF.

    Returns True if it passed with no warnings. Raises RuntimeError on a hard
    failure (empty curated file or coverage below the hard floor).
    """
    markdown = md_path.read_text(encoding="utf-8")
    if not markdown.strip():
        raise RuntimeError(f"Curated file {md_path} is EMPTY")

    md_flat = re.sub(r"\s+", "", markdown.lower())
    md_vocab = set(tokens(markdown))

    ref_tokens = tokens(pdf_reference_text(pdf_path))
    if ref_tokens:
        present = sum(1 for w in ref_tokens if w in md_vocab)
        coverage = present / len(ref_tokens)
    else:
        # Scanned PDF: no independent text to cross-check; anchors only.
        coverage = None

    missing_anchors = [a for a in CRITICAL_ANCHORS if a.replace(" ", "") not in md_flat]

    log.info("Verifying %s  <-  %s", md_path.name, pdf_path.name)
    log.info("    curated chars  : %d", len(markdown))
    if coverage is None:
        log.info("    token coverage : n/a (scanned PDF — no text layer to compare)")
    else:
        log.info("    token coverage : %.1f%% (%d PDF tokens)", coverage * 100, len(ref_tokens))
    log.info(
        "    anchors        : %d/%d present%s",
        len(CRITICAL_ANCHORS) - len(missing_anchors),
        len(CRITICAL_ANCHORS),
        "" if not missing_anchors else f"  MISSING={missing_anchors}",
    )

    if coverage is not None and coverage < COVERAGE_FAIL_THRESHOLD:
        raise RuntimeError(
            f"{md_path.name} failed verification: token coverage {coverage:.1%} "
            f"below hard floor {COVERAGE_FAIL_THRESHOLD:.0%}"
        )

    ok = True
    if coverage is not None and coverage < COVERAGE_WARN_THRESHOLD:
        log.warning(
            "%s: token coverage %.1f%% below expected %.0f%% — review the file.",
            md_path.name,
            coverage * 100,
            COVERAGE_WARN_THRESHOLD * 100,
        )
        ok = False
    if missing_anchors:
        log.warning("%s: missing expected anchors %s", md_path.name, missing_anchors)
        ok = False
    return ok


def main() -> int:
    if not REFERENCE_DIR.is_dir():
        log.error("Source folder not found: %s", REFERENCE_DIR.resolve())
        return 1
    if not CONTEXTMD_DIR.is_dir():
        log.error("Curated Markdown folder not found: %s", CONTEXTMD_DIR.resolve())
        return 1

    pdfs = sorted(REFERENCE_DIR.glob("*.pdf"))
    if not pdfs:
        log.error("No PDF files found in %s", REFERENCE_DIR.resolve())
        return 1

    log.info(
        "Verifying %d curated document(s) in %s against source PDFs in %s",
        len(pdfs),
        CONTEXTMD_DIR,
        REFERENCE_DIR,
    )

    all_clean = True
    for pdf in pdfs:
        md_path = find_curated(pdf.stem)
        if md_path is None:
            log.error(
                "No curated Markdown in %s/ matches PDF '%s' (expected stem '%s').",
                CONTEXTMD_DIR,
                pdf.name,
                slugify(pdf.stem),
            )
            return 1
        all_clean = verify(pdf, md_path) and all_clean

    log.info(
        "Verification %s for %d document(s).",
        "clean" if all_clean else "completed WITH WARNINGS (see above)",
        len(pdfs),
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())