app/ocr.py

"""Surya OCR intake (PRD FR-1).

Accepts .png/.jpg/.pdf and extracts candidate demographic fields. Any field
whose OCR confidence < threshold is blanked and flagged `needs_review=True` so
the frontend forces manual validation.

Surya's dependencies (transformers 5.x + llama.cpp) conflict with the embedding
stack, so the actual OCR runs in an isolated interpreter (backend/.venv-ocr)
via `ocr_worker.py`, invoked here as a subprocess.

PHIPA: the upload bytes are written to a temp file only for the duration of the
call and deleted immediately afterwards (ephemeral retention).
"""
from __future__ import annotations

import json
import logging
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path

from .config import settings
from .schemas import ExtractedField, IntakeResponse

logger = logging.getLogger(__name__)

_WORKER = Path(__file__).with_name("ocr_worker.py")
_ISOLATED_PY = Path(__file__).resolve().parents[1] / ".venv-ocr" / "bin" / "python"

# Lightweight regex heuristics to pull structured demographics from OCR text.
FIELD_PATTERNS: dict[str, re.Pattern] = {
    # Ontario health card: 10 digits + optional 2-letter version code
    "health_card_number": re.compile(r"\b(\d{4}[-\s]?\d{3}[-\s]?\d{3}[-\s]?[A-Z]{0,2})\b"),
    "date_of_birth": re.compile(
        r"\b(\d{4}[-/]\d{2}[-/]\d{2}|\d{2}[-/]\d{2}[-/]\d{4})\b"
    ),
}


def _ocr_interpreter() -> str:
    if settings.ocr_python:
        return settings.ocr_python
    if _ISOLATED_PY.exists():
        return str(_ISOLATED_PY)
    return sys.executable  # fall back to current venv (needs surya installed)


def _run_worker(content: bytes, filename: str) -> tuple[str, list[tuple[str, float]]]:
    suffix = os.path.splitext(filename)[1] or ".png"
    in_fd, in_path = tempfile.mkstemp(suffix=suffix)
    out_path = in_path + ".json"
    try:
        with os.fdopen(in_fd, "wb") as f:
            f.write(content)
        subprocess.run(
            [_ocr_interpreter(), str(_WORKER), in_path, out_path],
            check=True,
            capture_output=True,
            timeout=300,
        )
        data = json.loads(Path(out_path).read_text())
        lines = [(l["text"], float(l["confidence"])) for l in data["lines"]]
        return data["raw_text"], lines
    except subprocess.CalledProcessError as exc:  # noqa: BLE001
        logger.error("OCR worker failed: %s", exc.stderr.decode(errors="ignore")[-500:])
        raise RuntimeError("OCR processing failed") from exc
    finally:
        for p in (in_path, out_path):
            try:
                os.unlink(p)
            except OSError:
                pass


def run_ocr(content: bytes, filename: str) -> IntakeResponse:
    raw_text, lines = _run_worker(content, filename)
    fields = _extract_fields(raw_text, lines)
    return IntakeResponse(fields=fields, raw_text=raw_text)


def _extract_fields(
    raw_text: str, lines: list[tuple[str, float]]
) -> list[ExtractedField]:
    """Match demographic patterns and attach the confidence of the source line."""
    threshold = settings.ocr_confidence_threshold
    out: list[ExtractedField] = []

    for field_name, pattern in FIELD_PATTERNS.items():
        match = None
        line_conf = 0.0
        for text, conf in lines:
            m = pattern.search(text)
            if m:
                match = m.group(1)
                line_conf = conf
                break

        needs_review = (match is None) or (line_conf < threshold)
        out.append(
            ExtractedField(
                field_name=field_name,
                # FR-1 fallback: blank the box when confidence is too low
                value=None if needs_review else match,
                confidence=round(line_conf, 3),
                needs_review=needs_review,
            )
        )
    return out