src/endopath/reocr.py

"""Re-OCR the rendered page images to a cleaner ground-truth report text (#212).

The prepared TCGA corpus OCR (`data/raw/TCGA_Reports.csv`) is pervasively damaged: 90% of reports hold
some damage and half are heavily damaged, with tables shredded into period-separated fragments
(`benchmark.ocr_damage_density`). Extraction reads that text, so a damaged transcription produces damaged
drafts and evidence spans that will not match the page.

This module transcribes the page images directly with a vision model, preserving layout and tables, to
produce a report text that both the extraction pass and the evidence check can rely on. The model is
reached through the app's provider seam (`endopath.llm.get_client`), so the same call runs against the
hosted Anthropic API or a local, on-premise model with no change here.

`ClaudePageReader` implements the `textsource.PageReader` protocol, so it plugs into the existing
`VisionLlmTextSource`: registering it with `textsource.set_default_page_reader(ClaudePageReader())` makes
the ingest path read pages through it. `scripts/reocr_pages.py` uses it to re-OCR a selected set of cases
and persist the result as `reports.report_text` (`storage.set_report_text`).

Cost guard: every page transcription is cached by the page image's sha256 checksum
(`PageTranscriptCache`), so re-running the script never re-bills a page whose pixels are unchanged, and a
`--limit`/`--worst`/`--case` selection in the script bounds how many pages are sent. No page is skipped
silently: the reader reports per-run token counts and cache hits.
"""

from __future__ import annotations

import base64
import dataclasses
import hashlib
import json
import os
from pathlib import Path
from typing import Optional

# Consistent with every other LLM call site in the app (llm_extraction, induction, resolution, the chat
# co-pilots): the codebase standardizes on this model, and the vision extract in resolution.py already
# reads page images with it. Override with REOCR_MODEL to trade cost for capability on a bulk run;
# claude-opus-4-8 is the higher-capability, higher-cost alternative reachable through the same seam.
REOCR_MODEL = (os.environ.get("REOCR_MODEL") or "").strip() or "claude-sonnet-5"

# A single page of a pathology report transcribes to at most a few thousand tokens; this ceiling leaves
# ample headroom and stays well under the streaming threshold, so a plain non-streaming call is safe.
DEFAULT_MAX_TOKENS = 8000

DEFAULT_CACHE_PATH = Path("data") / "interim" / "reocr_cache.json"

_IMAGE_MEDIA_TYPES = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg"}

# The pages of one report are joined with a blank line, matching the corpus's own paragraph spacing, so
# the concatenated text reads as one continuous document. No synthetic page markers are injected: a
# marker is text the model could later quote, and invariant 2 verifies an evidence quote as a verbatim
# substring of exactly this stored text.
PAGE_SEPARATOR = "\n\n"

TRANSCRIPTION_PROMPT = """\
You are transcribing a scanned pathology report page into plain text for a clinical dataset. Reproduce \
the page exactly as printed.

Rules:
- Transcribe every word, number, and punctuation mark verbatim. Do not paraphrase, summarize, correct, \
translate, re-order, or add commentary of your own.
- Preserve the reading order and the line breaks of the page.
- When the page contains a table, render it as a Markdown table: one row per printed row, columns in \
their printed order, so the rows stay readable instead of collapsing into a list of fragments.
- Keep a field label and its value on the same line, as printed ("Label: value").
- For a checkbox or a selectable option, write [x] for a filled or checked mark and [ ] for an empty one.
- If a word is genuinely illegible, write [illegible] in its place. Do not guess at it.
- Output only the transcription. No preamble, no explanation, no code fences.
"""


def page_checksum(path: Path | str) -> str:
    """sha256 of a page image's bytes: the cache key, so an unchanged page is never re-transcribed and a
    re-rendered page (different pixels) is."""
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


class PageTranscriptCache:
    """A JSON-backed page-transcript cache keyed by `checksum` and validated against the model that
    produced it, so switching REOCR_MODEL re-transcribes rather than serving another model's text. The
    file is written atomically. `dirty` lets a caller batch a single save after a run instead of writing
    per page."""

    def __init__(self, path: Path | str = DEFAULT_CACHE_PATH) -> None:
        self.path = Path(path)
        self.dirty = False
        self._data: dict[str, dict] = {}
        if self.path.exists():
            try:
                self._data = json.loads(self.path.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, OSError):
                self._data = {}

    def get(self, checksum: str, model: str) -> Optional[str]:
        entry = self._data.get(checksum)
        if entry is None or entry.get("model") != model:
            return None
        return entry.get("text")

    def put(self, checksum: str, model: str, text: str) -> None:
        self._data[checksum] = {"model": model, "text": text, "chars": len(text)}
        self.dirty = True

    def save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(self.path.suffix + ".tmp")
        tmp.write_text(json.dumps(self._data, ensure_ascii=False, indent=0), encoding="utf-8")
        tmp.rename(self.path)
        self.dirty = False

    def __len__(self) -> int:
        return len(self._data)


@dataclasses.dataclass
class PageTranscript:
    """One page's transcription and the tokens it cost. A cache hit reports zero tokens; the caller sums
    these across a run for a real cost figure rather than an estimate."""

    text: str
    checksum: str
    cached: bool
    input_tokens: int = 0
    output_tokens: int = 0


def _image_block(path: Path) -> dict:
    media_type = _IMAGE_MEDIA_TYPES.get(path.suffix.lower())
    if media_type is None:
        raise ValueError(f"unsupported image type for re-OCR: {path}")
    return {
        "type": "image",
        "source": {
            "type": "base64",
            "media_type": media_type,
            "data": base64.standard_b64encode(path.read_bytes()).decode("ascii"),
        },
    }


def _strip_code_fences(text: str) -> str:
    """Drop a wrapping ``` fence if the model added one despite the instruction, so the stored text is
    the transcription and not a fenced block. A fence that is not on the first/last line is left alone
    (it could be part of the page content)."""
    lines = text.strip().splitlines()
    if len(lines) >= 2 and lines[0].startswith("```") and lines[-1].strip() == "```":
        return "\n".join(lines[1:-1]).strip()
    return text.strip()


def transcribe_page(
    path: Path | str,
    *,
    client=None,
    model: str = REOCR_MODEL,
    cache: Optional[PageTranscriptCache] = None,
    max_tokens: int = DEFAULT_MAX_TOKENS,
) -> PageTranscript:
    """Transcribe one page image, returning its text and token cost. Uses `cache` when the page's
    checksum + model already produced text, so a re-run pays nothing for an unchanged page. The model is
    reached through the provider seam (`endopath.llm.get_client`), so a local model swaps in with no
    change here."""
    path = Path(path)
    checksum = page_checksum(path)
    if cache is not None:
        hit = cache.get(checksum, model)
        if hit is not None:
            return PageTranscript(text=hit, checksum=checksum, cached=True)

    if client is None:
        from endopath import llm

        # The transcription stage's own client: a backend serving one model for both stages returns the
        # same client extraction uses, and the GPU backend returns its vision model (#83).
        client = llm.get_reocr_client()

    response = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        messages=[
            {
                "role": "user",
                "content": [_image_block(path), {"type": "text", "text": TRANSCRIPTION_PROMPT}],
            }
        ],
    )
    if getattr(response, "stop_reason", None) == "max_tokens":
        raise RuntimeError(
            f"re-OCR of {path} hit max_tokens before the page finished; raise max_tokens (currently "
            f"{max_tokens})."
        )
    text = "".join(block.text for block in response.content if getattr(block, "type", None) == "text")
    text = _strip_code_fences(text)

    usage = getattr(response, "usage", None)
    input_tokens = getattr(usage, "input_tokens", None) or 0
    output_tokens = getattr(usage, "output_tokens", None) or 0

    if cache is not None:
        cache.put(checksum, model, text)

    return PageTranscript(
        text=text,
        checksum=checksum,
        cached=False,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
    )


@dataclasses.dataclass
class RunStats:
    pages: int = 0
    cached_pages: int = 0
    input_tokens: int = 0
    output_tokens: int = 0

    @property
    def billed_pages(self) -> int:
        return self.pages - self.cached_pages


class ClaudePageReader:
    """A `textsource.PageReader` that transcribes a report's rendered pages with a vision model and joins
    them into one report text. Registering it with `textsource.set_default_page_reader(...)` makes the
    ingest path read pages through it; `scripts/reocr_pages.py` uses it directly. Token cost and cache
    hits from the most recent `read` are on `last_run` for a caller that reports cost."""

    def __init__(
        self,
        *,
        client=None,
        model: str = REOCR_MODEL,
        cache: Optional[PageTranscriptCache] = None,
        max_tokens: int = DEFAULT_MAX_TOKENS,
        page_separator: str = PAGE_SEPARATOR,
    ) -> None:
        self._client = client
        self._model = model
        self._cache = cache
        self._max_tokens = max_tokens
        self._page_separator = page_separator
        self.last_run = RunStats()

    def read(self, page_images: list[Path]) -> str:
        self.last_run = RunStats()
        parts: list[str] = []
        for path in page_images:
            transcript = transcribe_page(
                path,
                client=self._client,
                model=self._model,
                cache=self._cache,
                max_tokens=self._max_tokens,
            )
            self.last_run.pages += 1
            if transcript.cached:
                self.last_run.cached_pages += 1
            self.last_run.input_tokens += transcript.input_tokens
            self.last_run.output_tokens += transcript.output_tokens
            if transcript.text:
                parts.append(transcript.text)
        if self._cache is not None and self._cache.dirty:
            self._cache.save()
        return self._page_separator.join(parts)