parentcalendar/phase0/score.py

"""Phase 0: score an extraction run against the golden set.

Usage:  python phase0/score.py private/runs/<timestamp>
        python phase0/score.py --selftest
"""
import json
import re
import sys
from datetime import date
from difflib import SequenceMatcher
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
GOLDEN = ROOT / "private" / "golden"
FIELDS = ["date", "end_date", "all_day", "start_time", "end_time", "kind"]
TARGETS = {"no_edit": 0.85, "recall": 0.95, "flagged": 0.80}  # Phase 0 gate from the spec


def norm(s):
    return re.sub(r"[^a-z0-9]", "", (s or "").lower())


def similar(a, b):
    a, b = norm(a), norm(b)
    if a and b and (a in b or b in a):
        return 1.0
    return SequenceMatcher(None, a, b).ratio()


def days_apart(a, b):
    try:
        return abs((date.fromisoformat(a) - date.fromisoformat(b)).days)
    except (TypeError, ValueError):
        return 999


def match(expected, actual):
    """Pairs (i, j): same date with a similar title first, then a near date (<= 7 days) with a very similar title."""
    pairs, used_e, used_a = [], set(), set()
    for same_day, threshold in ((True, 0.5), (False, 0.8)):
        candidates = sorted(
            ((similar(e["title"], a["title"]), i, j)
             for i, e in enumerate(expected) for j, a in enumerate(actual)
             if (e["date"] == a["date"]) == same_day and days_apart(e["date"], a["date"]) <= 7),
            reverse=True)
        for score, i, j in candidates:
            if score >= threshold and i not in used_e and j not in used_a:
                pairs.append((i, j))
                used_e.add(i)
                used_a.add(j)
    return pairs


def value(event, field):
    """An end_date equal to the start date means the same as no end_date."""
    v = event.get(field)
    return None if field == "end_date" and v == event.get("date") else v


def score_events(expected, actual):
    s = {"expected": 0, "perfect": 0, "wrong": 0, "missing": 0, "invented": 0,
         "wrong_flagged": 0, "invented_flagged": 0, "false_flags": 0,
         "field_ok": {f: 0 for f in FIELDS}, "matched": 0, "details": []}
    pairs = match(expected, actual)
    matched_e = {i for i, _ in pairs}
    matched_a = {j for _, j in pairs}
    for i, j in pairs:
        e, a = expected[i], actual[j]
        if e.get("optional"):
            continue
        s["expected"] += 1
        s["matched"] += 1
        diffs = [f for f in FIELDS if value(e, f) != value(a, f)]
        for f in FIELDS:
            s["field_ok"][f] += f not in diffs
        flagged = bool(a.get("flags"))
        if diffs:
            s["wrong"] += 1
            s["wrong_flagged"] += flagged
            s["details"].append(f"WRONG    {e['date']} {e['title']!r}: " + ", ".join(
                f"{f} {a.get(f)!r} should be {e.get(f)!r}" for f in diffs) + (f"  [flagged: {', '.join(a['flags'])}]" if flagged else "  [NOT flagged]"))
        else:
            s["perfect"] += 1
            s["false_flags"] += flagged
            if flagged:
                s["details"].append(f"OK+FLAG  {e['date']} {e['title']!r}  [{', '.join(a['flags'])}]")
    for i, e in enumerate(expected):
        if i not in matched_e and not e.get("optional"):
            s["expected"] += 1
            s["missing"] += 1
            s["details"].append(f"MISSING  {e['date']} {e['title']!r}")
    for j, a in enumerate(actual):
        if j not in matched_a:
            flagged = bool(a.get("flags"))
            s["invented"] += 1
            s["invented_flagged"] += flagged
            s["details"].append(f"EXTRA    {a['date']} {a['title']!r}" + ("  [flagged]" if flagged else "  [NOT flagged]"))
    return s


def score_menu(expected, actual):
    """A menu day is right when its items (joined, normalized) and no_lunch both match."""
    s = {"days": len(expected), "right": 0, "wrong": 0, "wrong_flagged": 0, "false_flags": 0, "details": []}
    got = {m["date"]: m for m in actual}
    for e in expected:
        a = got.pop(e["date"], None)
        if a is None:
            s["details"].append(f"MISSING  {e['date']}")
        elif norm("".join(e["items"])) == norm("".join(a["items"])) and e["no_lunch"] == a["no_lunch"]:
            s["right"] += 1
            s["false_flags"] += bool(a.get("flags"))
        else:
            s["wrong"] += 1
            s["wrong_flagged"] += bool(a.get("flags"))
            s["details"].append(f"WRONG    {e['date']}: {a['items']} (no_lunch={a['no_lunch']}) should be {e['items']} (no_lunch={e['no_lunch']})"
                                + ("  [flagged]" if a.get("flags") else "  [NOT flagged]"))
    for d in sorted(got):
        s["details"].append(f"EXTRA    {d}: {got[d]['items']}")
    return s


def rates(s):
    errors = s["wrong"] + s["invented"]
    return {
        "no_edit": s["perfect"] / max(1, s["expected"] + s["invented"]),
        "recall": s["matched"] / max(1, s["expected"]),
        "flagged": (s["wrong_flagged"] + s["invented_flagged"]) / errors if errors else 1.0,
    }


def add(total, s):
    for k, v in s.items():
        if isinstance(v, int):
            total[k] = total.get(k, 0) + v
        elif k == "field_ok":
            for f, n in v.items():
                total.setdefault(k, {}).setdefault(f, 0)
                total[k][f] += n


def main(run_dir):
    run_dir = Path(run_dir)
    totals, menu_total, tokens = {}, {"days": 0, "right": 0}, [0, 0]
    for path in sorted(run_dir.glob("*.json")):
        golden = json.loads((GOLDEN / path.name).read_text(encoding="utf-8"))
        run = json.loads(path.read_text(encoding="utf-8"))
        result = run["result"]
        tokens[0] += run["usage"]["input_tokens"]
        tokens[1] += run["usage"]["output_tokens"]
        ev = score_events(golden["events"], result["events"])
        r = rates(ev)
        print(f"\n== {path.stem}  ({golden['category']}, {run['usage']['input_tokens']}+{run['usage']['output_tokens']} tokens, {run['seconds']}s)")
        print(f"   events: {ev['perfect']}/{ev['expected']} perfect, {ev['wrong']} wrong, {ev['missing']} missing, "
              f"{ev['invented']} extra | no-edit {r['no_edit']:.0%}, recall {r['recall']:.0%}, errors flagged {r['flagged']:.0%}")
        if golden.get("menu_items"):
            m = score_menu(golden["menu_items"], result["menu_items"])
            print(f"   menu:   {m['right']}/{m['days']} days right, {m['wrong_flagged']}/{m['wrong']} wrong days flagged, "
                  f"{m['false_flags']} right days flagged")
            for k in ("days", "right", "wrong", "wrong_flagged", "false_flags"):
                menu_total[k] = menu_total.get(k, 0) + m[k]
            ev["details"] += m["details"]
        for line in ev["details"]:
            print("   " + line)
        if golden["category"] != "bad" and not golden.get("menu_items"):
            add(totals.setdefault("gate", {}), ev)
        add(totals.setdefault(golden["category"], {}), ev)

    print("\n== Totals")
    for name, t in totals.items():
        if not t.get("expected"):
            continue
        r = rates(t)
        line = f"   {name:6} no-edit {r['no_edit']:.0%}, recall {r['recall']:.0%}, errors flagged {r['flagged']:.0%}"
        if name == "gate":
            line += "  -> " + ", ".join(f"{k} {'PASS' if r[k] >= v else 'FAIL'}" for k, v in TARGETS.items())
        print(line)
    if totals.get("gate", {}).get("matched"):
        g = totals["gate"]
        print("   field accuracy (gate pages): " + ", ".join(f"{f} {n / g['matched']:.0%}" for f, n in g["field_ok"].items()))
    if menu_total["days"]:
        print(f"   menus  {menu_total['right']}/{menu_total['days']} days right, "
              f"{menu_total['wrong_flagged']}/{menu_total['wrong']} wrong days flagged, {menu_total['false_flags']} right days flagged")
    print(f"   tokens {tokens[0]} in, {tokens[1]} out")


def selftest():
    ev = lambda d, t, **kw: {"date": d, "end_date": None, "all_day": True, "start_time": None,
                             "end_time": None, "kind": "event", "title": t, **kw}
    expected = [ev("2026-10-02", "Spirit Day"), ev("2026-10-20", "PTO", all_day=False, start_time="16:00"),
                ev("2026-10-29", "Book Logs Due", kind="deadline"), ev("2026-10-31", "Halloween", optional=True)]
    actual = [ev("2026-10-02", "Spirit day!", end_date="2026-10-02"),         # perfect (same-day end_date)
              ev("2026-10-20", "PTO Meeting", all_day=False, start_time="04:00", flags=["low confidence"]),  # wrong, flagged
              ev("2026-10-03", "Pizza Party"),                                # invented, not flagged
              ev("2026-10-31", "Happy Halloween")]                            # optional: ignored
    s = score_events(expected, actual)
    assert (s["expected"], s["perfect"], s["wrong"], s["missing"], s["invented"]) == (3, 1, 1, 1, 1), s
    r = rates(s)
    assert r["no_edit"] == 1 / 4 and r["recall"] == 2 / 3 and r["flagged"] == 1 / 2, r
    # a near-date match counts as found-but-wrong, not missing + extra
    s = score_events([ev("2026-10-05", "Coin War")], [ev("2026-10-06", "PTO's Coin War")])
    assert (s["wrong"], s["missing"], s["invented"]) == (1, 0, 0), s
    m = score_menu([{"date": "2026-10-05", "items": ["Pork Chopette", "WG bun"], "no_lunch": False}],
                   [{"date": "2026-10-05", "items": ["Pork Chopette WG Bun"], "no_lunch": False}])
    assert m["right"] == 1, m
    print("selftest ok")


if __name__ == "__main__":
    sys.stdout.reconfigure(encoding="utf-8")  # menu text can hold characters the Windows console code page can't
    if sys.argv[1:] == ["--selftest"]:
        selftest()
    elif len(sys.argv) == 2:
        main(sys.argv[1])
    else:
        sys.exit(__doc__)