serving/app/services/parser.py

"""Deterministic document parsing.

This is not an LLM agent and must not become one. PDF/DOCX extraction is a
solved library problem with exact, reproducible character offsets; a model would
paraphrase rather than transcribe, destroying the `(page, start_char, end_char)`
anchoring that the evidence-citation layer depends on.

Failure is loud. A resume is parsed once and reused across every job, so a
guessed or partial parse would silently bias every future score for that
candidate — better to raise and let the document be re-uploaded or OCR'd.
"""

from __future__ import annotations

import io
import zipfile
from dataclasses import dataclass, field
from typing import Literal

import fitz
from docx import Document as DocxDocument
from docx.opc.exceptions import PackageNotFoundError

from app.config import DOCX_MIME
from app.exceptions import (
    DocumentParseError,
    EmptyDocumentError,
    UnsupportedMediaTypeError,
)
from app.logging import get_logger

logger = get_logger(__name__)

PARSER_VERSION = "pymupdf-1.24+python-docx-1.1/v1"
PAGE_SEPARATOR = "\n\n"
MIN_CHARS_PER_PAGE_FOR_TEXT_LAYER = 20

# Resource ceilings. A file can pass the byte-size check and still be a
# decompression bomb: a few hundred kilobytes of PDF can declare tens of
# thousands of pages, or expand into gigabytes of text. Both are cheap to
# construct and expensive to parse, so the parser bounds its own work rather
# than trusting that upstream size validation was sufficient.
MAX_PAGES = 500
MAX_EXTRACTED_CHARS = 5_000_000  # ~5 MB of text

ParseStatus = Literal["ok", "low_yield", "failed"]


@dataclass(frozen=True, slots=True)
class TextSpan:
    """One styled run of text with the geometry needed to judge visibility.

    The sanitizer cannot detect white-on-white or off-canvas text from a plain
    string, so these attributes are extracted in the same pass as the text.

    Attributes:
        text: The span's characters.
        page: One-based page number.
        size: Font size in points.
        color: Packed sRGB integer as reported by the PDF, or None if unknown.
        bbox: (x0, y0, x1, y1) in page coordinates, or None if unknown.
        page_rect: The page's own (x0, y0, x1, y1), for off-canvas comparison.
    """

    text: str
    page: int
    size: float
    color: int | None
    bbox: tuple[float, float, float, float] | None
    page_rect: tuple[float, float, float, float] | None


@dataclass(frozen=True, slots=True)
class ParsedPage:
    """One page of extracted text, anchored into the whole-document string.

    Attributes:
        page: One-based page number.
        text: The page's extracted text.
        start_char: Inclusive offset of `text` within `ParsedDocument.text`.
        end_char: Exclusive offset of `text` within `ParsedDocument.text`.
    """

    page: int
    text: str
    start_char: int
    end_char: int


@dataclass(frozen=True, slots=True)
class ParsedDocument:
    """The result of parsing one resume or job description.

    Attributes:
        text: Full extracted text, the string all offsets index into.
        pages: Per-page slices with exact offsets.
        page_count: Number of pages found.
        media_type: The media type actually parsed.
        spans: Styled runs with geometry, consumed by the sanitizer.
        metadata: Document metadata, an injection surface in its own right.
        parser_version: Provenance for replay.
        needs_ocr: True when the text layer is too thin to trust.
        parse_status: "ok" when text was recovered, "low_yield" when OCR is needed.
        warnings: Non-fatal notes raised during parsing.
    """

    text: str
    pages: tuple[ParsedPage, ...]
    page_count: int
    media_type: str
    parser_version: str = PARSER_VERSION
    needs_ocr: bool = False
    parse_status: ParseStatus = "ok"
    warnings: tuple[str, ...] = field(default=())
    spans: tuple[TextSpan, ...] = field(default=())
    metadata: dict[str, str] = field(default_factory=dict)


def _assemble(
    pages: list[str],
    media_type: str,
    spans: tuple[TextSpan, ...] = (),
    metadata: dict[str, str] | None = None,
) -> ParsedDocument:
    """Join page texts and compute exact offsets for each page.

    Offsets are derived from the same concatenation that produces `text`, so
    `text[start_char:end_char] == page.text` holds by construction rather than
    by arithmetic that could drift.

    Args:
        pages: Per-page extracted text, in order.
        media_type: The media type that was parsed.
        spans: Styled runs with geometry, when the format exposes them.
        metadata: Document metadata, when the format exposes it.

    Returns:
        The assembled document.
    """
    parts: list[ParsedPage] = []
    cursor = 0
    for index, page_text in enumerate(pages, start=1):
        start = cursor
        end = start + len(page_text)
        parts.append(ParsedPage(page=index, text=page_text, start_char=start, end_char=end))
        cursor = end + len(PAGE_SEPARATOR)

    full_text = PAGE_SEPARATOR.join(pages)
    total_chars = sum(len(p) for p in pages)
    thin = total_chars < MIN_CHARS_PER_PAGE_FOR_TEXT_LAYER * max(len(pages), 1)

    if thin:
        logger.info(
            "parse_low_yield",
            page_count=len(pages),
            total_chars=total_chars,
            needs_ocr=True,
        )

    return ParsedDocument(
        text=full_text,
        pages=tuple(parts),
        page_count=len(pages),
        media_type=media_type,
        needs_ocr=thin,
        parse_status="low_yield" if thin else "ok",
        spans=spans,
        metadata=metadata or {},
    )


def _extract_spans(page: fitz.Page, page_number: int) -> list[TextSpan]:
    """Collect styled runs and their geometry from one PDF page.

    Uses the structured `dict` extraction rather than plain text so the
    sanitizer can see font size, colour, and position — the attributes that
    distinguish text a human can read from text hidden for a model to find.

    Args:
        page: The PyMuPDF page.
        page_number: One-based page number.

    Returns:
        Every non-empty span on the page.
    """
    rect = page.rect
    page_rect = (rect.x0, rect.y0, rect.x1, rect.y1)
    collected: list[TextSpan] = []

    payload = page.get_text("dict")
    for block in payload.get("blocks", ()):
        for line in block.get("lines", ()):
            for span in line.get("spans", ()):
                text = str(span.get("text", ""))
                if not text.strip():
                    continue
                raw_bbox = span.get("bbox")
                bbox: tuple[float, float, float, float] | None = None
                if raw_bbox is not None and len(raw_bbox) == 4:
                    bbox = (
                        float(raw_bbox[0]),
                        float(raw_bbox[1]),
                        float(raw_bbox[2]),
                        float(raw_bbox[3]),
                    )
                collected.append(
                    TextSpan(
                        text=text,
                        page=page_number,
                        size=float(span.get("size", 0.0)),
                        color=span.get("color"),
                        bbox=bbox,
                        page_rect=page_rect,
                    )
                )
    return collected


def _parse_pdf(content: bytes) -> ParsedDocument:
    """Extract text from a PDF using PyMuPDF.

    Args:
        content: Raw PDF bytes.

    Returns:
        The parsed document.

    Raises:
        DocumentParseError: If the PDF cannot be opened or read.
    """
    try:
        with fitz.open(stream=content, filetype="pdf") as document:
            if document.page_count > MAX_PAGES:
                logger.warning(
                    "parse_pdf_rejected_page_count",
                    page_count=document.page_count,
                    limit=MAX_PAGES,
                )
                raise DocumentParseError(
                    f"The PDF exceeds the {MAX_PAGES}-page limit."
                )

            pages: list[str] = []
            spans: list[TextSpan] = []
            total_chars = 0
            for index, page in enumerate(document, start=1):
                text = page.get_text("text")
                total_chars += len(text)
                if total_chars > MAX_EXTRACTED_CHARS:
                    logger.warning(
                        "parse_pdf_rejected_text_volume", limit=MAX_EXTRACTED_CHARS
                    )
                    raise DocumentParseError(
                        "The PDF expands to more text than this service will process."
                    )
                pages.append(text)
                spans.extend(_extract_spans(page, index))
            metadata = {
                str(k): str(v) for k, v in (document.metadata or {}).items() if v
            }
    except DocumentParseError:
        raise
    except Exception as err:  # noqa: BLE001 - normalized to a domain error
        logger.warning("parse_pdf_failed", error=str(err), error_type=type(err).__name__)
        raise DocumentParseError("The PDF could not be opened.") from err

    if not pages:
        raise DocumentParseError("The PDF contains no pages.")
    return _assemble(pages, "application/pdf", tuple(spans), metadata)


def _parse_docx(content: bytes) -> ParsedDocument:
    """Extract text from a DOCX using python-docx.

    DOCX has no intrinsic pagination — pagination is a rendering property, not a
    storage one — so the whole document is reported as a single logical page
    rather than inventing page breaks that would make offsets meaningless.

    Args:
        content: Raw DOCX bytes.

    Returns:
        The parsed document.

    Raises:
        DocumentParseError: If the archive is not a readable Word document.
    """
    try:
        document = DocxDocument(io.BytesIO(content))
        paragraphs = [p.text for p in document.paragraphs]
    except (
        PackageNotFoundError,
        zipfile.BadZipFile,
        ValueError,
        KeyError,
        OSError,
    ) as err:
        logger.warning("parse_docx_failed", error=str(err), error_type=type(err).__name__)
        raise DocumentParseError("The DOCX could not be opened.") from err

    body = "\n".join(paragraphs)
    if len(body) > MAX_EXTRACTED_CHARS:
        logger.warning("parse_docx_rejected_text_volume", limit=MAX_EXTRACTED_CHARS)
        raise DocumentParseError(
            "The DOCX expands to more text than this service will process."
        )
    return _assemble([body], DOCX_MIME)


def parse_document(content: bytes, media_type: str) -> ParsedDocument:
    """Parse a resume or job description into text with exact offsets.

    Args:
        content: Raw file bytes.
        media_type: The media type to parse as. Callers should pass the result
            of `app.utils.parsing.detect_media_type`, not a client-supplied
            header.

    Returns:
        The parsed document, with `needs_ocr` set when the text layer is thin.

    Raises:
        EmptyDocumentError: If `content` is empty.
        UnsupportedMediaTypeError: If `media_type` is not PDF or DOCX.
        DocumentParseError: If the document is structurally unreadable.
    """
    if not content:
        raise EmptyDocumentError()
    if media_type == "application/pdf":
        return _parse_pdf(content)
    if media_type == DOCX_MIME:
        return _parse_docx(content)
    raise UnsupportedMediaTypeError()