manga-recap-pipeline/audit_registry.py

"""Registry audit for one chapter, after vision + identity + reconcile and before anything downstream.

Runs inside manga-orchestrator (reads /data/manga.db). `audit_speakers.py` answers the attribution
questions and needs the dialogue stage; this one answers the questions that decide whether dialogue is
worth running at all:

  1. did the bbox fix land — are stored boxes pixels, or still gemma's 0-1000 grid,
  2. how many characters did the rebaseline mint, and did one of them absorb the chapter again,
  3. purity and fragmentation against `eval/chapter-truth.json`, the measurement spine,
  4. what happened on panel 7, the worked example.

Usage: docker exec manga-orchestrator python3 /app/audit_registry.py [chapter_id] [panel_index] [truth]
       python3 audit_registry.py --selftest      # scoring only, no db

Both files have to be inside the container, and `docker compose up --build` drops them:
  docker cp audit_registry.py manga-orchestrator:/app/
  docker cp eval/chapter-truth.json manga-orchestrator:/app/eval/
"""
import collections
import json
import os
import sqlite3
import sys

IOU_MIN = 0.5   # ponytail: a fixed floor. Boxes move a little between runs, people do not.


def _iou(a, b):
    ix = max(0, min(a[2], b[2]) - max(a[0], b[0]))
    iy = max(0, min(a[3], b[3]) - max(a[1], b[1]))
    inter = ix * iy
    if not inter:
        return 0.0
    ua = (a[2] - a[0]) * (a[3] - a[1]) + (b[2] - b[0]) * (b[3] - b[1]) - inter
    return inter / ua if ua > 0 else 0.0


def _score(truth, found):
    """Match truth rows to this run's assignments by page-space IoU, then cluster them.

    truth: [{page, box, person}], found: [{page, box, character_id}]. Greedy on best IoU, each row
    and each assignment used once. Returns (per_cluster, fragmentation, unmatched_truth).

    Purity is the largest share of one true person inside a cluster, so it needs no character_id
    from the truth file. It has to stay that way: /characters/reset remints every id.
    """
    pairs = sorted(
        ((_iou(t["box"], f["box"]), ti, fi)
         for ti, t in enumerate(truth) for fi, f in enumerate(found)
         if t["page"] == f["page"] and _iou(t["box"], f["box"]) >= IOU_MIN),
        key=lambda p: -p[0])
    used_t, used_f, per_cluster, holders = set(), set(), {}, {}
    for _, ti, fi in pairs:
        if ti in used_t or fi in used_f:
            continue
        used_t.add(ti)
        used_f.add(fi)
        person, cid = truth[ti]["person"], found[fi]["character_id"]
        per_cluster.setdefault(cid, collections.Counter())[person] += 1
        holders.setdefault(person, set()).add(cid)
    frag = {p: len(ids) for p, ids in holders.items()}
    return per_cluster, frag, [t for i, t in enumerate(truth) if i not in used_t]


def _selftest():
    truth = [
        {"page": 0, "box": [0, 0, 100, 100], "person": "a"},
        {"page": 0, "box": [200, 0, 300, 100], "person": "a"},
        {"page": 0, "box": [400, 0, 500, 100], "person": "b"},
        {"page": 0, "box": [600, 0, 700, 100], "person": "art"},
        {"page": 1, "box": [0, 0, 100, 100], "person": "a"},
    ]
    found = [
        {"page": 0, "box": [4, 4, 104, 104], "character_id": "c1"},     # shifted by a re-crop
        {"page": 0, "box": [200, 0, 300, 100], "character_id": "c2"},   # a again, on a second id
        {"page": 0, "box": [400, 0, 500, 100], "character_id": "c1"},   # b, folded into c1
        {"page": 0, "box": [600, 0, 700, 100], "character_id": "c1"},   # the art, also c1
        {"page": 1, "box": [0, 0, 100, 100], "character_id": "c9"},     # same box, other page
    ]
    per_cluster, frag, unmatched = _score(truth, found)
    assert dict(per_cluster["c1"]) == {"a": 1, "b": 1, "art": 1}, per_cluster
    assert max(per_cluster["c1"].values()) / sum(per_cluster["c1"].values()) == 1 / 3
    assert frag == {"a": 3, "b": 1, "art": 1}, frag   # a is split over c1, c2 and c9
    assert not unmatched, unmatched
    # two candidates for one row: the tighter box wins and the looser one is left out
    per_cluster, _, unmatched = _score(
        truth[:1], found[:1] + [{"page": 0, "box": [0, 0, 100, 100], "character_id": "c8"}])
    assert list(per_cluster) == ["c8"], per_cluster
    assert not unmatched, unmatched
    # a row nothing overlaps stays unmatched rather than snapping to the nearest box
    _, _, unmatched = _score(truth, [{"page": 0, "box": [0, 0, 20, 20], "character_id": "c1"}])
    assert len(unmatched) == 5, unmatched
    print("selftest ok")


if "--selftest" in sys.argv:
    _selftest()
    raise SystemExit

CHAPTER = sys.argv[1] if len(sys.argv) > 1 else "7c944dd4-e972-42c7-ba60-9f6939548e80"
# panel_index, the db column. NEXT.md's "panel 7" is panel_order 7, which is this chapter's
# panel_index 6, the wide office shot. panel_index 7 is a balloon-only panel vision skips.
WORKED_EXAMPLE = int(sys.argv[2]) if len(sys.argv) > 2 else 6
TRUTH = sys.argv[3] if len(sys.argv) > 3 else "/app/eval/chapter-truth.json"
if not os.path.exists(TRUTH):
    TRUTH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "eval/chapter-truth.json")

c = sqlite3.connect("/data/manga.db")
c.row_factory = sqlite3.Row
manga_id = c.execute("SELECT manga_id FROM chapters WHERE chapter_id=?", (CHAPTER,)).fetchone()[0]

reg = {r["character_id"]: dict(r) for r in c.execute(
    "SELECT character_id, name, aliases, gender, ref_image_uris, embedding_uri, merged_into "
    "FROM characters WHERE manga_id=?", (manga_id,))}
# a merge keeps the losing row and sets merged_into, so a raw COUNT overstates the cast.
live = {k: r for k, r in reg.items() if not r["merged_into"]}


def _label(cid):
    """Two characters can carry the same name — the naming defect mints duplicates — so a label
    that is only the name silently sums two clusters into one line."""
    name = (reg.get(cid, {}).get("name") or "").strip()
    return f"{name} [{cid[10:18]}]" if name else cid[:20]

panels = c.execute(
    'SELECT panel_id, panel_index, page_index, bbox FROM panels WHERE chapter_id=? ORDER BY panel_order',
    (CHAPTER,)).fetchall()

# 1. coordinate space. A 0-1000 grid box on a panel wider or taller than 1000px cannot exceed 1000,
# and clamps AT 1000. Real pixel boxes track the panel and scatter past it. The tell is the ratio of
# the largest coordinate to the panel dimension, plus how many boxes sit exactly on 1000.
detections = 0
past_1000 = at_1000 = 0
max_ratio = 0.0
assigned_total = 0
per_char = collections.Counter()
worked = None
found = []   # every assignment as a page-space box, for the truth match below
skipped = lines = lines_skipped = 0

for p in panels:
    row = c.execute("SELECT result_json FROM vision_results WHERE panel_id=?", (p["panel_id"],)).fetchone()
    if not row:
        continue
    v = json.loads(row["result_json"])
    # the vision blob carries no panel size. panels.bbox is the panel's box on its page and is
    # [x, y, w, h], not corners — panel 3 of this chapter is [0, 615, 900, 106].
    pb = json.loads(p["bbox"] or "null")
    pw, ph = (pb[2], pb[3]) if pb and len(pb) == 4 else (None, None)
    ox, oy = (pb[0], pb[1]) if pb and len(pb) == 4 else (0, 0)
    assigns = {a["local_id"]: (a["character_id"], a["confidence"]) for a in c.execute(
        "SELECT local_id, character_id, confidence FROM identity_assignments WHERE panel_id=?",
        (p["panel_id"],))}
    assigned_total += len(assigns)
    for cid, _ in assigns.values():
        per_char[_label(cid)] += 1
    people = [ch for ch in (v.get("characters") or []) if ch.get("bbox")]
    detections += len(people)
    n_lines = len(v.get("dialogue") or [])
    lines += n_lines
    if v.get("skip"):
        skipped += 1
        lines_skipped += n_lines
    for ch in people:
        x1, y1, x2, y2 = ch["bbox"]
        past_1000 += 1 if max(x2, y2) > 1000 else 0
        at_1000 += 1 if 1000 in (x2, y2) else 0
        if pw and ph:
            max_ratio = max(max_ratio, x2 / pw, y2 / ph)
        cid = assigns.get(ch["local_id"], (None,))[0]
        if cid:
            found.append({"page": p["page_index"], "character_id": cid,
                          "box": [x1 + ox, y1 + oy, x2 + ox, y2 + oy]})
    if p["panel_index"] == WORKED_EXAMPLE:
        worked = (p, v, people, assigns, pw, ph)

named = [r for r in live.values() if (r["name"] or "").strip()]
print(f"registry: {len(live)} live characters ({len(reg) - len(live)} merged away), "
      f"{len(named)} named -> {sorted((r['name'] or '') for r in named)}")
print(f"detections: {detections}  assignments: {assigned_total} "
      f"= {100*assigned_total/max(detections,1):.0f}% coverage")
# a skipped panel is one vision judged to hold no character. It keeps its dialogue, so those lines
# have no visible speaker to attribute to and are the case for an `offscreen` speaker_ref kind.
print(f"vision skipped {skipped}/{len(panels)} panels, holding "
      f"{lines_skipped}/{lines} dialogue lines")
if per_char:
    top, n = per_char.most_common(1)[0]
    print(f"assignment spread: {dict(per_char.most_common(8))}")
    print(f"  top character holds {n}/{assigned_total} = {100*n/max(assigned_total,1):.0f}% "
          f"({'ABSORBING, same signature as before' if n > 0.5 * assigned_total else 'ok'})")
print(f"bbox space: {past_1000}/{detections} boxes exceed 1000, {at_1000} sit exactly on 1000, "
      f"largest coord/panel-dimension = {max_ratio:.2f}")
print(f"  verdict: {'PIXELS' if past_1000 or max_ratio > 0.02 and at_1000 == 0 else 'STILL 0-1000 GRID'}")
missing_refs = [k for k, r in live.items() if not r["ref_image_uris"] or not r["embedding_uri"]]
print(f"live characters missing a ref crop or embedding: {len(missing_refs)}")

# 3. purity and fragmentation against the eyeball pass. Purity is per cluster, fragmentation is per
# real person. Both are computed off page-space geometry, so a re-crop and a /characters/reset do not
# invalidate the truth file.
truth = json.load(open(TRUTH)) if os.path.exists(TRUTH) else None
if not truth:
    print(f"\nno truth file at {TRUTH}, skipping purity and fragmentation")
elif truth["chapter_id"] != CHAPTER:
    print(f"\ntruth file is for chapter {truth['chapter_id'][:8]}, not this one. skipping.")
else:
    rows, base = truth["occurrences"], truth.get("baseline", {})
    per_cluster, frag, unmatched = _score(rows, found)
    matched = len(rows) - len(unmatched)
    print(f"\ntruth: {len(rows)} labelled occurrences, {matched} matched an assignment "
          f"at IoU >= {IOU_MIN}, {len(unmatched)} unmatched")
    for u in unmatched:
        src = u.get("from_19_44", {})
        print(f"  unmatched  {u['person']:14} was ord {src.get('panel_order')} {src.get('local_id')} "
              f"box {u['box']}")
    print("purity per cluster, dominant person first:")
    for cid, cnt in sorted(per_cluster.items(), key=lambda kv: -sum(kv[1].values())):
        person, correct = cnt.most_common(1)[0]
        total = sum(cnt.values())
        want = (base.get("clusters") or {}).get(person)
        delta = ""
        if want:
            delta = (" = baseline" if (want["assignments"], want["correct"]) == (total, correct)
                     else f" vs baseline {want['correct']}/{want['assignments']} = {want['purity']:.2f}")
        print(f"  {_label(cid):22} {person:14} {correct}/{total} = {correct/total:.2f}{delta}")
        for other, n in cnt.most_common()[1:]:
            print(f"  {'':22} {'wrong: ' + other:14} {n}")
    print("fragmentation per person, ids holding their occurrences:")
    for person in truth.get("people", {}):
        want = (base.get("fragmentation") or {}).get(person)
        got = frag.get(person, 0)
        print(f"  {person:14} {got}" + ("" if want is None else
              (" = baseline" if got == want else f" vs baseline {want}")))
    scope_only = len(found) - matched
    print(f"assignments outside the truth's scope: {scope_only}/{len(found)}, never checked by eye")

if worked:
    p, v, people, assigns, pw, ph = worked
    print(f"\npanel_index {WORKED_EXAMPLE} ({p['panel_id']}), {pw}x{ph}:")
    if not people:
        print(f"  no boxed detection. skip={v.get('skip')!r} "
              f"dialogue_status={v.get('dialogue_status')!r}, "
              f"{len(v.get('characters') or [])} unboxed character entries")
    for ch in people:
        cid, conf = assigns.get(ch["local_id"], (None, None))
        print(f"  {ch['local_id']:10} {ch['bbox']!s:28} {_label(cid) if cid else '-- none --':22} "
              f"{'' if conf is None else f'{conf:.2f}'}")
else:
    print(f"\npanel_index {WORKED_EXAMPLE} not found in this chapter")