Spaces:
Running
Running
github-actions[bot]
Deploy from GitHub 1af737237f5d10dfa9cf12849d98f47c3b30ce37 (filtered: code + Dockerfile + README + NOTICES only)
a2f55b2 | """ | |
| 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") | |
| 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() | |