Blame for scripts/eval_closed_generators.py

"""
Evaluate the live detector on closed-generator images AND real photos (P0 benchmark).

Why this exists
---------------
We deliberately never *train* on closed-API generator outputs (ToS rule,
docs/decisions.md 2026-05-14), so we detect them only by family resemblance to
the open generators we can train on. That leaves us partly blind to how well we
do on the consumer tools people actually use. This script closes that gap on
the MEASUREMENT side.

The May 2026 run was 11 screenshotted AI images with **no real-photo arm**, so
it could report a detection rate but not a false-positive rate — the exact
number that separates a useful detector from one that just says "AI" a lot.
This version reports both, per tool and per condition, plus a threshold sweep
so the production 0.55 cutoff can be re-checked against a set large enough to
justify it. Spec: docs/closed-generator-position-review-2026-07-30.md section 6.

IMPORTANT — licensing / privacy guardrails
-------------------------------------------
- These images are for EVALUATION ONLY. They must never be added to a training
  manifest or `data/splits*/`. Training reads only the split CSVs, so images
  placed under the eval folder cannot leak into training by accident.
- The default eval folder lives under `data/` which is fully .gitignored, so
  ToS-encumbered images are never committed to the repo.
- Nothing is uploaded anywhere; this runs the detector locally.

Folder layout
-------------
    data/closed_eval/
      ai/<tool>/original/      untouched downloads   (C2PA path testable)
      ai/<tool>/screenshot/    screenshots of those  (the real sharing path)
      ai/<tool>/logo_intact/   generator logo left in frame
      real/phone_original/     own camera photos
      real/phone_screenshot/   the SAME photos, screenshotted  <- key control
      real/edited/             filtered / edited real photos
      real/stock/              web / stock real photos

Everything under `ai/` is scored as known-synthetic; everything under `real/`
as known-authentic. Folders starting with `_` (e.g. `_legacy_may2026/`) are
skipped unless --include-legacy is passed.

If neither `ai/` nor `real/` exists, the target is treated as a flat set of AI
images (the pre-P0 behaviour) and a warning is printed — no false-positive
rate can be computed in that mode.

Usage
-----
    # whole benchmark
    python scripts/eval_closed_generators.py --dir data/closed_eval

    # one condition only
    python scripts/eval_closed_generators.py --dir data/closed_eval/ai/chatgpt/original

    # freeze a report to diff against after a model change
    python scripts/eval_closed_generators.py --dir data/closed_eval \
        --report-out data/reports/closed_eval_v0.4.0.json
"""
from __future__ import annotations

import argparse
import json
from dataclasses import dataclass, field
from pathlib import Path

_IMG_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp"}
_PROD_THRESHOLD = 0.55
_SWEEP = [0.50, 0.55, 0.60, 0.65, 0.70, 0.75, 0.80, 0.85, 0.90, 0.95]

# Byte markers for an embedded C2PA manifest. Used only to sanity-check the
# provenance reader against the files themselves — see the self-check below.
_C2PA_MARKERS = (b"caBX", b"jumbf", b"c2pa")

_NAN = float("nan")


@dataclass
class Group:
    """One reporting bucket: an (arm, tool, condition) triple of images."""

    name: str
    arm: str  # "ai" (known synthetic) | "real" (known authentic)
    paths: list[Path]
    tool: str | None = None
    condition: str | None = None
    records: list[dict] = field(default_factory=list)
    errors: int = 0


def _images(d: Path) -> list[Path]:
    return [p for p in sorted(d.rglob("*")) if p.suffix.lower() in _IMG_EXTS]


def _skipped(d: Path) -> list[Path]:
    """Image-like files the scanner cannot read (e.g. iPhone .HEIC).

    Reported loudly rather than dropped: a folder that looks full but is half
    unreadable produces a rate computed on a silently different sample. This is
    the same failure shape as the 2026-08 C2PA outage - see docs/decisions.md.
    """
    return [
        p for p in sorted(d.rglob("*"))
        if p.is_file()
        and p.suffix.lower() not in _IMG_EXTS
        and p.suffix.lower() in {".heic", ".heif", ".avif", ".tif", ".tiff", ".gif", ".dng", ".raw"}
    ]


def _loose(d: Path) -> list[Path]:
    return [
        p for p in sorted(d.iterdir())
        if p.is_file() and p.suffix.lower() in _IMG_EXTS
    ]


def _subdirs(d: Path, include_legacy: bool) -> list[Path]:
    return [
        p for p in sorted(d.iterdir())
        if p.is_dir() and (include_legacy or not p.name.startswith("_"))
    ]


def _discover(root: Path, include_legacy: bool) -> tuple[list[Group], list[str]]:
    """Map the folder tree onto labelled groups. Returns (groups, warnings)."""
    groups: list[Group] = []
    warnings: list[str] = []
    ai_root, real_root = root / "ai", root / "real"

    if not ai_root.is_dir() and not real_root.is_dir():
        # Pre-P0 flat layout, or a single condition folder passed directly.
        imgs = _images(root)
        if imgs:
            groups.append(Group(name=root.name or str(root), arm="ai", paths=imgs))
        warnings.append(
            "No ai/ or real/ subfolder found - treating every image as known-AI. "
            "No false-positive rate can be computed in this mode."
        )
        return groups, warnings

    if ai_root.is_dir():
        for tool_dir in _subdirs(ai_root, include_legacy):
            for cond_dir in _subdirs(tool_dir, include_legacy):
                imgs = _images(cond_dir)
                if imgs:
                    groups.append(Group(
                        name=f"ai/{tool_dir.name}/{cond_dir.name}", arm="ai",
                        paths=imgs, tool=tool_dir.name, condition=cond_dir.name,
                    ))
            stray = _loose(tool_dir)
            if stray:
                groups.append(Group(
                    name=f"ai/{tool_dir.name}/(loose)", arm="ai", paths=stray,
                    tool=tool_dir.name, condition="(loose)",
                ))
                warnings.append(
                    f"ai/{tool_dir.name}/ has {len(stray)} loose image(s) outside a "
                    f"condition subfolder - they cannot be paired original<->screenshot."
                )

    if real_root.is_dir():
        for bucket_dir in _subdirs(real_root, include_legacy):
            imgs = _images(bucket_dir)
            if imgs:
                groups.append(Group(
                    name=f"real/{bucket_dir.name}", arm="real", paths=imgs,
                    condition=bucket_dir.name,
                ))

    if include_legacy:
        for d in sorted(root.iterdir()):
            if d.is_dir() and d.name.startswith("_") and _images(d):
                groups.append(Group(name=d.name, arm="ai", paths=_images(d)))

    for d in (ai_root, real_root):
        if not d.is_dir():
            continue
        skipped = _skipped(d)
        if skipped:
            kinds = sorted({s.suffix.lower() for s in skipped})
            warnings.append(
                f"{len(skipped)} file(s) under {d.name}/ are NOT readable by the "
                f"scanner and were EXCLUDED from every number below "
                f"({', '.join(kinds)}). They are not counted as collected - the "
                f"rates come from the readable files only."
            )

    if not any(g.arm == "real" for g in groups):
        warnings.append(
            "Real arm is empty - a detection rate alone cannot tell you whether the "
            "detector is accurate or merely biased toward 'AI'. Populate "
            "real/phone_original/ and real/phone_screenshot/ before drawing "
            "conclusions."
        )
    return groups, warnings


def _classify(p_ai: float, threshold: float) -> str:
    """Reproduce the production rule: the top class wins only if it clears the bar."""
    if p_ai >= threshold:
        return "ai_generated"
    if (1.0 - p_ai) >= threshold:
        return "authentic"
    return "uncertain"


def _rates(records: list[dict], arm: str, threshold: float) -> dict:
    n = len(records)
    if n == 0:
        return {"n": 0}
    calls = [_classify(r["p_ai"], threshold) for r in records]
    said_ai = sum(c == "ai_generated" for c in calls)
    said_real = sum(c == "authentic" for c in calls)
    unsure = sum(c == "uncertain" for c in calls)
    out = {
        "n": n,
        "uncertain_rate": unsure / n,
        "mean_p_ai": sum(r["p_ai"] for r in records) / n,
    }
    if arm == "ai":
        out["detect_rate"] = said_ai / n
        out["miss_rate"] = said_real / n           # known-AI image called real
    else:
        out["false_positive_rate"] = said_ai / n   # real photo called AI
        out["correct_rate"] = said_real / n
    return out


def _headline(r: dict, arm: str) -> str:
    if arm == "ai":
        return (f"detected_AI={r['detect_rate']:.1%}  "
                f"missed_as_real={r['miss_rate']:.1%}")
    return (f"FALSE_POSITIVE={r['false_positive_rate']:.1%}  "
            f"correct={r['correct_rate']:.1%}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--dir", type=Path, default=Path("data/closed_eval"),
        help="Benchmark root (expects ai/ and real/ subfolders).",
    )
    parser.add_argument("--report-out", type=Path, default=None)
    parser.add_argument(
        "--include-legacy", action="store_true",
        help="Include _-prefixed folders such as _legacy_may2026/. Off by default: "
             "that sample is 11 screenshot-only images with no real arm and must "
             "not be merged into P0 counts.",
    )
    parser.add_argument(
        "--threshold", type=float, default=_PROD_THRESHOLD,
        help="Decision threshold for the headline numbers (default: production 0.55).",
    )
    args = parser.parse_args()

    if not args.dir.exists():
        raise SystemExit(
            f"Eval folder {args.dir} does not exist. See this script's header."
        )

    # Import here so --help works without loading torch/CLIP.
    from deepfake_scanner.config import settings
    from deepfake_scanner.detectors.ensemble import run_ensemble
    # The SAME decoder the API uses. Calling Image.open().convert("RGB") here
    # instead measured a preprocessing path production does not run: 31 of the
    # 100 real benchmark photographs carry an EXIF rotation tag, so the
    # benchmark scored them sideways and reported 21% where production-
    # equivalent decoding gives 24%. Found in review 2026-09-14. Never decode
    # independently of the service in this script.
    from deepfake_scanner.preprocess import decode
    from deepfake_scanner.provenance.c2pa import check_c2pa

    groups, warnings = _discover(args.dir, args.include_legacy)
    if not groups:
        raise SystemExit(f"No images found under {args.dir}.")

    print(f"Detector model_version: {settings.model_version}")
    print(f"Decision threshold: {args.threshold}\n")
    for w in warnings:
        print(f"  ! {w}")
    if warnings:
        print()

    # --- inference ---------------------------------------------------------
    c2pa_bytes_seen = c2pa_read_ok = 0
    for g in groups:
        for path in g.paths:
            try:
                raw = path.read_bytes()
                out = run_ensemble(decode(raw))
            except Exception as e:  # corrupt/unsupported file - count, don't crash
                g.errors += 1
                print(f"  [skip] {path.name}: {e}")
                continue
            prov = check_c2pa(raw)
            has_marker = any(m in raw for m in _C2PA_MARKERS)
            c2pa_bytes_seen += int(has_marker)
            c2pa_read_ok += int(prov.present)
            g.records.append({
                "file": path.name,
                "p_ai": out.probabilities.ai_generated,
                "verdict": out.verdict,
                "c2pa_present": prov.present,
                "c2pa_issuer": prov.issuer,
                "c2pa_marker_in_bytes": has_marker,
                # A validated chain declaring AI content decides the verdict in
                # production (see verdict.py), so the benchmark measures the
                # same combination the service actually ships.
                "provenance_says_ai": bool(prov.present and prov.valid
                                           and prov.ai_declared),
            })

    report: dict = {
        "model_version": settings.model_version,
        "threshold": args.threshold,
        "groups": {},
        "warnings": warnings,
    }

    # --- per-group report --------------------------------------------------
    for arm, title in (("ai", "AI ARM (known synthetic)"),
                       ("real", "REAL ARM (known authentic)")):
        members = [g for g in groups if g.arm == arm and g.records]
        if not members:
            continue
        print(title)
        print("-" * len(title))
        for g in members:
            r = _rates(g.records, arm, args.threshold)
            c2pa_n = sum(rec["c2pa_present"] for rec in g.records)
            report["groups"][g.name] = {
                **r, "arm": arm, "tool": g.tool, "condition": g.condition,
                "errors": g.errors, "c2pa_present_n": c2pa_n,
            }
            print(f"  {g.name:36s} n={r['n']:<4d} {_headline(r, arm)}  "
                  f"uncertain={r['uncertain_rate']:.1%}  "
                  f"mean_p_ai={r['mean_p_ai']:.3f}  c2pa={c2pa_n}/{r['n']}")
        pooled = [rec for g in members for rec in g.records]
        r = _rates(pooled, arm, args.threshold)
        report[f"overall_{arm}"] = r
        print(f"  {'ALL ' + arm.upper():36s} n={r['n']:<4d} {_headline(r, arm)}  "
              f"uncertain={r['uncertain_rate']:.1%}\n")

    # --- threshold sweep ---------------------------------------------------
    ai_recs = [rec for g in groups if g.arm == "ai" for rec in g.records]
    real_recs = [rec for g in groups if g.arm == "real" for rec in g.records]
    both = ai_recs + real_recs

    print("THRESHOLD SWEEP")
    print("-" * 15)
    print(f"  {'thresh':>6s}  {'AI detected':>11s}  {'AI missed':>9s}  "
          f"{'real FPR':>8s}  {'real correct':>12s}  {'uncertain(all)':>14s}")
    sweep = {}
    for t in _SWEEP:
        a = _rates(ai_recs, "ai", t) if ai_recs else {}
        b = _rates(real_recs, "real", t) if real_recs else {}
        unc = sum(_classify(r["p_ai"], t) == "uncertain" for r in both) / max(len(both), 1)
        sweep[f"{t:.2f}"] = {"ai": a, "real": b, "uncertain_rate_all": unc}
        print(f"  {t:>6.2f}  {a.get('detect_rate', _NAN):>10.1%}  "
              f"{a.get('miss_rate', _NAN):>8.1%}  "
              f"{b.get('false_positive_rate', _NAN):>7.1%}  "
              f"{b.get('correct_rate', _NAN):>11.1%}  {unc:>13.1%}")
    report["threshold_sweep"] = sweep
    if not real_recs:
        print("\n  (real FPR columns are nan because the real arm is empty)")

    # --- production routing: provenance overriding the model ----------------
    # verdict.py lets a validated C2PA chain that declares AI content override
    # the classifier, one-directionally. Report what the shipped combination
    # scores, not just the model in isolation.
    print()
    print("WITH PROVENANCE ROUTING (what production actually returns)")
    print("-" * 57)
    routed = {}
    for arm, members in (("ai", [g for g in groups if g.arm == "ai" and g.records]),
                         ("real", [g for g in groups if g.arm == "real" and g.records])):
        for g in members:
            n = len(g.records)
            model_ai = sum(_classify(r["p_ai"], args.threshold) == "ai_generated"
                           for r in g.records)
            both = sum(_classify(r["p_ai"], args.threshold) == "ai_generated"
                       or r["provenance_says_ai"] for r in g.records)
            rescued = sum(r["provenance_says_ai"]
                          and _classify(r["p_ai"], args.threshold) != "ai_generated"
                          for r in g.records)
            label = "detected" if arm == "ai" else "FALSE POSITIVE"
            routed[g.name] = {"n": n, "model_only": model_ai / n, "routed": both / n,
                              "rescued_by_provenance": rescued}
            print(f"  {g.name:36s} n={n:<4d} {label}: model {model_ai/n:5.1%} "
                  f"-> routed {both/n:5.1%}   ({rescued} decided by signature)")
    report["provenance_routing"] = routed
    if any(g.arm == "real" for g in groups):
        print("  NOTE: on the real arm the routed column is the false-positive rate;")
        print("        provenance can only push toward ai_generated, so it can only rise.")

    # --- provenance self-check ---------------------------------------------
    # A silently broken C2PA reader is indistinguishable from "no image carried a
    # manifest", so compare what the reader found against what is in the bytes.
    print(f"\nC2PA: reader found {c2pa_read_ok} manifest(s); "
          f"{c2pa_bytes_seen} file(s) contain an embedded C2PA marker.")
    if c2pa_bytes_seen and not c2pa_read_ok:
        print("  ! The provenance reader found NOTHING in files that demonstrably "
              "carry a C2PA manifest.")
        print("  ! Treat every c2pa column above as UNMEASURED, not as absent.")
    report["c2pa_selfcheck"] = {
        "reader_found": c2pa_read_ok,
        "marker_in_bytes": c2pa_bytes_seen,
    }

    if args.report_out is not None:
        args.report_out.parent.mkdir(parents=True, exist_ok=True)
        report["per_image"] = {g.name: g.records for g in groups if g.records}
        with args.report_out.open("w", encoding="utf-8") as fh:
            json.dump(report, fh, indent=2, sort_keys=True)
        print(f"\nsaved report to {args.report_out}")


if __name__ == "__main__":
    main()