deepfake-moe / backend /app /analysis.py
Devraj1990's picture
Harden calibrated evidence analysis
c00019b verified
Raw
History Blame Contribute Delete
113 kB
from __future__ import annotations
import hashlib
import json
import math
import shutil
import subprocess
import tempfile
import time
from dataclasses import dataclass, field
from functools import lru_cache
from io import BytesIO
from pathlib import Path
from typing import Any
import numpy as np
from PIL import ExifTags, Image, ImageChops, ImageFilter, ImageStat, PngImagePlugin, UnidentifiedImageError
from .config import Settings, get_settings
GENERATIVE_MARKERS = {
"stable diffusion",
"midjourney",
"dall-e",
"dalle",
"comfyui",
"automatic1111",
"invokeai",
"firefly",
"imagen",
"flux",
"fooocus",
"novelai",
}
EDITING_MARKERS = {
"photoshop",
"lightroom",
"gimp",
"affinity",
"snapseed",
"canva",
"pixlr",
}
DEFAULT_MODEL_PROFILE = {
"weight": 0.4,
"ai_threshold": 0.86,
"real_threshold": 0.2,
"note": "Conservative default detector calibration. Validate this model on the local golden set before trusting it in production.",
}
MODEL_PROFILES: dict[str, dict[str, Any]] = {
"buildborderless/CommunityForensics-DeepfakeDet-ViT": {
"weight": 1.0,
"ai_threshold": 0.90,
"real_threshold": 0.10,
"lean_ai_threshold": 0.70,
"input_size": 384,
"multi_view": True,
"ai_labels": ["label_1"],
"real_labels": ["label_0"],
"expert_group": "broad_primary",
"note": (
"Primary broad detector based on Community Forensics. It was trained across thousands of generator "
"variants, but is still conservatively thresholded for real-world false-positive control."
),
},
"Ateeqq/ai-vs-human-image-detector": {
"weight": 0.30,
"ai_threshold": 0.95,
"real_threshold": 0.08,
"lean_ai_threshold": 0.95,
"note": "Strong visual detector; aggressively down-weighted because portrait-style real photos can false-positive.",
},
"dima806/ai_vs_real_image_detection": {
"weight": 0.2,
"ai_threshold": 0.92,
"real_threshold": 0.15,
"note": "CIFAKE-lineage detector; down-weighted for real-world social, portrait, and compressed images.",
},
"jacoballessio/ai-image-detect-distilled": {
"weight": 0.45,
"ai_threshold": 0.88,
"real_threshold": 0.12,
"lean_ai_threshold": 0.60,
"note": "Distilled detector used as a counterbalance against portrait false positives.",
},
"SadraCoding/SDXL-Deepfake-Detector": {
"weight": 0.50,
"ai_threshold": 0.88,
"real_threshold": 0.18,
"portrait_only": True,
"min_portrait_score": 0.65,
"expert_group": "portrait_specialist",
"note": "Portrait-only specialist. It is skipped for non-portrait images so it cannot dominate generic scenes.",
},
}
@dataclass
class DetectorSignal:
name: str
status: str
label: str
ai_probability: float | None
manipulation_probability: float | None
confidence: str
evidence: list[str]
weight: float
details: dict[str, Any] = field(default_factory=dict)
def as_dict(self) -> dict[str, Any]:
return {
"name": self.name,
"status": self.status,
"label": self.label,
"ai_probability": self.ai_probability,
"manipulation_probability": self.manipulation_probability,
"confidence": self.confidence,
"evidence": self.evidence,
"weight": self.weight,
"details": self.details,
}
def analyze_image_bytes(
image_bytes: bytes,
*,
source_context: dict | None = None,
settings: Settings | None = None,
) -> dict[str, Any]:
settings = settings or get_settings()
started = time.perf_counter()
image = _load_image(image_bytes, settings)
metadata = extract_metadata(image, image_bytes)
hashes = compute_hashes(image, image_bytes)
c2pa = inspect_c2pa(image_bytes, metadata["format"])
forensics = compute_forensics(image, image_bytes)
detectors = [
metadata_detector(metadata, c2pa),
forensic_detector(forensics),
]
detectors.extend(huggingface_detectors(image, settings))
verdict = aggregate_verdict(detectors, metadata, c2pa, forensics)
layers = build_layers(metadata, hashes, c2pa, forensics, detectors, source_context)
analytical_layers = build_analytical_layer_breakdown(image, image_bytes, metadata, c2pa, forensics, detectors)
explainability = build_explainability(verdict, detectors, metadata, c2pa, forensics, analytical_layers)
result = {
"schema_version": "0.2.0",
"status": "completed",
"verdict": verdict,
"summary": {
"headline": _headline(verdict["label"]),
"plain_language": _plain_language(verdict),
"limitations": [
"Image authenticity cannot be proven from pixels alone.",
"Social platforms often strip metadata, so missing EXIF is not proof of AI generation.",
"No login scraping, private account access, face-search, or private identity inference was performed.",
],
},
"explainability": explainability,
"analytical_layers": analytical_layers,
"source_context": source_context or {},
"layers": layers,
"technical_appendix": {
"metadata": metadata,
"hashes": hashes,
"c2pa": c2pa,
"forensics": forensics,
"analytical_layers": analytical_layers,
"detectors": [detector.as_dict() for detector in detectors],
"runtime_ms": round((time.perf_counter() - started) * 1000, 2),
"reproducibility": {
"pipeline": (
"metadata + C2PA/provenance + perceptual hashes + compression/noise checks + "
"calibrated multi-view open-source model ensemble"
),
"primary_detector": "buildborderless/CommunityForensics-DeepfakeDet-ViT",
"score_semantics": "AI evidence score for triage; not a statistically calibrated probability of truth.",
"model_training": "No custom model was trained by this application.",
},
},
"next_steps": [
"Preserve the original file and URLs if this may become evidence.",
"Use platform reporting tools for non-consensual intimate imagery or impersonation.",
"Treat the verdict as an evidence summary, not a legal or forensic certificate.",
],
}
return result
def _load_image(image_bytes: bytes, settings: Settings) -> Image.Image:
try:
image = Image.open(BytesIO(image_bytes))
image.load()
except (UnidentifiedImageError, OSError) as exc:
raise ValueError("Uploaded content is not a readable image.") from exc
width, height = image.size
if width * height > settings.max_image_pixels:
raise ValueError("Image dimensions exceed the configured safety limit.")
return image
def extract_metadata(image: Image.Image, image_bytes: bytes) -> dict[str, Any]:
exif = {}
gps_present = False
try:
raw_exif = image.getexif()
for tag_id, value in raw_exif.items():
tag = ExifTags.TAGS.get(tag_id, str(tag_id))
if tag == "GPSInfo":
gps_present = True
exif[tag] = "[redacted: GPS metadata present]"
continue
exif[tag] = _safe_metadata_value(value)
except Exception:
exif = {}
png_text = {}
if isinstance(image, PngImagePlugin.PngImageFile):
for key, value in image.text.items():
png_text[key] = str(value)[:1000]
xmp = extract_xmp(image_bytes)
marker_text = _metadata_marker_text(exif, png_text, xmp)
generative_markers = sorted(marker for marker in GENERATIVE_MARKERS if marker in marker_text)
editing_markers = sorted(marker for marker in EDITING_MARKERS if marker in marker_text)
software_values = _collect_software_values(exif, png_text, xmp, generative_markers, editing_markers)
return {
"format": image.format or "unknown",
"width": image.width,
"height": image.height,
"mode": image.mode,
"has_exif": bool(exif),
"gps_present": gps_present,
"exif": exif,
"png_text": png_text,
"xmp_present": bool(xmp),
"xmp_excerpt": _xmp_report_summary(xmp, generative_markers, editing_markers),
"software_values": software_values,
"generative_markers": generative_markers,
"editing_markers": editing_markers,
}
def extract_xmp(image_bytes: bytes) -> str | None:
lower = image_bytes.lower()
start = lower.find(b"<x:xmpmeta")
if start == -1:
start = lower.find(b"<?xpacket")
if start == -1:
return None
end = lower.find(b"</x:xmpmeta>", start)
if end == -1:
end = lower.find(b"<?xpacket end=", start)
if end == -1:
end = min(start + 6000, len(image_bytes))
else:
end = min(end + 12, len(image_bytes))
return image_bytes[start:end].decode("utf-8", errors="replace")
def compute_hashes(image: Image.Image, image_bytes: bytes) -> dict[str, Any]:
rgb = image.convert("RGB")
return {
"sha256": hashlib.sha256(image_bytes).hexdigest(),
"average_hash": _average_hash(rgb),
"difference_hash": _difference_hash(rgb),
}
def inspect_c2pa(image_bytes: bytes, image_format: str | None) -> dict[str, Any]:
executable = shutil.which("c2patool") or shutil.which("c2pa")
if not executable:
return {
"status": "unavailable",
"claim": None,
"evidence": ["No c2patool/c2pa executable was found on PATH."],
}
suffix = f".{(image_format or 'img').lower()}"
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as handle:
handle.write(image_bytes)
temp_path = Path(handle.name)
try:
completed = subprocess.run(
[executable, str(temp_path), "--json"],
check=False,
capture_output=True,
text=True,
timeout=8,
)
if completed.returncode != 0:
return {
"status": "not_found",
"claim": None,
"evidence": ["No readable C2PA manifest was found."],
}
parsed = json.loads(completed.stdout)
text = json.dumps(parsed).lower()
generated = any(marker in text for marker in ("ai", "generated", "synthetic", "model"))
return {
"status": "found",
"claim": "ai_generated_or_synthetic" if generated else "content_credentials_present",
"manifest": parsed,
"evidence": ["C2PA/content credentials metadata was readable."],
}
except Exception as exc:
return {
"status": "error",
"claim": None,
"evidence": [f"C2PA inspection failed: {exc.__class__.__name__}."],
}
finally:
temp_path.unlink(missing_ok=True)
def compute_forensics(image: Image.Image, image_bytes: bytes) -> dict[str, Any]:
rgb = image.convert("RGB")
ela = _ela_metrics(rgb)
noise = _noise_metrics(rgb)
entropy = round(float(image.convert("L").entropy()), 4)
jpeg_markers = _jpeg_marker_summary(image_bytes)
quality = _input_quality_metrics(image, image_bytes, entropy)
manipulation_score = _clamp(
0.42 * ela["normalized_mean"]
+ 0.38 * noise["tile_inconsistency"]
+ 0.20 * jpeg_markers["double_quantization_hint"]
)
artificiality_score = _clamp(
0.35 * (1.0 - min(entropy / 8.0, 1.0))
+ 0.30 * noise["low_noise_hint"]
+ 0.35 * ela["normalized_mean"]
)
return {
"ela": ela,
"noise": noise,
"entropy": entropy,
"jpeg_markers": jpeg_markers,
"quality": quality,
"manipulation_score": round(manipulation_score, 4),
"artificiality_score": round(artificiality_score, 4),
"notes": [
"Forensic scores are heuristic signals and are weaker than a validated detector or signed provenance.",
"Cropping, screenshots, and social-media recompression can mimic manipulation artifacts.",
],
}
def build_analytical_layer_breakdown(
image: Image.Image,
image_bytes: bytes,
metadata: dict[str, Any],
c2pa: dict[str, Any],
forensics: dict[str, Any],
detectors: list[DetectorSignal],
) -> list[dict[str, Any]]:
rgb = image.convert("RGB")
gray = rgb.convert("L")
rgb_array = np.asarray(rgb, dtype=np.float32)
gray_array = np.asarray(gray, dtype=np.float32)
edges = np.asarray(gray.filter(ImageFilter.FIND_EDGES), dtype=np.float32)
luminance = _luminance_layer(gray_array)
chroma = _chroma_layer(rgb_array)
edge_geometry = _edge_geometry_layer(edges)
frequency = _frequency_layer(gray_array)
tile_anomalies = _tile_anomaly_layer(gray_array, edges)
model_layer = _model_consensus_layer(detectors)
transform_layer = _model_transform_robustness_layer(detectors)
return [
_evidence_layer(
layer_id="source_provenance",
name="Source, Metadata, And Provenance",
layer_type="container_metadata",
question="Does the file carry provenance or generation/editing metadata?",
method="EXIF/XMP/PNG text extraction plus optional C2PA manifest inspection.",
ai_signal=0.9 if metadata["generative_markers"] or c2pa.get("claim") == "ai_generated_or_synthetic" else 0.5,
manipulation_signal=0.62 if metadata["editing_markers"] else 0.25,
confidence="medium" if metadata["generative_markers"] or c2pa.get("claim") else "low",
evidence=[
f"Generative markers: {', '.join(metadata['generative_markers']) or 'none'}.",
f"Editing markers: {', '.join(metadata['editing_markers']) or 'none'}.",
f"C2PA status: {c2pa['status']}.",
"EXIF metadata present." if metadata["has_exif"] else "No EXIF metadata found.",
],
metrics={
"has_exif": metadata["has_exif"],
"c2pa_status": c2pa["status"],
"generative_marker_count": len(metadata["generative_markers"]),
"editing_marker_count": len(metadata["editing_markers"]),
},
limitations=[
"Most social platforms strip metadata.",
"Metadata can be added, removed, or forged.",
],
),
_evidence_layer(
layer_id="input_quality",
name="Input Quality And Robustness",
layer_type="robustness_check",
question="Is the file quality strong enough for high-confidence automated detection?",
method="Check resolution, file size, aspect ratio, entropy, and format conditions known to affect detector reliability.",
ai_signal=0.5,
manipulation_signal=forensics["quality"]["risk_score"],
confidence="low",
evidence=forensics["quality"]["evidence"],
metrics=forensics["quality"],
limitations=[
"Low quality does not mean fake.",
"This layer adjusts confidence; it is not an AI detector by itself.",
],
),
model_layer,
transform_layer,
_evidence_layer(
layer_id="luminance",
name="Luminance Layer",
layer_type="pixel_decomposition",
question="Do brightness/contrast statistics look unusually flat, clipped, or over-regularized?",
method="Analyze grayscale entropy, contrast, clipping, and local brightness variation.",
ai_signal=luminance["ai_signal"],
manipulation_signal=luminance["manipulation_signal"],
confidence="low",
evidence=luminance["evidence"],
metrics=luminance["metrics"],
limitations=[
"Studio lighting, compression, and screenshots can look smooth or clipped.",
"This layer is a weak supporting signal only.",
],
),
_evidence_layer(
layer_id="chroma",
name="Chroma And Color Layer",
layer_type="pixel_decomposition",
question="Do color statistics show unusually uniform or synthetic-looking saturation?",
method="Analyze RGB channel spread, channel correlation, and saturation distribution.",
ai_signal=chroma["ai_signal"],
manipulation_signal=chroma["manipulation_signal"],
confidence="low",
evidence=chroma["evidence"],
metrics=chroma["metrics"],
limitations=[
"Color grading and camera profiles can dominate this signal.",
"This does not identify which generator, if any, created the image.",
],
),
_evidence_layer(
layer_id="edge_geometry",
name="Edge And Geometry Layer",
layer_type="structural_decomposition",
question="Are edges too inconsistent, too clean, or locally unnatural?",
method="Run edge extraction and compare edge density/variance across tiles.",
ai_signal=edge_geometry["ai_signal"],
manipulation_signal=edge_geometry["manipulation_signal"],
confidence="low",
evidence=edge_geometry["evidence"],
metrics=edge_geometry["metrics"],
limitations=[
"Texture-rich real scenes and bokeh-heavy portraits can both skew edge statistics.",
"This is anomaly evidence, not identity or attribution evidence.",
],
),
_evidence_layer(
layer_id="noise_residual",
name="Noise Residual Layer",
layer_type="forensic_residual",
question="Is sensor-like noise consistent across image regions?",
method="Estimate edge/noise residual variance across a 4x4 tile grid.",
ai_signal=round(0.35 + forensics["noise"]["low_noise_hint"] * 0.45, 4),
manipulation_signal=forensics["noise"]["tile_inconsistency"],
confidence="low",
evidence=[
f"Noise tile variance mean: {forensics['noise']['tile_variance_mean']}.",
f"Noise tile inconsistency: {forensics['noise']['tile_inconsistency']}.",
f"Low-noise hint: {forensics['noise']['low_noise_hint']}.",
],
metrics=forensics["noise"],
limitations=[
"Denoising, resizing, and platform recompression can erase normal sensor noise.",
"Generated images can also contain synthetic noise.",
],
),
_evidence_layer(
layer_id="compression_ela",
name="Compression And ELA Layer",
layer_type="forensic_residual",
question="Do recompression artifacts suggest editing, screenshots, or pasted regions?",
method="Recompress image to JPEG quality 90 and measure error-level deltas plus JPEG markers.",
ai_signal=forensics["artificiality_score"],
manipulation_signal=forensics["manipulation_score"],
confidence="low",
evidence=[
f"ELA normalized mean: {forensics['ela']['normalized_mean']}.",
f"JPEG DQT marker count: {forensics['jpeg_markers']['dqt_marker_count']}.",
f"Double-quantization hint: {forensics['jpeg_markers']['double_quantization_hint']}.",
],
metrics={
"ela": forensics["ela"],
"jpeg_markers": forensics["jpeg_markers"],
"manipulation_score": forensics["manipulation_score"],
"artificiality_score": forensics["artificiality_score"],
},
limitations=[
"ELA is fragile and often reacts to normal recompression.",
"A clean ELA result does not prove authenticity.",
],
),
_evidence_layer(
layer_id="frequency",
name="Frequency Spectrum Layer",
layer_type="frequency_decomposition",
question="Does the image have unusual high-frequency or overly smooth spectral structure?",
method="Compute grayscale FFT energy distribution across low/mid/high frequency bands.",
ai_signal=frequency["ai_signal"],
manipulation_signal=frequency["manipulation_signal"],
confidence="low",
evidence=frequency["evidence"],
metrics=frequency["metrics"],
limitations=[
"Frequency patterns are affected by camera sharpening, resizing, and compression.",
"This layer is useful for consistency checks, not stand-alone classification.",
],
),
_evidence_layer(
layer_id="tile_regions",
name="Tile Region Anomaly Layer",
layer_type="spatial_decomposition",
question="Do any local regions behave differently from the rest of the image?",
method="Split the image into a 4x4 grid and compare brightness/noise/edge residual z-scores.",
ai_signal=tile_anomalies["ai_signal"],
manipulation_signal=tile_anomalies["manipulation_signal"],
confidence="low",
evidence=tile_anomalies["evidence"],
metrics=tile_anomalies["metrics"],
limitations=[
"This identifies suspicious regions, not the cause.",
"Natural subject/background boundaries often produce regional differences.",
],
),
]
def _evidence_layer(
*,
layer_id: str,
name: str,
layer_type: str,
question: str,
method: str,
ai_signal: float,
manipulation_signal: float,
confidence: str,
evidence: list[str],
metrics: dict[str, Any],
limitations: list[str],
) -> dict[str, Any]:
ai_signal = round(_clamp(ai_signal), 4)
manipulation_signal = round(_clamp(manipulation_signal), 4)
if ai_signal >= 0.72 and confidence in {"medium", "high"}:
conclusion = "supports_ai_generated"
elif ai_signal <= 0.28 and manipulation_signal < 0.45:
conclusion = "supports_real_or_camera_origin"
elif manipulation_signal >= 0.65 and confidence in {"medium", "high"}:
conclusion = "supports_manipulation_or_editing"
elif ai_signal >= 0.58 or manipulation_signal >= 0.5:
conclusion = "weak_anomaly"
else:
conclusion = "neutral_or_inconclusive"
if layer_type in {"container_metadata", "pretrained_model_inference"}:
decision_role = "primary_evidence"
elif layer_type in {"forensic_residual", "robustness_check"}:
decision_role = "guard_or_supporting_evidence"
else:
decision_role = "review_context_only"
if conclusion == "supports_ai_generated":
direction = "toward_ai_generated"
elif conclusion == "supports_real_or_camera_origin":
direction = "toward_real_origin"
elif conclusion == "supports_manipulation_or_editing":
direction = "toward_manipulation"
elif conclusion == "weak_anomaly":
direction = "weak_anomaly_only"
else:
direction = "neutral"
confidence_weight = {"high": 1.0, "medium": 0.72, "low": 0.35, "none": 0.0}.get(confidence, 0.0)
directional_distance = max(abs(ai_signal - 0.5) * 2.0, max(0.0, manipulation_signal - 0.5) * 2.0)
influence = directional_distance * confidence_weight if decision_role != "review_context_only" else 0.0
counterfactual = {
"toward_ai_generated": "A lower calibrated model/provenance score or contradictory real-origin evidence would weaken this layer.",
"toward_real_origin": "Missing camera-compatible evidence or independent synthetic provenance would weaken this layer.",
"toward_manipulation": "A cleaner residual pattern across the original file would weaken this manipulation signal.",
"weak_anomaly_only": "The anomaly must repeat across independent methods before it can materially affect the verdict.",
"neutral": "Additional original-file provenance or stronger independent model agreement would make this layer informative.",
}[direction]
return {
"id": layer_id,
"name": name,
"type": layer_type,
"question": question,
"method": method,
"conclusion": conclusion,
"ai_signal": ai_signal,
"manipulation_signal": manipulation_signal,
"confidence": confidence,
"reliability": confidence,
"decision_role": decision_role,
"direction": direction,
"influence": round(_clamp(influence), 4),
"used_by_arbiter": decision_role != "review_context_only",
"counterfactual": counterfactual,
"evidence": evidence,
"metrics": metrics,
"limitations": limitations,
}
def _luminance_layer(gray_array: np.ndarray) -> dict[str, Any]:
normalized = gray_array / 255.0
entropy = _array_entropy(gray_array)
contrast = float(np.std(normalized))
clipped_dark = float(np.mean(gray_array <= 3))
clipped_bright = float(np.mean(gray_array >= 252))
tile_means = _tile_feature_values(gray_array, lambda tile: float(np.mean(tile) / 255.0))
local_variation = float(np.std(tile_means)) if tile_means else 0.0
flatness_hint = _clamp((0.12 - contrast) / 0.12)
clipping_hint = _clamp((clipped_dark + clipped_bright - 0.04) / 0.16)
uneven_lighting_hint = _clamp((local_variation - 0.24) / 0.36)
ai_signal = _clamp(0.35 + 0.25 * flatness_hint + 0.20 * clipping_hint + 0.20 * (1 - min(entropy / 8.0, 1.0)))
evidence = [
f"Grayscale entropy: {entropy:.3f}.",
f"Luminance contrast standard deviation: {contrast:.3f}.",
f"Clipped dark/bright pixels: {(clipped_dark + clipped_bright):.2%}.",
f"Local brightness variation across tiles: {local_variation:.3f}.",
]
return {
"ai_signal": ai_signal,
"manipulation_signal": _clamp(0.25 + clipping_hint * 0.35 + uneven_lighting_hint * 0.25),
"evidence": evidence,
"metrics": {
"entropy": round(entropy, 4),
"contrast_std": round(contrast, 4),
"clipped_dark_ratio": round(clipped_dark, 4),
"clipped_bright_ratio": round(clipped_bright, 4),
"tile_brightness_std": round(local_variation, 4),
"uneven_lighting_hint": round(uneven_lighting_hint, 4),
},
}
def _chroma_layer(rgb_array: np.ndarray) -> dict[str, Any]:
normalized = rgb_array / 255.0
channel_means = np.mean(normalized, axis=(0, 1))
channel_stds = np.std(normalized, axis=(0, 1))
max_channel = np.max(normalized, axis=2)
min_channel = np.min(normalized, axis=2)
saturation = np.where(max_channel == 0, 0, (max_channel - min_channel) / (max_channel + 1e-6))
saturation_mean = float(np.mean(saturation))
saturation_std = float(np.std(saturation))
channel_balance = float(np.std(channel_means))
uniform_color_hint = _clamp((0.12 - saturation_std) / 0.12)
oversaturation_hint = _clamp((saturation_mean - 0.62) / 0.28)
ai_signal = _clamp(0.35 + 0.25 * uniform_color_hint + 0.20 * oversaturation_hint + 0.10 * _clamp(channel_balance / 0.2))
evidence = [
f"Mean saturation: {saturation_mean:.3f}.",
f"Saturation variation: {saturation_std:.3f}.",
f"RGB channel mean balance spread: {channel_balance:.3f}.",
]
return {
"ai_signal": ai_signal,
"manipulation_signal": _clamp(0.25 + 0.35 * oversaturation_hint + 0.25 * uniform_color_hint),
"evidence": evidence,
"metrics": {
"rgb_channel_means": [round(float(value), 4) for value in channel_means],
"rgb_channel_stds": [round(float(value), 4) for value in channel_stds],
"saturation_mean": round(saturation_mean, 4),
"saturation_std": round(saturation_std, 4),
"channel_balance_spread": round(channel_balance, 4),
},
}
def _edge_geometry_layer(edges: np.ndarray) -> dict[str, Any]:
normalized = edges / 255.0
edge_density = float(np.mean(normalized > 0.12))
edge_strength = float(np.mean(normalized))
tile_edges = _tile_feature_values(edges, lambda tile: float(np.mean(tile) / 255.0))
tile_std = float(np.std(tile_edges)) if tile_edges else 0.0
too_clean_hint = _clamp((0.045 - edge_density) / 0.045)
inconsistency_hint = _clamp(tile_std / 0.18)
ai_signal = _clamp(0.35 + 0.25 * too_clean_hint + 0.25 * inconsistency_hint)
evidence = [
f"Edge density: {edge_density:.3f}.",
f"Average edge strength: {edge_strength:.3f}.",
f"Edge variation across tiles: {tile_std:.3f}.",
]
return {
"ai_signal": ai_signal,
"manipulation_signal": _clamp(0.25 + 0.5 * inconsistency_hint),
"evidence": evidence,
"metrics": {
"edge_density": round(edge_density, 4),
"edge_strength_mean": round(edge_strength, 4),
"tile_edge_std": round(tile_std, 4),
},
}
def _frequency_layer(gray_array: np.ndarray) -> dict[str, Any]:
small = np.asarray(Image.fromarray(gray_array.astype(np.uint8)).resize((256, 256), Image.Resampling.BILINEAR), dtype=np.float32)
small = small - float(np.mean(small))
spectrum = np.abs(np.fft.fftshift(np.fft.fft2(small)))
energy = spectrum**2
height, width = energy.shape
y, x = np.ogrid[:height, :width]
center_y = (height - 1) / 2.0
center_x = (width - 1) / 2.0
radius = np.sqrt((y - center_y) ** 2 + (x - center_x) ** 2)
max_radius = float(np.max(radius)) or 1.0
low = energy[radius <= max_radius * 0.18].sum()
mid = energy[(radius > max_radius * 0.18) & (radius <= max_radius * 0.45)].sum()
high = energy[radius > max_radius * 0.45].sum()
total = float(low + mid + high + 1e-6)
low_ratio = float(low / total)
mid_ratio = float(mid / total)
high_ratio = float(high / total)
smooth_hint = _clamp((0.07 - high_ratio) / 0.07)
noisy_hint = _clamp((high_ratio - 0.38) / 0.22)
ai_signal = _clamp(0.35 + 0.3 * smooth_hint + 0.15 * noisy_hint)
evidence = [
f"Low-frequency energy ratio: {low_ratio:.3f}.",
f"Mid-frequency energy ratio: {mid_ratio:.3f}.",
f"High-frequency energy ratio: {high_ratio:.3f}.",
]
return {
"ai_signal": ai_signal,
"manipulation_signal": _clamp(0.25 + 0.35 * noisy_hint + 0.2 * smooth_hint),
"evidence": evidence,
"metrics": {
"low_frequency_ratio": round(low_ratio, 4),
"mid_frequency_ratio": round(mid_ratio, 4),
"high_frequency_ratio": round(high_ratio, 4),
"smooth_frequency_hint": round(smooth_hint, 4),
"noisy_frequency_hint": round(noisy_hint, 4),
},
}
def _tile_anomaly_layer(gray_array: np.ndarray, edges: np.ndarray) -> dict[str, Any]:
tile_rows = _tile_stats(gray_array, edges)
if not tile_rows:
return {
"ai_signal": 0.5,
"manipulation_signal": 0.25,
"evidence": ["Tile analysis was not available."],
"metrics": {"tiles": []},
}
brightness = np.array([tile["brightness"] for tile in tile_rows], dtype=np.float32)
noise = np.array([tile["noise"] for tile in tile_rows], dtype=np.float32)
edge = np.array([tile["edge"] for tile in tile_rows], dtype=np.float32)
anomaly_scores = _z_scores(brightness) + _z_scores(noise) + _z_scores(edge)
for tile, score in zip(tile_rows, anomaly_scores, strict=False):
tile["anomaly_score"] = round(float(score), 4)
max_score = float(np.max(anomaly_scores)) if len(anomaly_scores) else 0.0
mean_score = float(np.mean(anomaly_scores)) if len(anomaly_scores) else 0.0
for tile in tile_rows:
severity = float(tile["anomaly_score"]) / (max_score + 1e-6) if max_score else 0.0
tile["severity"] = round(_clamp(severity), 4)
tile["severity_band"] = _severity_band(severity)
top_tiles = sorted(tile_rows, key=lambda item: item["anomaly_score"], reverse=True)[:4]
high_tiles = [tile for tile in tile_rows if tile["severity_band"] == "high"]
anomaly_hint = _clamp((max_score - 3.0) / 5.0)
evidence = [
f"Strongest tile anomaly score: {max_score:.3f}.",
f"Mean tile anomaly score: {mean_score:.3f}.",
f"Region map: {len(tile_rows)} tiles analyzed; {len(high_tiles)} high-severity regional differences.",
"Top anomalous tiles: "
+ ", ".join(f"row {tile['row']} col {tile['col']} score {tile['anomaly_score']}" for tile in top_tiles)
+ ".",
]
return {
"ai_signal": _clamp(0.35 + 0.25 * anomaly_hint),
"manipulation_signal": _clamp(0.25 + 0.55 * anomaly_hint),
"evidence": evidence,
"metrics": {
"max_tile_anomaly_score": round(max_score, 4),
"mean_tile_anomaly_score": round(mean_score, 4),
"high_severity_tile_count": len(high_tiles),
"tile_grid": sorted(tile_rows, key=lambda item: (item["row"], item["col"])),
"top_tiles": top_tiles,
},
}
def _model_consensus_layer(detectors: list[DetectorSignal]) -> dict[str, Any]:
model_signals = [
detector for detector in detectors if detector.name.startswith("hf:") and detector.status == "ok" and detector.ai_probability is not None
]
ensemble = next((detector for detector in detectors if detector.name == "open_source_model_ensemble" and detector.status == "ok"), None)
values = [float(detector.ai_probability) for detector in model_signals if detector.ai_probability is not None]
if not values:
return _evidence_layer(
layer_id="visual_model_consensus",
name="Visual Model Consensus Layer",
layer_type="pretrained_model_inference",
question="Do pretrained visual detectors classify the image as AI-generated?",
method="Run configured visual detectors, apply model-specific abstention bands, and aggregate calibrated votes.",
ai_signal=0.5,
manipulation_signal=0.25,
confidence="none",
evidence=["No visual model scores were available."],
metrics={"enabled_models": 0},
limitations=[
"Model inference must be enabled with AIDA_ENABLE_HF_MODEL=true.",
"No model score means the final verdict relies on non-model evidence only.",
],
)
ai_votes = sum(detector.label == "model_likely_ai_generated" for detector in model_signals)
real_votes = sum(detector.label == "model_likely_human_or_real" for detector in model_signals)
calibrated_values = [_calibrated_model_stance(detector) for detector in model_signals]
raw_disagreement = float(max(values) - min(values)) if len(values) >= 2 else 0.0
stance_disagreement = float(max(calibrated_values) - min(calibrated_values)) if len(calibrated_values) >= 2 else 0.0
confidence = ensemble.confidence if ensemble else ("medium" if stance_disagreement < 0.35 else "low")
average = float(ensemble.ai_probability) if ensemble and ensemble.ai_probability is not None else _weighted_average(
values,
[float(detector.weight or DEFAULT_MODEL_PROFILE["weight"]) for detector in model_signals],
)
return _evidence_layer(
layer_id="visual_model_consensus",
name="Visual Model Consensus Layer",
layer_type="pretrained_model_inference",
question="Do pretrained visual detectors classify the image as AI-generated?",
method="Run configured visual detectors, apply model-specific abstention bands, and aggregate calibrated votes.",
ai_signal=average,
manipulation_signal=0.25,
confidence=confidence,
evidence=[
f"{ai_votes}/{len(values)} models voted AI-generated.",
f"{real_votes}/{len(values)} models voted real/human-origin.",
f"Calibrated stance disagreement: {stance_disagreement:.3f}.",
f"Raw-score range (diagnostic only): {raw_disagreement:.3f}.",
f"Calibrated model evidence score: {average:.3f}.",
*[detector.evidence[0] for detector in model_signals if detector.evidence],
],
metrics={
"enabled_models": len(values),
"ai_votes": ai_votes,
"real_votes": real_votes,
"inconclusive_votes": len(values) - ai_votes - real_votes,
"average_ai_probability": round(average, 4),
"score_semantics": "calibrated evidence score",
"raw_average_ai_probability": round(float(np.mean(values)), 4),
"calibrated_stance_disagreement": round(stance_disagreement, 4),
"raw_score_range": round(raw_disagreement, 4),
"model_scores": [
{
"name": detector.name,
"ai_probability": detector.ai_probability,
"label": detector.label,
"reliability_weight": detector.weight,
"details": detector.details,
}
for detector in model_signals
],
},
limitations=[
"Pretrained detectors can fail on new generators, screenshots, crops, or heavy compression.",
"This layer is stronger when multiple models agree and weaker when they disagree.",
],
)
def _model_transform_robustness_layer(detectors: list[DetectorSignal]) -> dict[str, Any]:
primary = next(
(
detector
for detector in detectors
if detector.name.startswith("hf:")
and detector.status == "ok"
and _model_profile(detector.name.removeprefix("hf:")).get("expert_group") == "broad_primary"
),
None,
)
details = primary.details if primary else {}
view_scores = details.get("view_scores") or []
if not primary or not view_scores:
return _evidence_layer(
layer_id="model_transform_robustness",
name="Model Transform Robustness Layer",
layer_type="robustness_check",
question="Does the broad detector preserve its stance under benign image transformations?",
method="Compare the primary model across original, crop, JPEG, mirror, and social-resize views.",
ai_signal=0.5,
manipulation_signal=0.25,
confidence="none",
evidence=["Multi-view primary-model diagnostics were unavailable."],
metrics={"view_count": 0, "view_scores": []},
limitations=[
"This layer requires the configured broad primary model.",
"Transform stability improves confidence but cannot prove authenticity.",
],
)
stable_enough = bool(details.get("stable_enough"))
ai_signal = float(primary.ai_probability or 0.5) if stable_enough else 0.5
confidence = "medium" if stable_enough and len(view_scores) >= 5 else "low"
return _evidence_layer(
layer_id="model_transform_robustness",
name="Model Transform Robustness Layer",
layer_type="robustness_check",
question="Does the broad detector preserve its stance under benign image transformations?",
method="Compare the primary model across original, 92% crop, JPEG-85, horizontal flip, and 75% social-resize views.",
ai_signal=ai_signal,
manipulation_signal=0.25,
confidence=confidence,
evidence=[
f"Primary view median AI score: {float(details.get('median_ai_score') or 0.5):.3f}.",
(
f"Stability range {float(details.get('stability_range') or 0):.3f}; median absolute deviation "
f"{float(details.get('median_absolute_deviation') or 0):.3f}; interquartile range "
f"{float(details.get('interquartile_range') or 0):.3f}."
),
(
f"Consistent view requirement: {details.get('required_consistent_views', 0)} of {len(view_scores)}; "
f"stability gate {'passed' if stable_enough else 'did not pass'}."
),
],
metrics=details,
limitations=[
"Benign transformations test robustness, not the semantic truth of the image.",
"A consistently wrong model can still appear stable, so this layer never acts alone.",
],
)
def _array_entropy(values: np.ndarray) -> float:
histogram, _ = np.histogram(values.astype(np.uint8), bins=256, range=(0, 255), density=False)
probabilities = histogram.astype(np.float64)
probabilities = probabilities / (probabilities.sum() + 1e-9)
probabilities = probabilities[probabilities > 0]
return float(-np.sum(probabilities * np.log2(probabilities)))
def _tile_feature_values(values: np.ndarray, reducer) -> list[float]:
rows = []
height, width = values.shape[:2]
y_edges = np.linspace(0, height, 5, dtype=int)
x_edges = np.linspace(0, width, 5, dtype=int)
for row_index in range(4):
for col_index in range(4):
top, bottom = int(y_edges[row_index]), int(y_edges[row_index + 1])
left, right = int(x_edges[col_index]), int(x_edges[col_index + 1])
tile = values[top:bottom, left:right]
if tile.size:
rows.append(float(reducer(tile)))
return rows
def _tile_stats(gray_array: np.ndarray, edges: np.ndarray) -> list[dict[str, Any]]:
rows = []
height, width = gray_array.shape[:2]
y_edges = np.linspace(0, height, 5, dtype=int)
x_edges = np.linspace(0, width, 5, dtype=int)
for row_index in range(4):
for col_index in range(4):
top, bottom = int(y_edges[row_index]), int(y_edges[row_index + 1])
left, right = int(x_edges[col_index]), int(x_edges[col_index + 1])
gray_tile = gray_array[top:bottom, left:right]
edge_tile = edges[top:bottom, left:right]
if gray_tile.size and edge_tile.size:
rows.append(
{
"row": row_index + 1,
"col": col_index + 1,
"brightness": round(float(np.mean(gray_tile) / 255.0), 4),
"noise": round(float(np.var(edge_tile)), 4),
"edge": round(float(np.mean(edge_tile) / 255.0), 4),
}
)
return rows
def _severity_band(value: float) -> str:
if value >= 0.75:
return "high"
if value >= 0.45:
return "medium"
return "low"
def _z_scores(values: np.ndarray) -> np.ndarray:
if len(values) == 0:
return values
return np.abs((values - float(np.mean(values))) / (float(np.std(values)) + 1e-6))
def metadata_detector(metadata: dict[str, Any], c2pa: dict[str, Any]) -> DetectorSignal:
evidence: list[str] = []
ai_probability = 0.5
manipulation_probability = 0.25
label = "inconclusive"
confidence = "low"
if metadata["generative_markers"]:
ai_probability = 0.9
label = "likely_ai_generated"
confidence = "medium"
evidence.append(f"Generative software markers found: {', '.join(metadata['generative_markers'])}.")
elif c2pa.get("claim") == "ai_generated_or_synthetic":
ai_probability = 0.92
label = "likely_ai_generated"
confidence = "high"
evidence.append("C2PA/content credentials appear to declare synthetic or AI-generated content.")
elif metadata["editing_markers"]:
manipulation_probability = 0.62
label = "possibly_edited"
confidence = "low"
evidence.append(f"Editing software markers found: {', '.join(metadata['editing_markers'])}.")
elif metadata["has_exif"]:
ai_probability = 0.35
label = "camera_metadata_present"
confidence = "low"
evidence.append("Some EXIF metadata is present; this is compatible with camera-origin media but not proof.")
else:
evidence.append("No strong metadata signal was found.")
return DetectorSignal(
name="metadata_provenance",
status="ok",
label=label,
ai_probability=ai_probability,
manipulation_probability=manipulation_probability,
confidence=confidence,
evidence=evidence,
weight=0.22,
)
def forensic_detector(forensics: dict[str, Any]) -> DetectorSignal:
manipulation = forensics["manipulation_score"]
artificiality = forensics["artificiality_score"]
evidence = [
f"ELA normalized mean: {forensics['ela']['normalized_mean']}.",
f"Noise tile inconsistency: {forensics['noise']['tile_inconsistency']}.",
]
if manipulation >= 0.68:
label = "likely_manipulated_or_recompressed"
confidence = "medium"
elif artificiality >= 0.7:
label = "synthetic_artifact_signal"
confidence = "low"
else:
label = "no_strong_forensic_signal"
confidence = "low"
return DetectorSignal(
name="compression_noise_forensics",
status="ok",
label=label,
ai_probability=round(0.35 + artificiality * 0.45, 4),
manipulation_probability=round(manipulation, 4),
confidence=confidence,
evidence=evidence,
weight=0.28,
)
def huggingface_detectors(image: Image.Image, settings: Settings) -> list[DetectorSignal]:
if not settings.enable_hf_model:
return [
DetectorSignal(
name="open_source_model_ensemble",
status="unavailable",
label="not_configured",
ai_probability=None,
manipulation_probability=None,
confidence="none",
evidence=["Set AIDA_ENABLE_HF_MODEL=true and AIDA_HF_MODEL_IDS to use pretrained open-source visual detectors."],
weight=0.65,
)
]
model_ids = _configured_model_ids(settings)
portrait_metrics = _portrait_likelihood_metrics(image)
signals = [_huggingface_detector_for_model(image, model_id, portrait_metrics) for model_id in model_ids]
ok_signals = [signal for signal in signals if signal.status == "ok" and signal.ai_probability is not None]
if len(ok_signals) >= 2:
signals.append(model_ensemble_signal(ok_signals))
return signals
def optional_huggingface_detector(image: Image.Image, settings: Settings) -> DetectorSignal:
"""Backward-compatible single-signal adapter used by older tests/callers."""
return huggingface_detectors(image, settings)[0]
def _huggingface_detector_for_model(image: Image.Image, model_id: str, portrait_metrics: dict[str, Any] | None = None) -> DetectorSignal:
name = f"hf:{model_id}"
profile = _model_profile(model_id)
portrait_metrics = portrait_metrics or {"portrait_score": 0.0, "evidence": ["Portrait gate was not evaluated."]}
if profile.get("portrait_only") and float(portrait_metrics.get("portrait_score") or 0.0) < float(profile.get("min_portrait_score") or 0.0):
return DetectorSignal(
name=name,
status="unavailable",
label="portrait_gate_not_met",
ai_probability=None,
manipulation_probability=None,
confidence="none",
evidence=[
f"Skipped portrait-only model {model_id}; portrait likelihood was {float(portrait_metrics.get('portrait_score') or 0.0):.2f}.",
*portrait_metrics.get("evidence", [])[:2],
],
weight=0.0,
)
try:
classifier = _get_huggingface_classifier(model_id)
view_scores: list[tuple[str, float]] = []
output_summaries: list[str] = []
for view_name, view_image in _model_inference_views(image, profile):
outputs = _normalize_classifier_outputs(classifier(view_image, top_k=None))
view_scores.append((view_name, _map_classifier_outputs_to_ai_probability(outputs, profile)))
output_summaries.append(f"{view_name}: {_summarize_labels(outputs)}")
except Exception as exc:
return DetectorSignal(
name=name,
status="unavailable",
label="runtime_error",
ai_probability=None,
manipulation_probability=None,
confidence="none",
evidence=[f"Open-source model {model_id} failed: {exc.__class__.__name__}."],
weight=0.0,
)
scores = [score for _, score in view_scores]
ai_score = float(np.median(scores))
stability_range = (max(scores) - min(scores)) if len(scores) >= 2 else 0.0
median_absolute_deviation = float(np.median(np.abs(np.asarray(scores) - ai_score))) if len(scores) >= 2 else 0.0
interquartile_range = float(np.percentile(scores, 75) - np.percentile(scores, 25)) if len(scores) >= 4 else stability_range
required_views = max(1, math.ceil(len(scores) * 2 / 3))
ai_view_votes = sum(score >= float(profile["ai_threshold"]) for score in scores)
real_view_votes = sum(score <= float(profile["real_threshold"]) for score in scores)
stable_enough = (
stability_range <= float(profile.get("max_view_range", 0.28))
and median_absolute_deviation <= float(profile.get("max_view_mad", 0.09))
and interquartile_range <= float(profile.get("max_view_iqr", 0.18))
)
if ai_score >= profile["ai_threshold"] and ai_view_votes >= required_views and stable_enough:
label = "model_likely_ai_generated"
elif ai_score <= profile["real_threshold"] and real_view_votes >= required_views and stable_enough:
label = "model_likely_human_or_real"
else:
label = "model_inconclusive"
confidence = (
"medium"
if label != "model_inconclusive" and stability_range <= 0.16 and median_absolute_deviation <= 0.05
else "low"
)
return DetectorSignal(
name=name,
status="ok",
label=label,
ai_probability=round(ai_score, 4),
manipulation_probability=None,
confidence=confidence,
evidence=[
f"Model {model_id} multi-view median AI score: {ai_score:.3f}.",
f"Inference views: {'; '.join(output_summaries)}.",
(
f"Transform stability: range {stability_range:.3f}, median absolute deviation "
f"{median_absolute_deviation:.3f}, interquartile range {interquartile_range:.3f}; "
f"AI-supporting views {ai_view_votes}/{len(scores)}; real-supporting views {real_view_votes}/{len(scores)}."
),
(
f"Calibration: AI threshold {profile['ai_threshold']:.2f}, real threshold "
f"{profile['real_threshold']:.2f}, reliability weight {profile['weight']:.2f}."
),
*(
[f"Portrait gate active: likelihood {float(portrait_metrics.get('portrait_score') or 0.0):.2f}."]
if profile.get("portrait_only")
else []
),
],
weight=float(profile["weight"]),
details={
"model_id": model_id,
"expert_group": profile.get("expert_group", "counter_model"),
"score_semantics": "model-specific AI score; interpreted only through this model's calibration bands",
"view_scores": [{"view": view_name, "ai_score": round(score, 4)} for view_name, score in view_scores],
"view_count": len(view_scores),
"median_ai_score": round(ai_score, 4),
"stability_range": round(stability_range, 4),
"median_absolute_deviation": round(median_absolute_deviation, 4),
"interquartile_range": round(interquartile_range, 4),
"stable_enough": stable_enough,
"required_consistent_views": required_views,
"ai_supporting_views": ai_view_votes,
"real_supporting_views": real_view_votes,
"ai_threshold": float(profile["ai_threshold"]),
"real_threshold": float(profile["real_threshold"]),
"lean_ai_threshold": float(profile.get("lean_ai_threshold", 0.6)),
"reliability_weight": float(profile["weight"]),
"calibration_note": profile.get("note", ""),
},
)
def model_ensemble_signal(signals: list[DetectorSignal]) -> DetectorSignal:
valid_signals = [signal for signal in signals if signal.ai_probability is not None]
values = [float(signal.ai_probability) for signal in valid_signals]
if not values:
return DetectorSignal(
name="open_source_model_ensemble",
status="unavailable",
label="no_model_scores",
ai_probability=None,
manipulation_probability=None,
confidence="none",
evidence=["No model scores were available for ensemble aggregation."],
weight=0.0,
)
weights = [float(signal.weight or DEFAULT_MODEL_PROFILE["weight"]) for signal in valid_signals]
calibrated_values = [_calibrated_model_stance(signal) for signal in valid_signals]
average = _weighted_average(calibrated_values, weights)
raw_average = float(np.mean(values))
raw_disagreement = float(max(values) - min(values)) if len(values) >= 2 else 0.0
stance_disagreement = float(max(calibrated_values) - min(calibrated_values)) if len(calibrated_values) >= 2 else 0.0
ai_votes = sum(signal.label == "model_likely_ai_generated" for signal in valid_signals)
lean_ai_votes = sum(
value >= float(_model_profile(signal.name.removeprefix("hf:")).get("lean_ai_threshold", 0.6))
for signal, value in zip(valid_signals, values, strict=False)
)
real_votes = sum(signal.label == "model_likely_human_or_real" for signal in valid_signals)
min_score = min(values)
primary_anchor = any(
_model_profile(signal.name.removeprefix("hf:")).get("expert_group") == "broad_primary"
and float(signal.ai_probability or 0.0)
>= float(_model_profile(signal.name.removeprefix("hf:")).get("lean_ai_threshold", 0.7))
for signal in valid_signals
)
aligned_primary_consensus = (
len(values) >= 3
and primary_anchor
and lean_ai_votes == len(values)
and real_votes == 0
and raw_average >= 0.74
and stance_disagreement <= 0.50
)
if real_votes > 0 and ai_votes > 0:
label = "ensemble_inconclusive"
elif len(values) >= 3 and ai_votes == len(values) and min_score >= 0.86 and raw_average >= 0.88 and stance_disagreement <= 0.25:
label = "ensemble_likely_ai_generated"
elif aligned_primary_consensus:
label = "ensemble_likely_ai_generated"
elif len(values) >= 4 and lean_ai_votes == len(values) and ai_votes >= len(values) - 1 and real_votes == 0 and min_score >= 0.78 and raw_average >= 0.86 and stance_disagreement <= 0.32:
label = "ensemble_likely_ai_generated"
elif real_votes == len(values) and average <= 0.28:
label = "ensemble_likely_real"
elif real_votes > ai_votes and ai_votes == 0 and average <= 0.35:
label = "ensemble_likely_real"
else:
label = "ensemble_inconclusive"
if label == "ensemble_likely_ai_generated" and stance_disagreement <= 0.18 and len(values) >= 3:
confidence = "high"
elif label == "ensemble_likely_ai_generated" and stance_disagreement <= 0.50 and real_votes == 0:
confidence = "medium"
else:
confidence = "low"
evidence = [
f"{ai_votes}/{len(values)} visual models voted AI-generated.",
f"{lean_ai_votes}/{len(values)} visual models leaned AI-generated at or above 60%.",
f"{real_votes}/{len(values)} visual models voted real/human.",
f"Broad-primary anchored alignment: {'yes' if aligned_primary_consensus else 'no'}.",
f"Calibrated stance disagreement: {stance_disagreement:.3f}; raw-score range retained as a diagnostic: {raw_disagreement:.3f}.",
f"Calibrated visual evidence score: {average:.3f}; raw model average: {raw_average:.3f}; weakest raw model score: {min_score:.3f}.",
"Raw detector scores were converted to model-specific stances before voting; values inside a model's abstention band contribute neutral evidence.",
]
return DetectorSignal(
name="open_source_model_ensemble",
status="ok",
label=label,
ai_probability=round(max(average, 0.74) if aligned_primary_consensus else average, 4),
manipulation_probability=None,
confidence=confidence,
evidence=evidence,
weight=0.9,
details={
"aggregation": "reliability-weighted calibrated stances",
"raw_score_average": round(raw_average, 4),
"raw_score_range": round(raw_disagreement, 4),
"calibrated_stance_average": round(average, 4),
"calibrated_stance_range": round(stance_disagreement, 4),
"ai_votes": ai_votes,
"lean_ai_votes": lean_ai_votes,
"real_votes": real_votes,
"abstain_votes": len(values) - ai_votes - real_votes,
"primary_anchored_alignment": aligned_primary_consensus,
"policy_note": "Raw outputs from different classifiers are not treated as directly comparable probabilities.",
},
)
@lru_cache(maxsize=6)
def _get_huggingface_classifier(model_id: str):
from transformers import AutoImageProcessor, pipeline # type: ignore
profile = _model_profile(model_id)
input_size = profile.get("input_size")
if input_size:
processor = AutoImageProcessor.from_pretrained(
model_id,
size={"height": int(input_size), "width": int(input_size)},
crop_size={"height": int(input_size), "width": int(input_size)},
)
return pipeline("image-classification", model=model_id, image_processor=processor, device=-1)
return pipeline("image-classification", model=model_id, device=-1)
def _configured_model_ids(settings: Settings) -> list[str]:
configured = settings.hf_model_ids or settings.hf_model_id
model_ids = [item.strip() for item in configured.split(",") if item.strip()]
return model_ids or [settings.hf_model_id]
def _model_profile(model_id: str) -> dict[str, Any]:
return MODEL_PROFILES.get(model_id, DEFAULT_MODEL_PROFILE)
def _model_inference_views(image: Image.Image, profile: dict[str, Any]) -> list[tuple[str, Image.Image]]:
rgb = image.convert("RGB")
if not profile.get("multi_view"):
return [("original", rgb)]
width, height = rgb.size
inset_x = max(1, int(width * 0.04))
inset_y = max(1, int(height * 0.04))
center_crop = rgb.crop((inset_x, inset_y, width - inset_x, height - inset_y))
recompressed = BytesIO()
rgb.save(recompressed, format="JPEG", quality=85, optimize=True)
recompressed.seek(0)
jpeg_view = Image.open(recompressed).convert("RGB")
jpeg_view.load()
mirrored = rgb.transpose(Image.Transpose.FLIP_LEFT_RIGHT)
resized_width = max(64, int(width * 0.75))
resized_height = max(64, int(height * 0.75))
social_resize = rgb.resize((resized_width, resized_height), Image.Resampling.LANCZOS)
return [
("original", rgb),
("center_crop_92pct", center_crop),
("jpeg_quality_85", jpeg_view),
("horizontal_flip", mirrored),
("social_resize_75pct", social_resize),
]
def _calibrated_model_stance(signal: DetectorSignal) -> float:
if signal.ai_probability is None:
return 0.5
model_id = signal.name.removeprefix("hf:")
profile = _model_profile(model_id)
score = float(signal.ai_probability)
ai_threshold = float(profile.get("ai_threshold", DEFAULT_MODEL_PROFILE["ai_threshold"]))
real_threshold = float(profile.get("real_threshold", DEFAULT_MODEL_PROFILE["real_threshold"]))
if score >= ai_threshold:
return _clamp(0.75 + 0.25 * ((score - ai_threshold) / max(1e-6, 1.0 - ai_threshold)))
if score <= real_threshold:
return _clamp(0.25 * (score / max(1e-6, real_threshold)))
return 0.5
def _is_portrait_specialist(signal: DetectorSignal) -> bool:
if not signal.name.startswith("hf:"):
return False
model_id = signal.name.removeprefix("hf:")
return _model_profile(model_id).get("expert_group") == "portrait_specialist"
def _weighted_average(values: list[float], weights: list[float]) -> float:
if not values:
return 0.5
total_weight = sum(max(0.0, weight) for weight in weights)
if total_weight <= 0:
return float(np.mean(values))
return float(sum(value * max(0.0, weight) for value, weight in zip(values, weights, strict=False)) / total_weight)
def _camera_like_real_prior(metadata: dict[str, Any], forensics: dict[str, Any]) -> bool:
image_format = str(metadata.get("format") or "").upper()
noise = forensics.get("noise") or {}
entropy = float(forensics.get("entropy") or 0.0)
tile_variance = float(noise.get("tile_variance_mean") or 0.0)
low_noise_hint = float(noise.get("low_noise_hint") or 0.0)
camera_container = image_format in {"JPEG", "JPG", "MPO"} or bool(metadata.get("has_exif"))
natural_residuals = entropy >= 5.0 and tile_variance >= 70.0 and low_noise_hint <= 0.75
weak_artifacts = float(forensics.get("artificiality_score") or 0.0) < 0.45 and float(forensics.get("manipulation_score") or 0.0) < 0.65
return camera_container and natural_residuals and weak_artifacts
def aggregate_verdict(
detectors: list[DetectorSignal],
metadata: dict[str, Any],
c2pa: dict[str, Any],
forensics: dict[str, Any],
) -> dict[str, Any]:
if metadata["generative_markers"]:
return {
"label": "likely_ai_generated",
"confidence": "medium",
"ai_probability": 0.9,
"manipulation_probability": max(0.35, forensics["manipulation_score"]),
"disagreement": 0.0,
"rationale": [f"Generative software markers found: {', '.join(metadata['generative_markers'])}."],
"score_label": "AI evidence score",
"score_interpretation": "Evidence strength for triage, not a calibrated probability of truth.",
}
if c2pa.get("claim") == "ai_generated_or_synthetic":
return {
"label": "likely_ai_generated",
"confidence": "high",
"ai_probability": 0.94,
"manipulation_probability": max(0.45, forensics["manipulation_score"]),
"disagreement": 0.0,
"rationale": ["Readable content credentials appear to declare synthetic or AI-generated content."],
"score_label": "AI evidence score",
"score_interpretation": "Evidence strength for triage, not a calibrated probability of truth.",
}
model_signal = next(
(detector for detector in detectors if detector.name == "open_source_model_ensemble" and detector.status == "ok"),
None,
)
model_detectors = [
detector for detector in detectors if detector.name.startswith("hf:") and detector.status == "ok" and detector.ai_probability is not None
]
model_values = [float(detector.ai_probability) for detector in model_detectors]
calibrated_model_values = [_calibrated_model_stance(detector) for detector in model_detectors]
raw_model_score_range = (max(model_values) - min(model_values)) if len(model_values) >= 2 else 0.0
model_disagreement = (
max(calibrated_model_values) - min(calibrated_model_values)
if len(calibrated_model_values) >= 2
else 0.0
)
model_ai_votes = sum(detector.label == "model_likely_ai_generated" for detector in model_detectors)
model_lean_ai_votes = sum(
float(detector.ai_probability or 0.0)
>= float(_model_profile(detector.name.removeprefix("hf:")).get("lean_ai_threshold", 0.6))
for detector in model_detectors
)
model_real_votes = sum(detector.label == "model_likely_human_or_real" for detector in model_detectors)
quality_risk = float((forensics.get("quality") or {}).get("risk_score") or 0.0)
model_min_score = min(model_values) if model_values else 0.0
camera_like_real_prior = _camera_like_real_prior(metadata, forensics)
if model_signal and model_signal.ai_probability is not None:
available = [
detector
for detector in detectors
if detector.status == "ok" and detector.ai_probability is not None and not detector.name.startswith("hf:")
]
else:
available = [detector for detector in detectors if detector.status == "ok" and detector.ai_probability is not None]
total_weight = sum(detector.weight for detector in available) or 1.0
ai_probability = sum((detector.ai_probability or 0.5) * detector.weight for detector in available) / total_weight
manipulation_values = [detector.manipulation_probability for detector in detectors if detector.manipulation_probability is not None]
manipulation_probability = max(manipulation_values) if manipulation_values else forensics["manipulation_score"]
ai_values = [detector.ai_probability for detector in detectors if detector.status == "ok" and detector.ai_probability is not None]
cross_layer_disagreement = (max(ai_values) - min(ai_values)) if len(ai_values) >= 2 else 0.0
disagreement = max(cross_layer_disagreement, model_disagreement)
strong_signals = [
detector
for detector in available
if detector.confidence in {"medium", "high"} and detector.label not in {"no_strong_forensic_signal", "camera_metadata_present"}
]
has_model = any(detector.name.startswith("hf:") and detector.status == "ok" for detector in detectors)
rationale = _collect_rationale(detectors)
non_model_ai_support = any(
detector.name in {"metadata_provenance", "compression_noise_forensics"}
and detector.confidence in {"medium", "high"}
and detector.ai_probability is not None
and detector.ai_probability >= 0.72
for detector in detectors
)
overwhelming_model_consensus = (
len(model_values) >= 3
and model_ai_votes == len(model_values)
and model_min_score >= 0.86
and model_real_votes == 0
and model_disagreement <= 0.22
)
primary_anchored_consensus = (
len(model_values) >= 3
and model_signal is not None
and model_signal.label == "ensemble_likely_ai_generated"
and model_lean_ai_votes == len(model_values)
and model_real_votes == 0
and model_disagreement <= 0.50
and any(
_model_profile(detector.name.removeprefix("hf:")).get("expert_group") == "broad_primary"
and float(detector.ai_probability or 0.0)
>= float(_model_profile(detector.name.removeprefix("hf:")).get("lean_ai_threshold", 0.7))
for detector in model_detectors
)
)
larger_model_consensus = (
overwhelming_model_consensus
or primary_anchored_consensus
or (
len(model_values) >= 4
and model_lean_ai_votes == len(model_values)
and model_ai_votes >= len(model_values) - 1
and model_real_votes == 0
and model_min_score >= 0.78
and model_disagreement <= 0.28
)
)
model_only_ai_claim = (
model_signal is not None
and model_signal.label == "ensemble_likely_ai_generated"
and not non_model_ai_support
and not larger_model_consensus
)
portrait_real_support = any(
_is_portrait_specialist(detector)
and detector.label == "model_likely_human_or_real"
and detector.ai_probability is not None
and detector.ai_probability <= 0.12
for detector in model_detectors
)
portrait_real_override = (
portrait_real_support
and (model_real_votes > model_ai_votes or camera_like_real_prior or metadata.get("has_exif"))
and model_disagreement >= 0.5
and not non_model_ai_support
and quality_risk < 0.55
and forensics["artificiality_score"] < 0.45
and forensics["manipulation_score"] < 0.65
)
weak_non_model_artifacts = (
not metadata["generative_markers"]
and c2pa.get("claim") != "ai_generated_or_synthetic"
and forensics["artificiality_score"] < 0.45
and forensics["manipulation_score"] < 0.65
)
uncorroborated_model_evidence = has_model and weak_non_model_artifacts and not non_model_ai_support
model_uncorroborated = uncorroborated_model_evidence and (
(model_disagreement >= 0.50 and not primary_anchored_consensus)
or (model_signal is not None and model_signal.confidence == "low")
or len(model_values) < 3
or model_real_votes > 0
or not larger_model_consensus
or model_only_ai_claim
)
model_only_probability_cap = None
if uncorroborated_model_evidence:
model_only_probability_cap = 0.72 if larger_model_consensus else 0.55
if len(model_values) < 3:
model_only_probability_cap = min(model_only_probability_cap, 0.52)
if model_real_votes > 0:
model_only_probability_cap = min(model_only_probability_cap, 0.52)
if model_disagreement >= 0.35 and not primary_anchored_consensus:
model_only_probability_cap = min(model_only_probability_cap, 0.52)
if quality_risk >= 0.45:
model_only_probability_cap = min(model_only_probability_cap, 0.54)
if metadata.get("has_exif") or camera_like_real_prior:
model_only_probability_cap = min(model_only_probability_cap, 0.48)
if model_only_probability_cap is not None and ai_probability > model_only_probability_cap:
ai_probability = model_only_probability_cap
rationale.append(
"Final AI evidence score was capped because visual-model suspicion was not corroborated by independent metadata, provenance, or forensic evidence."
)
if camera_like_real_prior:
rationale.append(
"Camera-like JPEG/container and residual evidence triggered the real-photo false-positive guard."
)
if model_real_votes > 0 or (model_disagreement >= 0.35 and not primary_anchored_consensus):
rationale.append(
"Detector disagreement or at least one real/human model vote prevented a stronger AI claim."
)
if primary_anchored_consensus:
rationale.append(
"The broad primary detector and every configured counter-expert independently leaned AI-generated, satisfying the primary-anchored consensus gate."
)
if quality_risk >= 0.55 and not non_model_ai_support and ai_probability > 0.68:
ai_probability = 0.68
rationale.append(
"Final AI evidence score was capped because input-quality risk is high; low-resolution, cropped, or heavily exported media weakens detector reliability."
)
if portrait_real_override and ai_probability > 0.38:
ai_probability = 0.38
rationale.append(
"Calibrated portrait-specialist evidence and at least one generic real/human vote reduced the final AI evidence score."
)
model_supports_ai = (
model_signal is not None
and model_signal.label == "ensemble_likely_ai_generated"
and model_signal.confidence in {"medium", "high"}
and (non_model_ai_support or larger_model_consensus)
)
if model_supports_ai and not model_uncorroborated and model_signal.ai_probability is not None:
upper_bound = 0.72 if uncorroborated_model_evidence else 0.88
if model_only_probability_cap is not None:
upper_bound = min(upper_bound, model_only_probability_cap)
ai_probability = max(ai_probability, min(float(model_signal.ai_probability), upper_bound))
if model_supports_ai and ai_probability >= 0.72:
label = "likely_ai_generated"
elif ai_probability >= 0.78 and (strong_signals or non_model_ai_support) and not model_uncorroborated:
label = "likely_ai_generated"
elif manipulation_probability >= 0.68:
label = "likely_manipulated_or_deepfake"
elif (
ai_probability <= 0.28
and manipulation_probability < 0.35
and (has_model or metadata["has_exif"] or c2pa.get("status") == "found")
):
label = "likely_real"
elif portrait_real_override and ai_probability <= 0.45:
label = "likely_real"
else:
label = "inconclusive"
if label == "inconclusive":
confidence = "low"
elif model_supports_ai:
confidence = "medium"
elif has_model and disagreement < 0.28:
confidence = "medium"
elif strong_signals and disagreement < 0.35:
confidence = "medium"
else:
confidence = "low"
return {
"label": label,
"confidence": confidence,
"ai_probability": round(float(ai_probability), 4),
"manipulation_probability": round(float(manipulation_probability), 4),
"disagreement": round(float(disagreement), 4),
"model_stance_disagreement": round(float(model_disagreement), 4),
"raw_model_score_range": round(float(raw_model_score_range), 4),
"cross_layer_disagreement": round(float(cross_layer_disagreement), 4),
"rationale": rationale,
"score_label": "AI evidence score",
"score_interpretation": "Evidence strength for triage, not a calibrated probability of truth.",
}
def build_layers(
metadata: dict[str, Any],
hashes: dict[str, Any],
c2pa: dict[str, Any],
forensics: dict[str, Any],
detectors: list[DetectorSignal],
source_context: dict | None,
) -> list[dict[str, Any]]:
return [
{
"name": "Input and Source",
"status": "complete",
"findings": [
f"Image dimensions: {metadata['width']} x {metadata['height']}.",
f"Input source: {(source_context or {}).get('domain') or 'direct upload'}.",
],
},
{
"name": "Metadata",
"status": "complete",
"findings": [
"EXIF metadata present." if metadata["has_exif"] else "No EXIF metadata found.",
"GPS metadata was present and redacted." if metadata["gps_present"] else "No GPS metadata was exposed.",
f"Software markers: {', '.join(metadata['software_values'])}." if metadata["software_values"] else "No software marker found.",
],
},
{
"name": "Provenance",
"status": c2pa["status"],
"findings": c2pa["evidence"],
},
{
"name": "Hashes",
"status": "complete",
"findings": [
f"SHA-256: {hashes['sha256']}.",
f"Perceptual hash: {hashes['average_hash']}.",
],
},
{
"name": "Compression and Noise",
"status": "complete",
"findings": [
f"Manipulation heuristic score: {forensics['manipulation_score']}.",
f"Artificiality heuristic score: {forensics['artificiality_score']}.",
],
},
{
"name": "Detector Ensemble",
"status": "complete",
"findings": [
f"{detector.name}: {detector.label} ({detector.status})." for detector in detectors
],
},
]
def build_explainability(
verdict: dict[str, Any],
detectors: list[DetectorSignal],
metadata: dict[str, Any],
c2pa: dict[str, Any],
forensics: dict[str, Any],
analytical_layers: list[dict[str, Any]],
) -> dict[str, Any]:
model_signals = [
detector
for detector in detectors
if detector.name.startswith("hf:") and detector.status == "ok" and detector.ai_probability is not None
]
ensemble = next((detector for detector in detectors if detector.name == "open_source_model_ensemble"), None)
model_scores = [
{
"name": signal.name,
"ai_probability": signal.ai_probability,
"calibrated_stance_score": round(_calibrated_model_stance(signal), 4),
"label": signal.label,
"reliability_weight": signal.weight,
"evidence": signal.evidence,
"details": signal.details,
}
for signal in model_signals
]
strongest = sorted(
[
{
"source": detector.name,
"label": detector.label,
"ai_probability": detector.ai_probability,
"manipulation_probability": detector.manipulation_probability,
"evidence": detector.evidence,
}
for detector in detectors
if detector.status == "ok"
],
key=lambda item: max(
abs((item["ai_probability"] or 0.5) - 0.5),
abs((item["manipulation_probability"] or 0.25) - 0.25),
),
reverse=True,
)
model_values = [float(signal.ai_probability) for signal in model_signals if signal.ai_probability is not None]
calibrated_model_values = [_calibrated_model_stance(signal) for signal in model_signals]
model_lean_ai_votes = sum(
float(signal.ai_probability or 0.0)
>= float(_model_profile(signal.name.removeprefix("hf:")).get("lean_ai_threshold", 0.6))
for signal in model_signals
)
primary_anchor_support = any(
_model_profile(signal.name.removeprefix("hf:")).get("expert_group") == "broad_primary"
and float(signal.ai_probability or 0.0)
>= float(_model_profile(signal.name.removeprefix("hf:")).get("lean_ai_threshold", 0.7))
for signal in model_signals
)
weighted_model_average = (
float(ensemble.ai_probability)
if ensemble and ensemble.ai_probability is not None
else _weighted_average(model_values, [float(signal.weight or DEFAULT_MODEL_PROFILE["weight"]) for signal in model_signals])
if model_values
else None
)
layer_votes = {
"supports_ai_generated": sum(layer["conclusion"] == "supports_ai_generated" for layer in analytical_layers),
"supports_real_or_camera_origin": sum(layer["conclusion"] == "supports_real_or_camera_origin" for layer in analytical_layers),
"supports_manipulation_or_editing": sum(layer["conclusion"] == "supports_manipulation_or_editing" for layer in analytical_layers),
"weak_anomaly": sum(layer["conclusion"] == "weak_anomaly" for layer in analytical_layers),
"neutral_or_inconclusive": sum(layer["conclusion"] == "neutral_or_inconclusive" for layer in analytical_layers),
}
decision_support = _decision_support(verdict, analytical_layers, layer_votes, c2pa, model_scores)
decision_attribution = _decision_attribution(verdict, detectors, metadata, c2pa, forensics, analytical_layers)
expert_opinions = _mixture_expert_opinions(verdict, detectors, metadata, c2pa, forensics)
regional_map = _regional_evidence_map(analytical_layers)
return {
"decision_trace": [
f"Final label is {verdict['label']} with {verdict['confidence']} confidence.",
f"Combined AI evidence score is {verdict['ai_probability']:.0%}; it is not a probability of truth.",
f"Combined manipulation evidence score is {verdict['manipulation_probability']:.0%}.",
(
f"Calibrated model-stance disagreement is {float(verdict.get('model_stance_disagreement') or 0):.0%}; "
f"raw model score range is {float(verdict.get('raw_model_score_range') or 0):.0%} and is diagnostic only."
),
(
"Layer ledger: "
f"{layer_votes['supports_ai_generated']} support AI, "
f"{layer_votes['supports_real_or_camera_origin']} support real/camera-origin, "
f"{layer_votes['supports_manipulation_or_editing']} support manipulation, "
f"{layer_votes['weak_anomaly']} weak anomalies, "
f"{layer_votes['neutral_or_inconclusive']} neutral/inconclusive."
),
],
"decision_support": decision_support,
"decision_attribution": decision_attribution,
"explanation_contract": {
"score_semantics": "Evidence strength for cautious triage, not a probability that the image is fake.",
"arbiter_inputs": [item["source"] for item in decision_attribution if item["used_by_arbiter"]],
"context_only_layers": [
layer["name"] for layer in analytical_layers if layer.get("decision_role") == "review_context_only"
],
"abstention_rule": "If provenance is absent and calibrated experts conflict, the system returns inconclusive instead of forcing a binary claim.",
"human_review_boundary": "The result cannot establish identity, consent, legal truth, or the exact generator from pixels alone.",
},
"expert_opinions": expert_opinions,
"model_consensus": {
"enabled_models": len(model_signals),
"ai_votes": sum(signal.label == "model_likely_ai_generated" for signal in model_signals),
"lean_ai_votes": model_lean_ai_votes,
"real_votes": sum(signal.label == "model_likely_human_or_real" for signal in model_signals),
"inconclusive_votes": sum(signal.label == "model_inconclusive" for signal in model_signals),
"average_ai_probability": round(weighted_model_average, 4) if weighted_model_average is not None else None,
"score_semantics": "calibrated evidence score after model-specific abstention bands",
"raw_average_ai_probability": round(float(np.mean(model_values)), 4) if model_values else None,
"calibrated_stance_disagreement": (
round(float(max(calibrated_model_values) - min(calibrated_model_values)), 4)
if len(calibrated_model_values) >= 2
else 0.0
),
"raw_score_range": round(float(max(model_values) - min(model_values)), 4) if len(model_values) >= 2 else 0.0,
"ensemble_label": ensemble.label if ensemble else None,
"ensemble_confidence": ensemble.confidence if ensemble else None,
"primary_anchored_alignment": bool(
ensemble
and ensemble.label == "ensemble_likely_ai_generated"
and primary_anchor_support
and model_lean_ai_votes == len(model_signals)
),
"models": model_scores,
},
"decision_standard": {
"policy": "victim_safe_calibrated_triage",
"likely_ai_requires": [
"Readable generative metadata or C2PA declaration,",
"or independent non-model forensic/provenance support,",
"or unusually strong, tightly aligned multi-model consensus with no model voting real/human-origin.",
],
"false_positive_controls": [
"Raw model scores are converted through model-specific AI/real/abstain bands before reliability weighting.",
"Uncorroborated model-only suspicion is capped instead of being treated as proof.",
"Camera-like JPEG/container evidence lowers the final AI evidence score unless independent AI evidence exists.",
"Detector disagreement and real/human votes force abstention.",
"A model-only claim needs either unanimous strong votes or broad-primary support aligned with every counter-expert.",
"The broad primary must remain stable across original, center-crop, JPEG, mirror, and social-resize views.",
"Low-quality crops, screenshots, and compressed exports cap confidence.",
],
"input_quality": forensics.get("quality", {}),
},
"non_model_evidence": {
"metadata": "Generative markers found." if metadata["generative_markers"] else "No generative metadata marker found.",
"c2pa": c2pa["status"],
"forensic_artificiality_score": forensics["artificiality_score"],
"forensic_manipulation_score": forensics["manipulation_score"],
},
"strongest_evidence": strongest[:5],
"regional_evidence_map": regional_map,
"layer_ledger": {
"counts": layer_votes,
"layers": analytical_layers,
},
"calibration_note": (
"The displayed score is calibrated evidence strength for cautious triage, not a probability that the image is fake. "
"Agreement across independent detectors and provenance is stronger than any single model score."
),
}
def _decision_attribution(
verdict: dict[str, Any],
detectors: list[DetectorSignal],
metadata: dict[str, Any],
c2pa: dict[str, Any],
forensics: dict[str, Any],
analytical_layers: list[dict[str, Any]],
) -> list[dict[str, Any]]:
attributions: list[dict[str, Any]] = []
metadata_signal = next((detector for detector in detectors if detector.name == "metadata_provenance"), None)
ensemble = next((detector for detector in detectors if detector.name == "open_source_model_ensemble"), None)
forensic_signal = next((detector for detector in detectors if detector.name == "compression_noise_forensics"), None)
if metadata_signal:
provenance_direction = (
"toward_ai_generated"
if metadata_signal.label == "likely_ai_generated"
else "toward_real_origin"
if metadata_signal.label == "camera_metadata_present"
else "neutral"
)
attributions.append(
{
"source": "Metadata and signed provenance",
"role": "primary_evidence",
"direction": provenance_direction,
"strength": round(abs(float(metadata_signal.ai_probability or 0.5) - 0.5) * 2.0, 4),
"used_by_arbiter": True,
"finding": metadata_signal.evidence[0] if metadata_signal.evidence else "No provenance finding was available.",
"counterfactual": "A verified contradictory C2PA claim or original camera file would materially change this attribution.",
}
)
if ensemble:
ensemble_direction = (
"toward_ai_generated"
if ensemble.label == "ensemble_likely_ai_generated"
else "toward_real_origin"
if ensemble.label == "ensemble_likely_real"
else "neutral"
)
attributions.append(
{
"source": "Calibrated visual-model ensemble",
"role": "primary_evidence",
"direction": ensemble_direction,
"strength": round(abs(float(ensemble.ai_probability or 0.5) - 0.5) * 2.0, 4),
"used_by_arbiter": True,
"finding": ensemble.evidence[0] if ensemble.evidence else "No ensemble finding was available.",
"counterfactual": "A real-origin model vote, unstable primary views, or missing primary alignment forces abstention.",
}
)
if forensic_signal:
forensic_direction = (
"toward_manipulation"
if float(forensic_signal.manipulation_probability or 0.0) >= 0.68
else "toward_ai_generated"
if float(forensic_signal.ai_probability or 0.5) >= 0.72
else "neutral"
)
forensic_strength = max(
abs(float(forensic_signal.ai_probability or 0.5) - 0.5) * 2.0,
max(0.0, float(forensic_signal.manipulation_probability or 0.25) - 0.5) * 2.0,
)
attributions.append(
{
"source": "Compression and noise forensic composite",
"role": "supporting_evidence",
"direction": forensic_direction,
"strength": round(_clamp(forensic_strength) * 0.45, 4),
"used_by_arbiter": True,
"finding": forensic_signal.evidence[0] if forensic_signal.evidence else "No forensic finding was available.",
"counterfactual": "Original, unrecompressed media can distinguish editing artifacts from platform recompression.",
}
)
quality = forensics.get("quality") or {}
quality_risk = float(quality.get("risk_score") or 0.0)
attributions.append(
{
"source": "Input-quality confidence guard",
"role": "confidence_guard",
"direction": "toward_abstention" if quality_risk >= 0.45 else "neutral",
"strength": round(quality_risk, 4),
"used_by_arbiter": True,
"finding": (quality.get("evidence") or ["No major input-quality risk was detected."])[0],
"counterfactual": "A larger original file with fewer exports or screenshots would reduce this confidence penalty.",
}
)
try:
camera_prior = _camera_like_real_prior(metadata, forensics)
except (KeyError, TypeError, ValueError):
camera_prior = False
if camera_prior:
attributions.append(
{
"source": "Camera-like real-photo guard",
"role": "false_positive_guard",
"direction": "toward_real_origin",
"strength": 0.7,
"used_by_arbiter": True,
"finding": "Camera-compatible container and natural residual evidence triggered the real-photo guard.",
"counterfactual": "Verified synthetic provenance or strong independent forensic support would override this guard.",
}
)
context_layers = sorted(
[layer for layer in analytical_layers if layer.get("decision_role") == "review_context_only"],
key=lambda layer: float(layer.get("influence") or 0.0),
reverse=True,
)
if context_layers:
attributions.append(
{
"source": "Diagnostic image decompositions",
"role": "review_context_only",
"direction": "not_used_for_classification",
"strength": 0.0,
"used_by_arbiter": False,
"finding": f"{len(context_layers)} luminance, color, edge, frequency, and regional layers are exposed for reviewer context.",
"counterfactual": "These layers require independent corroboration before they can affect a verdict.",
}
)
return attributions
def _decision_support(
verdict: dict[str, Any],
analytical_layers: list[dict[str, Any]],
layer_votes: dict[str, int],
c2pa: dict[str, Any],
model_scores: list[dict[str, Any]],
) -> dict[str, Any]:
label = verdict["label"]
if label == "likely_ai_generated":
supporting = [layer for layer in analytical_layers if layer["conclusion"] == "supports_ai_generated"]
counter = [layer for layer in analytical_layers if layer["conclusion"] == "supports_real_or_camera_origin"]
plain_summary = "The AI-generated verdict is mainly driven by visual detector evidence, then checked against metadata and forensic layers."
elif label == "likely_manipulated_or_deepfake":
supporting = [layer for layer in analytical_layers if layer["conclusion"] == "supports_manipulation_or_editing"]
counter = [layer for layer in analytical_layers if layer["conclusion"] == "supports_real_or_camera_origin"]
plain_summary = "The manipulation verdict is driven by editing or regional forensic signals, not by identity inference."
elif label == "likely_real":
supporting = [layer for layer in analytical_layers if layer["conclusion"] == "supports_real_or_camera_origin"]
counter = [
layer
for layer in analytical_layers
if layer["conclusion"] in {"supports_ai_generated", "supports_manipulation_or_editing"}
]
plain_summary = "The real-origin verdict is supported by camera/provenance-style evidence and a low AI evidence score."
else:
supporting = [
layer
for layer in analytical_layers
if layer["conclusion"] in {"supports_ai_generated", "supports_real_or_camera_origin", "supports_manipulation_or_editing"}
]
counter = [layer for layer in analytical_layers if layer["conclusion"] == "weak_anomaly"]
plain_summary = "The evidence does not cross a safe threshold; the report is preserving uncertainty instead of forcing a binary answer."
supporting = sorted(supporting, key=_layer_strength, reverse=True)
counter = sorted(counter, key=_layer_strength, reverse=True)
weak_anomalies = [layer for layer in analytical_layers if layer["conclusion"] == "weak_anomaly"]
uncertainty = []
if verdict["confidence"] == "low":
uncertainty.append("Confidence is low, so this should be treated as triage evidence rather than a final forensic certificate.")
if verdict["disagreement"] >= 0.35:
uncertainty.append(f"Cross-layer evidence disagreement is elevated at {verdict['disagreement']:.0%}.")
if c2pa.get("status") != "found":
uncertainty.append("No signed C2PA/content credential was available to independently confirm provenance.")
if len(model_scores) < 2:
uncertainty.append("Fewer than two visual model scores were available.")
if weak_anomalies:
uncertainty.append(f"{len(weak_anomalies)} layer(s) produced weak anomaly signals that need context.")
if layer_votes["neutral_or_inconclusive"] >= 4:
uncertainty.append(f"{layer_votes['neutral_or_inconclusive']} layer(s) were neutral or inconclusive.")
return {
"plain_summary": plain_summary,
"primary_drivers": _layer_summaries(supporting[:4])
or ["No single layer was strong enough on its own; the final label comes from combined probabilities."],
"counter_evidence": _layer_summaries(counter[:4]) or ["No strong counter-evidence was found in the analytical layers."],
"uncertainty_factors": uncertainty[:5] or ["No major uncertainty factor was detected, but image authenticity still cannot be proven from pixels alone."],
"what_would_help": [
"Original, uncompressed media from the device or platform export.",
"Public source URL, timestamps, captions, and account/page context.",
"Signed C2PA/content credentials where available.",
"Manual review when the result affects safety, reputation, or legal action.",
],
}
def _mixture_expert_opinions(
verdict: dict[str, Any],
detectors: list[DetectorSignal],
metadata: dict[str, Any],
c2pa: dict[str, Any],
forensics: dict[str, Any],
) -> list[dict[str, Any]]:
opinions: list[dict[str, Any]] = []
ensemble = next((detector for detector in detectors if detector.name == "open_source_model_ensemble"), None)
if ensemble:
opinions.append(
{
"expert": "Visual Detector Ensemble",
"opinion": ensemble.label,
"stance": _stance_from_label(ensemble.label),
"confidence": ensemble.confidence,
"score": ensemble.ai_probability,
"evidence": ensemble.evidence[:3],
}
)
portrait_signal = next((detector for detector in detectors if _is_portrait_specialist(detector)), None)
if portrait_signal:
opinions.append(
{
"expert": "Portrait Specialist",
"opinion": portrait_signal.label,
"stance": _stance_from_label(portrait_signal.label),
"confidence": portrait_signal.confidence,
"score": portrait_signal.ai_probability,
"evidence": portrait_signal.evidence[:3],
}
)
metadata_stance = "supports_ai" if metadata["generative_markers"] or c2pa.get("claim") == "ai_generated_or_synthetic" else "neutral"
opinions.append(
{
"expert": "Metadata And Provenance",
"opinion": "generative_marker_found" if metadata_stance == "supports_ai" else "no_generative_marker",
"stance": metadata_stance,
"confidence": "high" if c2pa.get("claim") else "low",
"score": 0.9 if metadata_stance == "supports_ai" else 0.5,
"evidence": [
f"Generative markers: {', '.join(metadata['generative_markers']) or 'none'}.",
f"C2PA status: {c2pa.get('status')}.",
],
}
)
forensic_stance = "supports_manipulation" if forensics["manipulation_score"] >= 0.68 else "neutral"
if forensics["artificiality_score"] >= 0.7:
forensic_stance = "supports_ai"
opinions.append(
{
"expert": "Forensic Residuals",
"opinion": forensic_stance,
"stance": forensic_stance,
"confidence": "medium" if forensic_stance != "neutral" else "low",
"score": max(forensics["artificiality_score"], forensics["manipulation_score"]),
"evidence": [
f"Artificiality score: {forensics['artificiality_score']}.",
f"Manipulation score: {forensics['manipulation_score']}.",
],
}
)
quality = forensics.get("quality") or {}
opinions.append(
{
"expert": "Input Quality Guard",
"opinion": f"{quality.get('risk_band', 'unknown')}_quality_risk",
"stance": "limits_confidence" if float(quality.get("risk_score") or 0.0) >= 0.25 else "neutral",
"confidence": "medium",
"score": quality.get("risk_score"),
"evidence": (quality.get("evidence") or [])[:3],
}
)
opinions.append(
{
"expert": "Safety Arbiter",
"opinion": verdict["label"],
"stance": _stance_from_label(verdict["label"]),
"confidence": verdict["confidence"],
"score": verdict["ai_probability"],
"evidence": verdict["rationale"][-3:],
}
)
return opinions
def _stance_from_label(label: str) -> str:
normalized = label.lower().replace("-", "_")
tokens = set(normalized.split("_"))
if normalized in {"portrait_gate_not_met", "ensemble_inconclusive", "model_inconclusive", "inconclusive"}:
return "neutral"
if (
"likely_ai_generated" in normalized
or "ai_generated" in normalized
or "synthetic" in tokens
or "fake" in tokens
or "deepfake" in tokens
):
return "supports_ai"
if "real" in tokens or "human" in tokens or "camera" in tokens:
return "supports_real"
if "manipulated" in tokens or "editing" in tokens:
return "supports_manipulation"
return "neutral"
def _layer_strength(layer: dict[str, Any]) -> float:
return max(abs(float(layer.get("ai_signal") or 0.5) - 0.5), abs(float(layer.get("manipulation_signal") or 0.25) - 0.25))
def _layer_summaries(layers: list[dict[str, Any]]) -> list[str]:
return [
(
f"{layer['name']}: {layer['conclusion']} "
f"(AI {float(layer['ai_signal']):.0%}, manipulation {float(layer['manipulation_signal']):.0%})."
)
for layer in layers
]
def _regional_evidence_map(analytical_layers: list[dict[str, Any]]) -> dict[str, Any] | None:
tile_layer = next((layer for layer in analytical_layers if layer.get("id") == "tile_regions"), None)
if not tile_layer:
return None
metrics = tile_layer.get("metrics") or {}
tile_grid = metrics.get("tile_grid") or []
if not tile_grid:
return None
return {
"grid": {"rows": 4, "cols": 4},
"tiles": tile_grid,
"max_score": metrics.get("max_tile_anomaly_score"),
"mean_score": metrics.get("mean_tile_anomaly_score"),
"high_severity_tile_count": metrics.get("high_severity_tile_count", 0),
"interpretation": (
"This abstract map compares regions against the rest of the same image. "
"It does not display the uploaded image and does not identify what caused a regional difference."
),
}
def _safe_metadata_value(value: Any) -> str:
if isinstance(value, bytes):
return value[:128].hex()
if isinstance(value, tuple):
return ", ".join(_safe_metadata_value(item) for item in value[:12])
return str(value)[:500]
def _metadata_marker_text(exif: dict[str, str], png_text: dict[str, str], xmp: str | None) -> str:
values = [str(value) for value in exif.values()]
values.extend(str(value) for value in png_text.values())
if xmp:
values.append(xmp)
return " ".join(values).lower()
def _compact_metadata_summary(value: str, *, limit: int = 120) -> str:
compact = " ".join(str(value).split())
if len(compact) <= limit:
return compact
return f"{compact[: limit - 1].rstrip()}..."
def _xmp_report_summary(xmp: str | None, generative_markers: list[str], editing_markers: list[str]) -> str | None:
if not xmp:
return None
markers = sorted(set(generative_markers + editing_markers))
marker_text = f"; markers: {', '.join(markers)}" if markers else ""
return f"XMP metadata present ({len(xmp)} characters{marker_text}); raw XMP omitted from stored report."
def _collect_software_values(
exif: dict[str, str],
png_text: dict[str, str],
xmp: str | None,
generative_markers: list[str],
editing_markers: list[str],
) -> list[str]:
values: list[str] = []
for key in ("Software", "ProcessingSoftware", "Make", "Model"):
if exif.get(key):
values.append(_compact_metadata_summary(str(exif[key])))
for key, value in png_text.items():
if key.lower() in {"software", "parameters", "prompt", "workflow", "generation_data"}:
values.append(f"{key}: {_compact_metadata_summary(str(value))}")
if xmp:
markers = sorted(set(generative_markers + editing_markers))
if markers:
values.append(f"XMP metadata mentions: {', '.join(markers)}")
else:
values.append("XMP metadata present")
return values
def _average_hash(image: Image.Image) -> str:
small = image.convert("L").resize((8, 8), Image.Resampling.LANCZOS)
pixels = list(small.getdata())
avg = sum(pixels) / len(pixels)
bits = "".join("1" if pixel >= avg else "0" for pixel in pixels)
return f"{int(bits, 2):016x}"
def _difference_hash(image: Image.Image) -> str:
small = image.convert("L").resize((9, 8), Image.Resampling.LANCZOS)
pixels = list(small.getdata())
bits = []
for row in range(8):
offset = row * 9
for col in range(8):
bits.append("1" if pixels[offset + col] > pixels[offset + col + 1] else "0")
return f"{int(''.join(bits), 2):016x}"
def _ela_metrics(image: Image.Image) -> dict[str, Any]:
buffer = BytesIO()
image.save(buffer, format="JPEG", quality=90)
buffer.seek(0)
recompressed = Image.open(buffer).convert("RGB")
diff = ImageChops.difference(image, recompressed)
stat = ImageStat.Stat(diff)
mean = float(sum(stat.mean) / len(stat.mean))
extrema = diff.getextrema()
max_delta = max(channel[1] for channel in extrema)
return {
"mean_delta": round(mean, 4),
"max_delta": int(max_delta),
"normalized_mean": round(_clamp((mean - 2.5) / 18.0), 4),
}
def _noise_metrics(image: Image.Image) -> dict[str, Any]:
gray = image.convert("L")
edges = gray.filter(ImageFilter.FIND_EDGES)
tile_scores = []
tile_w = max(8, gray.width // 4)
tile_h = max(8, gray.height // 4)
for top in range(0, gray.height, tile_h):
for left in range(0, gray.width, tile_w):
tile = edges.crop((left, top, min(left + tile_w, gray.width), min(top + tile_h, gray.height)))
arr = np.asarray(tile, dtype=np.float32)
tile_scores.append(float(arr.var()))
mean = float(np.mean(tile_scores)) if tile_scores else 0.0
std = float(np.std(tile_scores)) if tile_scores else 0.0
coefficient = std / (mean + 1e-6)
low_noise_hint = _clamp((90.0 - mean) / 90.0)
return {
"tile_variance_mean": round(mean, 4),
"tile_variance_std": round(std, 4),
"tile_inconsistency": round(_clamp(coefficient / 1.8), 4),
"low_noise_hint": round(low_noise_hint, 4),
}
def _jpeg_marker_summary(image_bytes: bytes) -> dict[str, Any]:
# Multiple quantization tables can be normal; this only contributes a weak hint.
dqt_count = image_bytes.count(b"\xff\xdb")
return {
"dqt_marker_count": dqt_count,
"double_quantization_hint": 0.3 if dqt_count >= 3 else 0.0,
}
def _input_quality_metrics(image: Image.Image, image_bytes: bytes, entropy: float) -> dict[str, Any]:
width, height = image.size
pixels = width * height
shortest_edge = min(width, height)
aspect_ratio = max(width, height) / max(1, shortest_edge)
bytes_per_pixel = len(image_bytes) / max(1, pixels)
flags: list[str] = []
risk = 0.0
if pixels < 120_000 or shortest_edge < 256:
flags.append("low_resolution_or_crop")
risk += 0.35
elif pixels < 300_000:
flags.append("moderate_resolution")
risk += 0.18
if aspect_ratio >= 2.75:
flags.append("extreme_aspect_ratio")
risk += 0.15
if bytes_per_pixel < 0.08 and pixels > 250_000:
flags.append("heavy_compression_or_platform_export")
risk += 0.18
if entropy <= 4.2:
flags.append("very_low_texture_entropy")
risk += 0.12
if image.format in {"PNG", "WEBP"} and pixels < 500_000:
flags.append("possible_screenshot_or_export")
risk += 0.10
risk_score = round(_clamp(risk), 4)
if flags:
evidence = [
f"Robustness risk flags: {', '.join(flags)}.",
f"Resolution: {width} x {height}; shortest edge {shortest_edge}px.",
f"Bytes per pixel: {bytes_per_pixel:.3f}; entropy: {entropy:.3f}.",
]
else:
evidence = [
"No major input-quality risk flags were detected.",
f"Resolution: {width} x {height}; shortest edge {shortest_edge}px.",
f"Bytes per pixel: {bytes_per_pixel:.3f}; entropy: {entropy:.3f}.",
]
return {
"risk_score": risk_score,
"risk_band": "high" if risk_score >= 0.55 else "medium" if risk_score >= 0.25 else "low",
"flags": flags,
"width": width,
"height": height,
"pixels": pixels,
"shortest_edge": shortest_edge,
"aspect_ratio": round(aspect_ratio, 4),
"bytes_per_pixel": round(bytes_per_pixel, 4),
"entropy": entropy,
"evidence": evidence,
}
def _portrait_likelihood_metrics(image: Image.Image) -> dict[str, Any]:
rgb = image.convert("RGB").resize((256, 256), Image.Resampling.BILINEAR)
arr = np.asarray(rgb, dtype=np.float32)
ycbcr = np.asarray(rgb.convert("YCbCr"), dtype=np.float32)
cb = ycbcr[:, :, 1]
cr = ycbcr[:, :, 2]
skin_mask = (cb >= 77) & (cb <= 135) & (cr >= 130) & (cr <= 180)
height, width = skin_mask.shape
center = skin_mask[int(height * 0.12) : int(height * 0.82), int(width * 0.25) : int(width * 0.75)]
side_left = skin_mask[:, : int(width * 0.18)]
side_right = skin_mask[:, int(width * 0.82) :]
center_skin_ratio = float(np.mean(center)) if center.size else 0.0
side_skin_ratio = float((np.mean(side_left) + np.mean(side_right)) / 2.0) if side_left.size and side_right.size else 0.0
centrality = _clamp((center_skin_ratio - side_skin_ratio + 0.12) / 0.42)
aspect_ratio = image.width / max(1, image.height)
aspect_hint = _clamp(1.0 - abs(aspect_ratio - 0.78) / 0.55)
gray = np.asarray(rgb.convert("L"), dtype=np.float32)
center_luma = float(np.mean(gray[int(height * 0.18) : int(height * 0.80), int(width * 0.28) : int(width * 0.72)]))
edge_luma = float(
np.mean(
np.concatenate(
[
gray[:, : int(width * 0.12)].reshape(-1),
gray[:, int(width * 0.88) :].reshape(-1),
]
)
)
)
subject_contrast = _clamp((center_luma - edge_luma + 35.0) / 90.0)
portrait_score = _clamp(0.45 * center_skin_ratio + 0.25 * centrality + 0.18 * aspect_hint + 0.12 * subject_contrast)
evidence = [
f"Portrait likelihood score: {portrait_score:.3f}.",
f"Central skin-tone ratio: {center_skin_ratio:.3f}; side skin-tone ratio: {side_skin_ratio:.3f}.",
f"Aspect hint: {aspect_hint:.3f}; subject contrast hint: {subject_contrast:.3f}.",
]
return {
"portrait_score": round(portrait_score, 4),
"center_skin_ratio": round(center_skin_ratio, 4),
"side_skin_ratio": round(side_skin_ratio, 4),
"centrality": round(centrality, 4),
"aspect_hint": round(aspect_hint, 4),
"subject_contrast": round(subject_contrast, 4),
"evidence": evidence,
}
def _map_classifier_outputs_to_ai_probability(
outputs: list[dict[str, Any]],
profile: dict[str, Any] | None = None,
) -> float:
profile = profile or {}
exact_ai_labels = {str(label).strip().lower() for label in profile.get("ai_labels", [])}
exact_real_labels = {str(label).strip().lower() for label in profile.get("real_labels", [])}
ai_score_total = 0.0
real_score_total = 0.0
for item in outputs:
label = str(item.get("label", "")).lower()
score = float(item.get("score", 0.0))
if label in exact_ai_labels or any(
token in label for token in ("ai", "fake", "generated", "synthetic", "deepfake", "artificial")
):
ai_score_total += score
if label in exact_real_labels or any(token in label for token in ("real", "human", "hum", "natural", "authentic")):
real_score_total += score
if ai_score_total or real_score_total:
return _clamp(ai_score_total / (ai_score_total + real_score_total + 1e-9))
return 0.5
def _normalize_classifier_outputs(outputs: Any) -> list[dict[str, Any]]:
if isinstance(outputs, list) and outputs and isinstance(outputs[0], list):
outputs = outputs[0]
if not isinstance(outputs, list):
return []
return [item for item in outputs if isinstance(item, dict)]
def _summarize_labels(outputs: list[dict[str, Any]]) -> str:
return ", ".join(f"{item.get('label')}={float(item.get('score', 0.0)):.3f}" for item in outputs[:5])
def _collect_rationale(detectors: list[DetectorSignal]) -> list[str]:
rationale: list[str] = []
for detector in detectors:
rationale.extend(detector.evidence[:2])
return rationale[:8]
def _headline(label: str) -> str:
return {
"likely_real": "The image currently looks more consistent with real camera-origin media.",
"likely_ai_generated": "The image has signals consistent with AI-generated or synthetic media.",
"likely_manipulated_or_deepfake": "The image has signals consistent with manipulation or deepfake-style editing.",
"inconclusive": "The analysis is inconclusive.",
}.get(label, "The analysis is inconclusive.")
def _plain_language(verdict: dict[str, Any]) -> str:
return (
f"Verdict: {verdict['label']} with {verdict['confidence']} confidence. "
f"AI evidence score is {verdict['ai_probability']:.0%}; manipulation evidence score is "
f"{verdict['manipulation_probability']:.0%}. Review the evidence layers before taking action."
)
def _clamp(value: float, lower: float = 0.0, upper: float = 1.0) -> float:
if math.isnan(value):
return 0.0
return max(lower, min(upper, float(value)))