mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-08-07 14:38:35 +02:00
242 lines
9.6 KiB
Python
242 lines
9.6 KiB
Python
"""Partition China-AIGC carriers into VENDOR COHORTS by their TC260 producer code.
|
|
|
|
THE PROBLEM THIS SOLVES
|
|
Coverage of uncovered vendors is the largest remaining detection lever
|
|
(`docs/verification-plan.md`, "Where detection work should go next"), and it is
|
|
blocked on EVIDENCE: nothing may be registered off a single frame, and the
|
|
previous session found exactly one confirmed positive each for `千问` and `百度`.
|
|
Harvesting more by PIXELS is circular -- a detector is what we are trying to
|
|
build -- and the generic shared-tail probe is too weak to label with (0.407 on a
|
|
bold positive against a clean p99 of 0.298; see `cjk_tail_probe.py`).
|
|
|
|
THE KEY
|
|
The TC260 label is not anonymous. Its `ContentProducer` field carries the
|
|
producer's Chinese Unified Social Credit Code (USCC), e.g.
|
|
``001191110102MACQD9K64010000`` -> USCC ``91110102MACQD9K640``, which names a
|
|
specific legal entity. So the metadata partitions carriers into per-ENTITY
|
|
cohorts without looking at a single pixel. A cohort is a LABEL: once one frame
|
|
in it is eyeballed, every frame in it is a labelled example of that vendor's
|
|
mark. That is what turns "one confirmed positive" into "30+ per vendor".
|
|
|
|
CLAUDE.md's "the generic TC260 label names no specific vendor" is about the
|
|
label MARKER (the bare presence of `TC260:AIGC`), which indeed names nobody.
|
|
The producer FIELD inside the block is a different thing and does name one.
|
|
|
|
Caveat kept in view: the code names the SIGNING ENTITY, which is not always the
|
|
consumer brand (an aggregator or a cloud host signs for several apps, and one
|
|
vendor can hold several codes). So a cohort is a strong grouping key and a
|
|
hypothesis about the brand -- the brand itself is settled by reading the crop,
|
|
which is what `--sheets` is for.
|
|
|
|
WHAT IT COSTS
|
|
Metadata only. The expensive pixel pass is NOT re-run: which detectors fired is
|
|
joined from `_visible_positives.jsonl` (a completed local evaluation artifact), per
|
|
the standing rule against relaunching finished sweeps to re-check them.
|
|
|
|
DATA SAFETY
|
|
Treat input datasets as sensitive and read-only. Contact sheets stay under
|
|
`.local-eval/`; nothing generated here is committed.
|
|
|
|
uv run python scripts/vendor_cohort_harvest.py
|
|
uv run python scripts/vendor_cohort_harvest.py --report-only --sheets 12
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import glob
|
|
import json
|
|
import os
|
|
import sys
|
|
from collections import Counter
|
|
from concurrent.futures import ProcessPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
CORPUS = REPO / ".local-eval" / "originals"
|
|
OUT = REPO / ".local-eval" / "vendor-cohorts.jsonl"
|
|
FIRED = REPO / ".local-eval" / "visible-positives.jsonl"
|
|
SHEET_DIR = REPO / ".local-eval" / "vendor-cohort-sheets"
|
|
|
|
# A producer code is `001` + `1` + USCC(18) + a 5-digit app/product suffix, so two
|
|
# codes sharing the USCC are the same legal entity registering different products.
|
|
# Slicing is defensive: anything not matching the layout is grouped by its raw value.
|
|
_USCC_START, _USCC_END = 4, 22
|
|
|
|
|
|
def uscc_of(code: str) -> str:
|
|
"""The 18-char Unified Social Credit Code embedded in a TC260 producer code."""
|
|
if len(code) >= _USCC_END and code[:3] == "001":
|
|
return code[_USCC_START:_USCC_END]
|
|
return code
|
|
|
|
|
|
def _one(path_str: str) -> dict[str, Any] | None:
|
|
from remove_ai_watermarks.metadata import aigc_label
|
|
|
|
try:
|
|
label = aigc_label(Path(path_str))
|
|
except Exception:
|
|
return None
|
|
if not label:
|
|
return None
|
|
producer = str(label.get("ContentProducer") or "")
|
|
return {
|
|
"path": path_str,
|
|
"producer": producer,
|
|
"uscc": uscc_of(producer),
|
|
"propagator": str(label.get("ContentPropagator") or ""),
|
|
"service_provider": str(label.get("ServiceProvider") or ""),
|
|
}
|
|
|
|
|
|
def load_fired() -> dict[str, list[str]]:
|
|
"""Map each path to detector keys from a completed local evaluation artifact."""
|
|
if not FIRED.exists():
|
|
print(f"WARNING: {FIRED.name} missing; cohorts will show no detector state")
|
|
return {}
|
|
out: dict[str, list[str]] = {}
|
|
for line in FIRED.read_text(encoding="utf-8").splitlines():
|
|
if not line.strip():
|
|
continue
|
|
rec = json.loads(line)
|
|
out[rec["path"]] = rec.get("keys") or []
|
|
return out
|
|
|
|
|
|
def scan(limit: int, workers: int, out_path: Path) -> list[dict[str, Any]]:
|
|
pool = sorted(glob.glob(str(CORPUS / "*" / "*")))
|
|
if limit:
|
|
pool = pool[:limit]
|
|
print(f"scanning {len(pool)} local files for TC260 labels workers={workers}", flush=True)
|
|
rows: list[dict[str, Any]] = []
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(out_path, "w", encoding="utf-8") as fh, ProcessPoolExecutor(max_workers=workers) as ex:
|
|
futures = [ex.submit(_one, p) for p in pool]
|
|
for i, fut in enumerate(as_completed(futures), 1):
|
|
try:
|
|
rec = fut.result()
|
|
except Exception: # noqa: S112 -- one bad file must not kill the scan
|
|
continue
|
|
if rec is not None:
|
|
fh.write(json.dumps(rec) + "\n")
|
|
rows.append(rec)
|
|
if i % 5000 == 0:
|
|
fh.flush()
|
|
print(f" {i}/{len(pool)} carriers={len(rows)}", flush=True)
|
|
return rows
|
|
|
|
|
|
def report(rows: list[dict[str, Any]], fired: dict[str, list[str]], min_size: int) -> None:
|
|
by_uscc: dict[str, list[dict[str, Any]]] = {}
|
|
for r in rows:
|
|
by_uscc.setdefault(r["uscc"], []).append(r)
|
|
|
|
print(f"\n{'=' * 92}\nVENDOR COHORTS ({len(rows)} TC260 carriers, {len(by_uscc)} distinct entities)\n{'=' * 92}")
|
|
print("\n`fires` = share of the cohort where SOME registered detector fires.")
|
|
print("A large cohort with a low fire rate is an uncovered vendor -- the harvest target.\n")
|
|
print(f"{'entity (USCC)':22s} {'n':>6s} {'fires':>7s} {'detectors seen':38s} {'products':>8s}")
|
|
print("-" * 92)
|
|
|
|
cohorts = sorted(by_uscc.items(), key=lambda kv: -len(kv[1]))
|
|
for uscc, members in cohorts:
|
|
if len(members) < min_size:
|
|
continue
|
|
keys: Counter[str] = Counter()
|
|
hit = 0
|
|
for m in members:
|
|
ks = fired.get(m["path"], [])
|
|
if ks:
|
|
hit += 1
|
|
keys.update(ks)
|
|
seen = ", ".join(f"{k}:{c}" for k, c in keys.most_common(4)) or "-- none --"
|
|
products = len({m["producer"] for m in members})
|
|
print(f"{uscc:22s} {len(members):6d} {100 * hit / len(members):6.1f}% {seen:38s} {products:8d}")
|
|
|
|
small = sum(1 for _, m in cohorts if len(m) < min_size)
|
|
if small:
|
|
print(f"\n({small} cohorts below --min-size {min_size} not shown)")
|
|
|
|
|
|
def _bands(img: Any, width: int, band: int) -> list[Any]:
|
|
"""Full-width top and bottom bands, scaled to a readable common width.
|
|
|
|
An unregistered vendor's placement is unknown, so cropping the bottom-RIGHT
|
|
corner (where the marks we already cover happen to sit) would beg the
|
|
question. Full-width bands catch any horizontal position, and the two bands
|
|
together cover every corner the standard's implementers actually use.
|
|
"""
|
|
import cv2
|
|
|
|
h = img.shape[0]
|
|
strip = max(24, int(h * band))
|
|
out = []
|
|
for piece in (img[:strip], img[h - strip :]):
|
|
scale = width / max(1, piece.shape[1])
|
|
out.append(cv2.resize(piece, (width, max(12, int(piece.shape[0] * scale))), interpolation=cv2.INTER_AREA))
|
|
return out
|
|
|
|
|
|
def sheets(rows: list[dict[str, Any]], fired: dict[str, list[str]], per: int, min_size: int) -> None:
|
|
"""Top/bottom bands per uncovered cohort, so the vendor and mark can be read off."""
|
|
import cv2
|
|
import numpy as np
|
|
|
|
from remove_ai_watermarks.image_io import imread
|
|
|
|
by_uscc: dict[str, list[dict[str, Any]]] = {}
|
|
for r in rows:
|
|
by_uscc.setdefault(r["uscc"], []).append(r)
|
|
|
|
SHEET_DIR.mkdir(parents=True, exist_ok=True)
|
|
print(f"\nwriting contact sheets -> {SHEET_DIR}")
|
|
|
|
for uscc, members in sorted(by_uscc.items(), key=lambda kv: -len(kv[1])):
|
|
if len(members) < min_size:
|
|
continue
|
|
quiet = [m for m in members if not fired.get(m["path"])]
|
|
if not quiet:
|
|
continue
|
|
width = 900
|
|
tiles: list[Any] = []
|
|
for m in quiet[:per]:
|
|
img = imread(m["path"])
|
|
if img is None:
|
|
continue
|
|
for b in _bands(img, width, 0.10):
|
|
tiles.append(b)
|
|
tiles.append(np.full((3, width, 3), 60, np.uint8))
|
|
if tiles:
|
|
dest = SHEET_DIR / f"{uscc}_n{len(members)}_quiet{len(quiet)}.png"
|
|
cv2.imwrite(str(dest), np.vstack(tiles))
|
|
print(f" {dest.name} ({len(tiles) // 4} frames)")
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--limit", type=int, default=0, help="cap files scanned (0 = all)")
|
|
ap.add_argument("--workers", type=int, default=max(1, (os.cpu_count() or 4) - 2))
|
|
ap.add_argument("--out", type=Path, default=OUT)
|
|
ap.add_argument("--report-only", action="store_true")
|
|
ap.add_argument("--min-size", type=int, default=5, help="hide cohorts smaller than this")
|
|
ap.add_argument("--sheets", type=int, default=0, help="crops per cohort contact sheet")
|
|
a = ap.parse_args()
|
|
|
|
if a.report_only:
|
|
rows = [json.loads(x) for x in a.out.read_text(encoding="utf-8").splitlines() if x.strip()]
|
|
else:
|
|
rows = scan(a.limit, a.workers, a.out)
|
|
|
|
fired = load_fired()
|
|
report(rows, fired, a.min_size)
|
|
if a.sheets:
|
|
sheets(rows, fired, a.sheets, a.min_size)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|