src/knowledge_parsing/report.py

"""Summarise every run into figures ready for a resourcing argument.

    py -m src.knowledge_parsing.report
    py -m src.knowledge_parsing.report --scale 6800

`--scale` extrapolates to a given page count (e.g. BUMA scale, ~6,700-6,800
pages) using the MEASURED seconds/page — documents served from cache are left
out, because including them would make the figure look far faster than reality.
"""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from .config import PipelineConfig

# Below this, the fixed cost of loading models dominates, so seconds/page comes
# out far higher than it would on a large document. Measured on this machine:
# 1 page -> 54.0 s/page, 5 pages -> 33.2, 20 pages -> 19.0.
MIN_PAGES_TO_EXTRAPOLATE = 20


def _duration(seconds: float) -> str:
    if seconds < 90:
        return f"{seconds:.0f} s"
    if seconds < 5400:
        return f"{seconds/60:.1f} min"
    return f"{seconds/3600:.1f} h"


def collect(runs_dir: Path) -> list[dict]:
    results = []
    for m in sorted(runs_dir.glob("*/manifest.json")):
        try:
            results.append(json.loads(m.read_text(encoding="utf-8")))
        except (OSError, json.JSONDecodeError) as e:
            # Manifest corrupt or half-written (a run that died mid-way). Skipped,
            # but reported — a report that silently loses a run gives a wrong
            # seconds/page figure with nothing to show for it.
            print(f"  ! skipping unreadable manifest: {m} ({type(e).__name__})")
    return results


def render(runs: list[dict], scale: int | None) -> None:
    if not runs:
        print("No runs yet. Try: py -m src.knowledge_parsing.run --input data/knowledge_docs/")
        return

    print(f"{'run':<17} {'backend':<9} {'ver':<8} {'docs':>4} {'pages':>5} "
          f"{'s/page':>8} {'warnings':>9}")
    print("-" * 66)

    for r in runs:
        s = r.get("summary", {})
        cfg = r.get("config", {})
        # the backend recorded in MinerU's output is trusted over the config
        recorded = {d.get("backend_recorded") for d in r.get("documents", [])} - {None}
        backend = "/".join(sorted(recorded)) if recorded else cfg.get("backend", "?")
        spp = s.get("seconds_per_page")
        print(f"{r.get('run_id',''):<17} {backend:<9} "
              f"{str(r.get('mineru_version') or '?')[:7]:<8} "
              f"{s.get('documents_ok',0):>4} {s.get('total_pages',0):>5} "
              f"{(f'{spp:.2f}' if spp else '-'):>8} {s.get('total_warnings',0):>9}")

    measured = [r for r in runs if (r.get("summary") or {}).get("seconds_per_page")]
    if not measured:
        print("\n(no run has actually parsed anything yet — all cache hits)")
        return

    print("\n--- extrapolation ---")
    target = scale or 6800
    doubtful = False
    for r in measured:
        s = r["summary"]
        spp = s["seconds_per_page"]
        pages = s.get("pages_measured", 0)
        cfg = r.get("config", {})
        note = ""
        if pages < MIN_PAGES_TO_EXTRAPOLATE:
            note = f"  ⚠ only {pages} page" + ("" if pages == 1 else "s")
            doubtful = True
        print(f"  {cfg.get('backend','?'):<9} {spp:>6.2f} s/page  →  "
              f"{target} pages = {_duration(spp * target)}{note}")

    if doubtful:
        print(
            f"\n  ⚠ A run under {MIN_PAGES_TO_EXTRAPOLATE} pages is NOT fit to "
            "extrapolate from.\n"
            "    Model loading is a fixed cost, so a small document looks far\n"
            "    slower per page. Measured earlier on this machine:\n"
            "      1 page -> 54.0 s/page | 5 pages -> 33.2 | 20 pages -> 19.0\n"
            "    Use a document of >= 20 pages before the number leaves the room."
        )

    if len(measured) >= 2:
        fastest = min(measured, key=lambda r: r["summary"]["seconds_per_page"])
        slowest = max(measured, key=lambda r: r["summary"]["seconds_per_page"])
        a = slowest["summary"]["seconds_per_page"]
        b = fastest["summary"]["seconds_per_page"]
        if b:
            print(f"\n  fastest vs slowest: {a/b:.1f}×")


def main() -> None:
    p = argparse.ArgumentParser(description="Summary figures across every run")
    p.add_argument("--runs-dir", type=Path, default=None)
    p.add_argument("--scale", type=int, default=None,
                   help="extrapolate to this many pages (e.g. 6800)")
    a = p.parse_args()
    cfg = PipelineConfig()
    render(collect(a.runs_dir or cfg.runs_dir), a.scale)


if __name__ == "__main__":
    main()