mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-10-14 05:20:46 +02:00
172 lines
6.8 KiB
Python
172 lines
6.8 KiB
Python
"""Shared validation for the publication-cleared engine-selection image matrix."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import re
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from PIL import Image
|
|
from watermark_benchmark import sha256_file
|
|
|
|
REUSE_BASIS = "project-maintainer-public-test-clearance"
|
|
CONTENT_FIELDS = (
|
|
"pair_id",
|
|
"provider",
|
|
"model",
|
|
"file",
|
|
"width",
|
|
"height",
|
|
"content_stratum",
|
|
"sha256",
|
|
"source_payload_sha256",
|
|
"reuse_basis",
|
|
"prompt",
|
|
)
|
|
_SAFE_ID = re.compile(r"[a-z0-9][a-z0-9-]*")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ContentFixture:
|
|
"""One validated source row from the publication-cleared content matrix."""
|
|
|
|
pair_id: str
|
|
provider: str
|
|
model: str
|
|
path: Path
|
|
width: int
|
|
height: int
|
|
content_stratum: str
|
|
sha256: str
|
|
source_payload_sha256: str
|
|
reuse_basis: str
|
|
prompt: str
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""Return a stable filename-safe carrier identity."""
|
|
return f"{self.pair_id}-{self.provider}"
|
|
|
|
@property
|
|
def seed(self) -> int:
|
|
"""Derive a stable 60-bit seed from the source bytes."""
|
|
return int(self.sha256[:15], 16)
|
|
|
|
|
|
def _require_text(row: dict[str, str | None], field: str, *, location: str) -> str:
|
|
value = row.get(field)
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError(f"{location}: {field} must be non-empty")
|
|
return value.strip()
|
|
|
|
|
|
def _require_sha256(row: dict[str, str | None], field: str, *, location: str) -> str:
|
|
value = _require_text(row, field, location=location)
|
|
if len(value) != 64 or any(character not in "0123456789abcdef" for character in value):
|
|
raise ValueError(f"{location}: {field} must be a lowercase sha256")
|
|
return value
|
|
|
|
|
|
def load_content_manifest(source_manifest: Path) -> list[ContentFixture]:
|
|
"""Load and validate one balanced, content-stratified source manifest."""
|
|
manifest = source_manifest.resolve()
|
|
with manifest.open(newline="", encoding="utf-8") as stream:
|
|
reader = csv.DictReader(stream)
|
|
if tuple(reader.fieldnames or ()) != CONTENT_FIELDS:
|
|
raise ValueError(f"{manifest}: expected columns {', '.join(CONTENT_FIELDS)}")
|
|
rows = list(reader)
|
|
if not rows:
|
|
raise ValueError(f"{manifest}: source manifest contains no rows")
|
|
|
|
fixtures: list[ContentFixture] = []
|
|
paths: set[Path] = set()
|
|
hashes: set[str] = set()
|
|
pair_provider: set[tuple[str, str]] = set()
|
|
stratum_provider: set[tuple[str, str]] = set()
|
|
pairs: dict[str, list[ContentFixture]] = defaultdict(list)
|
|
|
|
for line_number, row in enumerate(rows, start=2):
|
|
location = f"{manifest}:{line_number}"
|
|
if None in row:
|
|
raise ValueError(f"{location}: row contains extra columns")
|
|
pair_id = _require_text(row, "pair_id", location=location)
|
|
provider = _require_text(row, "provider", location=location)
|
|
if not _SAFE_ID.fullmatch(pair_id) or not _SAFE_ID.fullmatch(provider):
|
|
raise ValueError(f"{location}: pair_id and provider must be lowercase filename-safe identifiers")
|
|
source_sha256 = _require_sha256(row, "sha256", location=location)
|
|
source_payload_sha256 = _require_sha256(row, "source_payload_sha256", location=location)
|
|
reuse_basis = _require_text(row, "reuse_basis", location=location)
|
|
if reuse_basis != REUSE_BASIS:
|
|
raise ValueError(f"{location}: unsupported reuse_basis {reuse_basis!r}")
|
|
try:
|
|
width = int(_require_text(row, "width", location=location))
|
|
height = int(_require_text(row, "height", location=location))
|
|
except ValueError as exc:
|
|
raise ValueError(f"{location}: width and height must be positive integers") from exc
|
|
if width <= 0 or height <= 0:
|
|
raise ValueError(f"{location}: width and height must be positive integers")
|
|
|
|
relative_path = Path(_require_text(row, "file", location=location))
|
|
path = (manifest.parent / relative_path).resolve()
|
|
if not path.is_relative_to(manifest.parent):
|
|
raise ValueError(f"{location}: file must stay inside {manifest.parent}")
|
|
if path in paths:
|
|
raise ValueError(f"{location}: duplicate source path {relative_path}")
|
|
paths.add(path)
|
|
if source_sha256 in hashes:
|
|
raise ValueError(f"{location}: duplicate source sha256 {source_sha256}")
|
|
hashes.add(source_sha256)
|
|
identity = (pair_id, provider)
|
|
if identity in pair_provider:
|
|
raise ValueError(f"{location}: duplicate pair/provider {pair_id}/{provider}")
|
|
pair_provider.add(identity)
|
|
|
|
content_stratum = _require_text(row, "content_stratum", location=location)
|
|
stratum_identity = (content_stratum, provider)
|
|
if stratum_identity in stratum_provider:
|
|
raise ValueError(f"{location}: duplicate stratum/provider {content_stratum}/{provider}")
|
|
stratum_provider.add(stratum_identity)
|
|
|
|
if not path.is_file():
|
|
raise ValueError(f"{location}: source file is missing: {relative_path}")
|
|
actual_sha256 = sha256_file(path)
|
|
if actual_sha256 != source_sha256:
|
|
raise ValueError(
|
|
f"{location}: source sha256 mismatch for {relative_path}: expected {source_sha256}, got {actual_sha256}"
|
|
)
|
|
with Image.open(path) as image:
|
|
actual_size = image.size
|
|
actual_format = image.format
|
|
if actual_size != (width, height):
|
|
raise ValueError(f"{location}: expected size {(width, height)}, got {actual_size}")
|
|
if actual_format != "PNG":
|
|
raise ValueError(f"{location}: expected PNG source, got {actual_format}")
|
|
|
|
fixture = ContentFixture(
|
|
pair_id=pair_id,
|
|
provider=provider,
|
|
model=_require_text(row, "model", location=location),
|
|
path=path,
|
|
width=width,
|
|
height=height,
|
|
content_stratum=content_stratum,
|
|
sha256=source_sha256,
|
|
source_payload_sha256=source_payload_sha256,
|
|
reuse_basis=reuse_basis,
|
|
prompt=_require_text(row, "prompt", location=location),
|
|
)
|
|
fixtures.append(fixture)
|
|
pairs[pair_id].append(fixture)
|
|
|
|
providers = {fixture.provider for fixture in fixtures}
|
|
if len(providers) < 2:
|
|
raise ValueError(f"{manifest}: a stratified cohort requires at least two providers")
|
|
for pair_id, pair in sorted(pairs.items()):
|
|
if {fixture.provider for fixture in pair} != providers:
|
|
raise ValueError(f"{manifest}: pair {pair_id} does not contain every provider")
|
|
if len({fixture.prompt for fixture in pair}) != 1 or len({fixture.content_stratum for fixture in pair}) != 1:
|
|
raise ValueError(f"{manifest}: pair {pair_id} does not preserve one prompt and content stratum")
|
|
return fixtures
|