diff --git a/src/rights_filter/analysis/internal_analyzer.py b/src/rights_filter/analysis/internal_analyzer.py index f650e3e..a1d3cb1 100644 --- a/src/rights_filter/analysis/internal_analyzer.py +++ b/src/rights_filter/analysis/internal_analyzer.py @@ -7,6 +7,7 @@ from rights_filter.analysis.face_person_detection import ( ) from rights_filter.analysis.fingerprints import FingerprintService from rights_filter.analysis.preprocessing import ImagePayload +from rights_filter.analysis.provenance import provenance_evidence from rights_filter.domain.records import ( Evidence, EvidenceSource, @@ -44,6 +45,11 @@ class InternalAnalyzer: ) ) + # Read the original bytes here, while they are still in hand: the + # external-derivative build destroys all metadata and nothing read the + # rights fields first. + evidence.extend(provenance_evidence(image.content)) + for entry in self.repository.active_knowledge_entries(): for sample in entry.sample_fingerprints: similarity = self.fingerprint_service.similarity( diff --git a/src/rights_filter/analysis/provenance.py b/src/rights_filter/analysis/provenance.py new file mode 100644 index 0000000..e9cd5d4 --- /dev/null +++ b/src/rights_filter/analysis/provenance.py @@ -0,0 +1,279 @@ +"""Where did this image come from — read from the file itself, no model. + +Two signals share one evidence source because they answer the same operator +question ("이 이미지가 어디서 왔나") and because a new EvidenceSource costs a +dozen plumbing edits; paying that once for both is the point. + +1. Rights metadata. The copyright holder often writes the answer into the file + (EXIF Copyright/Artist, IPTC CopyrightNotice/Credit/Source). The pipeline + destroys all metadata when it builds the external derivative, and nothing + read it first — so the most authoritative field available was discarded + unread on every submission. This is transcription, not inference. + +2. Screen-capture forensics. Deterministic image processing only: letterbox + bars, a subtitle band, broadcast frame sizes. No model, no weights. Tuned for + precision over recall — it ships as a zero-point label, so a miss costs + nothing while a false label costs operator trust before there is any + calibration data to correct it. + +Both are read from the ORIGINAL bytes, so the hook has to sit where the original +is still in hand (InternalAnalyzer.analyze), not after derivative building. +""" + +from __future__ import annotations + +import logging +from io import BytesIO +from typing import Any + +from rights_filter.domain.records import Evidence, EvidenceSource + +logger = logging.getLogger(__name__) + +# Operator-facing names for the rights fields. +RIGHTS_FIELD_LABELS = { + "copyright": "저작권", + "artist": "작성자", + "copyright_notice": "저작권 고지", + "credit": "크레딧", + "source": "출처", + "byline": "작성자", +} + +# EXIF tags that state rights. Deliberately short: GPS, camera serial, owner +# name and lens data are NOT read, let alone stored. The whitelist is the +# privacy boundary, so it lives as a constant and is pinned by a test. +RIGHTS_EXIF_TAGS = { + 0x8298: "copyright", + 0x013B: "artist", +} + +# IPTC IIM datasets, keyed (record, dataset). +RIGHTS_IPTC_TAGS = { + (2, 116): "copyright_notice", + (2, 110): "credit", + (2, 115): "source", + (2, 80): "byline", +} + +# Frame sizes that essentially only occur as broadcast / screen output. On their +# own they mean little, which is why geometry alone never triggers the label. +BROADCAST_FRAME_SIZES = { + (1920, 1080), + (1280, 720), + (3840, 2160), + (720, 480), + (1440, 1080), +} + +# A letterbox bar has to be this dark and this uniform to count. +LETTERBOX_MAX_MEAN = 24.0 +LETTERBOX_MAX_STD = 8.0 +# ...and cover at least this share of the frame height, top and bottom. +LETTERBOX_MIN_BAR_RATIO = 0.04 + +# Subtitle band: the bottom strip carries far more edge energy than the frame as +# a whole when burned-in text is present. +SUBTITLE_BAND_TOP = 0.72 +SUBTITLE_EDGE_RATIO = 2.5 +# Floor only — "there is actually something in the band", so an almost featureless +# frame cannot pass on a ratio computed from near-zero noise. The RATIO is the +# discriminator; a high floor here would reject real single-line subtitles, which +# occupy a small fraction of a band this tall. +SUBTITLE_MIN_EDGE_DENSITY = 0.01 + +# Independent evidence categories required before the label is emitted. Geometry +# (letterbox + broadcast size) is ONE category: a letterboxed 1920x1080 frame is +# one observation stated twice, and counting it twice would let a cinematic crop +# alone trigger the label. +MIN_CAPTURE_CATEGORIES = 2 + + +def extract_rights_metadata(content: bytes) -> dict[str, str]: + """Rights fields the author wrote into the file. Empty dict when there are none.""" + if not content: + return {} + try: + from PIL import Image + except Exception: + return {} + + found: dict[str, str] = {} + try: + with Image.open(BytesIO(content)) as opened: + found.update(_exif_rights(opened)) + found.update(_iptc_rights(opened)) + except Exception: + # Unreadable container (e.g. .svg, which the store accepts but Pillow + # cannot open) is "no metadata", not an error worth failing analysis for. + logger.debug("rights metadata extraction skipped", exc_info=True) + return {} + return {key: value for key, value in found.items() if value} + + +def _exif_rights(opened: Any) -> dict[str, str]: + try: + exif = opened.getexif() + except Exception: + return {} + if not exif: + return {} + values: dict[str, str] = {} + for tag, name in RIGHTS_EXIF_TAGS.items(): + value = exif.get(tag) + if value: + values[name] = _clean_text(value) + return values + + +def _iptc_rights(opened: Any) -> dict[str, str]: + try: + from PIL import IptcImagePlugin + + info = IptcImagePlugin.getiptcinfo(opened) + except Exception: + return {} + if not info: + return {} + values: dict[str, str] = {} + for key, name in RIGHTS_IPTC_TAGS.items(): + value = info.get(key) + if isinstance(value, (list, tuple)): + value = value[0] if value else None + if value: + values[name] = _clean_text(value) + return values + + +def _clean_text(value: Any) -> str: + if isinstance(value, bytes): + value = value.decode("utf-8", errors="replace") + return " ".join(str(value).split())[:200] + + +def detect_screen_capture(content: bytes, width: int = 0, height: int = 0) -> dict[str, Any]: + """Deterministic screen/broadcast capture signals. No model involved.""" + decoded = _decode_gray(content) + if decoded is None: + return {} + + frame_height, frame_width = decoded.shape[:2] + categories: dict[str, Any] = {} + + geometry: list[str] = [] + if _has_letterbox(decoded): + geometry.append("letterbox") + if (frame_width, frame_height) in BROADCAST_FRAME_SIZES: + geometry.append("broadcast_frame_size") + if geometry: + categories["geometry"] = geometry + + if _has_subtitle_band(decoded): + categories["subtitle_band"] = True + + if len(categories) < MIN_CAPTURE_CATEGORIES: + return {} + return { + "categories": sorted(categories), + "detail": categories, + "frame": f"{frame_width} x {frame_height}", + } + + +def _decode_gray(content: bytes) -> Any | None: + if not content: + return None + try: + import cv2 + import numpy as np + except Exception: + return None + try: + decoded = cv2.imdecode(np.frombuffer(content, dtype=np.uint8), cv2.IMREAD_GRAYSCALE) + except Exception: + return None + if decoded is None or decoded.size == 0: + return None + return decoded + + +def _has_letterbox(gray: Any) -> bool: + height = gray.shape[0] + bar = max(1, int(height * LETTERBOX_MIN_BAR_RATIO)) + if height <= bar * 3: + return False + top, bottom = gray[:bar], gray[-bar:] + return _is_flat_dark(top) and _is_flat_dark(bottom) + + +def _is_flat_dark(band: Any) -> bool: + return float(band.mean()) <= LETTERBOX_MAX_MEAN and float(band.std()) <= LETTERBOX_MAX_STD + + +def _has_subtitle_band(gray: Any) -> bool: + try: + import cv2 + except Exception: + return False + height = gray.shape[0] + start = int(height * SUBTITLE_BAND_TOP) + if height - start < 8: + return False + edges = cv2.Canny(gray, 100, 200) + band_density = float(edges[start:].mean()) / 255.0 + frame_density = float(edges.mean()) / 255.0 + if band_density < SUBTITLE_MIN_EDGE_DENSITY: + return False + if frame_density <= 0: + return False + return (band_density / frame_density) >= SUBTITLE_EDGE_RATIO + + +def provenance_evidence(content: bytes) -> list[Evidence]: + """Both provenance signals for one image, as evidence rows. + + Emitted at zero score (see RiskScorer): they ship before any calibration + data exists, so no weight would be defensible yet. The operator sees them + and decides; decision_stats later shows whether they predict rejections. + """ + evidence: list[Evidence] = [] + + rights = extract_rights_metadata(content) + if rights: + evidence.append( + Evidence( + source=EvidenceSource.PROVENANCE, + reason=_rights_reason(rights), + # Transcription, not inference — the field either is in the file + # or it is not. That the field itself may be wrong is a separate + # question, which is why it is labelled self-reported. + confidence=1.0, + data={"provenance_kind": "rights_metadata", "rights_fields": rights}, + ) + ) + + capture = detect_screen_capture(content) + if capture: + evidence.append( + Evidence( + source=EvidenceSource.PROVENANCE, + reason=f"화면/방송 캡처 추정 ({'·'.join(capture['categories'])})", + confidence=0.6, + data={ + "provenance_kind": "screen_capture", + "capture_categories": capture["categories"], + "frame": capture["frame"], + }, + ) + ) + return evidence + + +def _rights_reason(rights: dict[str, str]) -> str: + parts = [ + f"{RIGHTS_FIELD_LABELS.get(key, key)} {value}" + for key, value in rights.items() + ] + # Self-reported is not a hedge, it is the accurate description: metadata is + # trivially forgeable, so this must never read as proof of clearance. + return "원본 파일 권리 표기(자기신고): " + " / ".join(parts) diff --git a/src/rights_filter/analysis/risk_scoring.py b/src/rights_filter/analysis/risk_scoring.py index e9d4300..10e4805 100644 --- a/src/rights_filter/analysis/risk_scoring.py +++ b/src/rights_filter/analysis/risk_scoring.py @@ -77,6 +77,15 @@ class RiskScorer: reasons.append(item.reason) if bucket: counted_naver_buckets.add(bucket) + elif item.source == EvidenceSource.PROVENANCE: + # Zero-point label by design. These signals ship before any + # calibration data exists, so no weight would be defensible yet: + # measure how they correlate with real rejections in + # decision_stats first, then assign points. Surface the reason so + # the operator can act on it even at zero. + if item.reason: + reasons.append(item.reason) + continue elif item.source == EvidenceSource.LLM_SUMMARY: continue elif item.source in { diff --git a/src/rights_filter/domain/records.py b/src/rights_filter/domain/records.py index b9fd9cd..b202cc0 100644 --- a/src/rights_filter/domain/records.py +++ b/src/rights_filter/domain/records.py @@ -16,6 +16,10 @@ def _new_id(prefix: str) -> str: class EvidenceSource(StrEnum): FINGERPRINT = "fingerprint" FACE_PERSON = "face_person" + # Where the image came from, read from the file itself: rights metadata the + # author embedded, and deterministic screen-capture signals. One source for + # both because they answer the same operator question. + PROVENANCE = "provenance" WEB_DETECTION = "web_detection" NAVER_SEARCH = "naver_search" LLM_SUMMARY = "llm_summary" @@ -62,6 +66,10 @@ class DataClass(StrEnum): LLM_SUMMARY = "llm_summary" PROVIDER_METADATA = "provider_metadata" OPERATOR_NOTE = "operator_note" + # Rights fields transcribed from the submitted file (EXIF/IPTC) plus + # deterministic capture signals. Self-reported by whoever made the file, so + # it is treated as a claim to verify, never as proof. + RIGHTS_METADATA = "rights_metadata" # Whole-image semantic vector used ONLY for image-to-image similarity # (crop / retouch / re-encode detection). Not a biometric template: faces are # masked out before the vector is computed and person-dominant frames are diff --git a/src/rights_filter/governance/policies.py b/src/rights_filter/governance/policies.py index 710edb2..2819294 100644 --- a/src/rights_filter/governance/policies.py +++ b/src/rights_filter/governance/policies.py @@ -76,6 +76,12 @@ class GovernancePolicyRegistry: "delete with decision audit policy", "append correction rather than overwriting audit trail", ), + DataClass.RIGHTS_METADATA: DataClassPolicy( + {"operator", "admin"}, + "evidence retention policy", + "delete with analysis evidence policy", + "re-extract from the corrected original", + ), DataClass.SCENE_EMBEDDING: DataClassPolicy( # Operators never read raw vectors — they read the similarity # result, which is ordinary evidence. Keeping the vector out diff --git a/src/rights_filter/server/store_serialization.py b/src/rights_filter/server/store_serialization.py index fbb2506..dd5a046 100644 --- a/src/rights_filter/server/store_serialization.py +++ b/src/rights_filter/server/store_serialization.py @@ -328,6 +328,10 @@ def _evidence_payload(submission_id: str, evidence: Evidence) -> dict[str, Any]: "summary": str(evidence.data.get("summary", "")), "sourceUrls": list(evidence.data.get("source_urls", []) or []), "sourceEvidenceIds": evidence.data.get("source_evidence_ids", []), + # Provenance signal (provenance rows only; empty elsewhere). + "provenanceKind": str(evidence.data.get("provenance_kind", "")), + "rightsFields": dict(evidence.data.get("rights_fields", {}) or {}), + "captureCategories": list(evidence.data.get("capture_categories", []) or []), # Face signal (face rows only; empty elsewhere). faceTier drives the # score, so losing it on the round trip would silently revert every # background-face row to the identifiable weight on the next rescore. @@ -399,6 +403,9 @@ def _domain_evidence_from_ui(payload: dict[str, Any]) -> Evidence: "prompt_id": payload.get("promptId", ""), "prompt_version": payload.get("promptVersion", ""), "latency_ms": payload.get("latencyMs", ""), + "provenance_kind": payload.get("provenanceKind", ""), + "rights_fields": payload.get("rightsFields", {}), + "capture_categories": payload.get("captureCategories", []), "tier": payload.get("faceTier", ""), "detector": payload.get("faceDetector", ""), "face_area_ratio": payload.get("faceAreaRatio", ""), @@ -430,6 +437,8 @@ def _domain_source_from_ui_payload(payload: dict[str, Any]) -> EvidenceSource: def _domain_source_from_ui(source: str) -> EvidenceSource: + if source == "provenance": + return EvidenceSource.PROVENANCE if source == "google": return EvidenceSource.WEB_DETECTION if source == "naver": @@ -486,12 +495,17 @@ def _ui_source(source: EvidenceSource) -> str: return "failure" if source == EvidenceSource.FACE_PERSON: return "face" + if source == EvidenceSource.PROVENANCE: + return "provenance" + # NOTE: this fallback is silent. A new EvidenceSource that forgets a branch + # above lands here and is mislabelled as a fingerprint rather than failing + # loudly — add the branch when adding the source. return "fingerprint" def _ui_group(source: EvidenceSource) -> str: ui_source = _ui_source(source) - if ui_source in {"fingerprint", "face"}: + if ui_source in {"fingerprint", "face", "provenance"}: return "internal" return ui_source diff --git a/tests/rights_filter/analysis/test_provenance.py b/tests/rights_filter/analysis/test_provenance.py new file mode 100644 index 0000000..f701191 --- /dev/null +++ b/tests/rights_filter/analysis/test_provenance.py @@ -0,0 +1,182 @@ +"""Provenance signals: rights metadata transcription + capture forensics. + +Both ship as zero-point labels, so the tests that matter most are the ones about +what is NOT read (privacy whitelist) and what does NOT trigger (precision). +""" + +from io import BytesIO + +import numpy as np +import pytest +from PIL import Image + +from rights_filter.analysis.provenance import ( + MIN_CAPTURE_CATEGORIES, + RIGHTS_EXIF_TAGS, + detect_screen_capture, + extract_rights_metadata, + provenance_evidence, +) +from rights_filter.analysis.risk_scoring import RiskScorer +from rights_filter.domain.records import EvidenceSource +from rights_filter.governance.policies import assert_no_biometric_template +from rights_filter.server.store_serialization import ( + _domain_evidence_from_ui, + _evidence_payload, +) + +EXIF_COPYRIGHT = 0x8298 +EXIF_ARTIST = 0x013B +EXIF_GPS_IFD = 0x8825 +EXIF_BODY_SERIAL = 0xA431 +EXIF_MAKE = 0x010F +EXIF_MODEL = 0x0110 + + +def _jpeg(width=640, height=480, exif=None, color=128) -> bytes: + image = Image.new("RGB", (width, height), (color, color, color)) + buffer = BytesIO() + if exif is None: + image.save(buffer, format="JPEG") + else: + image.save(buffer, format="JPEG", exif=exif) + return buffer.getvalue() + + +def _exif(tags): + exif = Image.Exif() + for tag, value in tags.items(): + exif[int(tag)] = value + return exif + + +def _capture_frame(width=1280, height=720, subtitles=True, letterbox=True) -> bytes: + frame = np.full((height, width), 130, dtype=np.uint8) + # Mid-frame texture so the bottom band stands out against something. + frame[int(height * 0.2) : int(height * 0.6), :] = 120 + if letterbox: + bar = int(height * 0.08) + frame[:bar, :] = 0 + frame[-bar:, :] = 0 + if subtitles: + # Blocky high-contrast glyph-like marks in the lower band. + row = int(height * 0.82) + for start in range(120, width - 120, 26): + frame[row : row + 22, start : start + 14] = 255 + buffer = BytesIO() + Image.fromarray(frame).save(buffer, format="PNG") + return buffer.getvalue() + + +# --- rights metadata --------------------------------------------------------- + + +def test_reads_the_copyright_the_author_wrote_into_the_file(): + content = _jpeg(exif=_exif({EXIF_COPYRIGHT: "(C) 2026 Example News"})) + + assert extract_rights_metadata(content)["copyright"] == "(C) 2026 Example News" + + +def test_reads_artist_field(): + content = _jpeg(exif=_exif({EXIF_ARTIST: "Hong Gildong"})) + + assert extract_rights_metadata(content)["artist"] == "Hong Gildong" + + +def test_image_without_rights_fields_yields_nothing(): + assert extract_rights_metadata(_jpeg()) == {} + + +def test_unreadable_content_is_no_metadata_not_an_error(): + # .svg is an accepted upload suffix that Pillow cannot open. + assert extract_rights_metadata(b"") == {} + assert extract_rights_metadata(b"") == {} + + +def test_privacy_fields_are_never_read(): + # The whitelist is the privacy boundary: location and camera identity must + # not be extracted even when present in the file. + content = _jpeg( + exif=_exif( + { + EXIF_COPYRIGHT: "(C) 2026 Example", + EXIF_BODY_SERIAL: "SN-12345", + EXIF_MAKE: "ExampleCorp", + EXIF_MODEL: "X-100", + } + ) + ) + + extracted = extract_rights_metadata(content) + + # Only the rights field comes back — camera identity does not, even though + # it is sitting right next to it in the same EXIF block. + assert extracted == {"copyright": "(C) 2026 Example"} + # GPS is not written here (its tag is an IFD pointer, not a value), so pin it + # at the whitelist instead: the boundary is the constant, not the fixture. + for tag in (EXIF_GPS_IFD, EXIF_BODY_SERIAL, EXIF_MAKE, EXIF_MODEL): + assert tag not in RIGHTS_EXIF_TAGS + + +def test_rights_reason_marks_the_value_as_self_reported(): + # Metadata is trivially forgeable; the reason must never read as clearance. + evidence = provenance_evidence(_jpeg(exif=_exif({EXIF_COPYRIGHT: "(C) Example"}))) + + assert "자기신고" in evidence[0].reason + + +# --- capture forensics ------------------------------------------------------- + + +def test_letterboxed_broadcast_frame_with_subtitles_is_flagged(): + result = detect_screen_capture(_capture_frame()) + + assert result + assert len(result["categories"]) >= MIN_CAPTURE_CATEGORIES + + +def test_plain_photograph_is_not_flagged(): + assert detect_screen_capture(_jpeg(width=800, height=533)) == {} + + +def test_geometry_alone_does_not_trigger_the_label(): + # A letterboxed 1280x720 frame is one observation stated twice. Counting it + # as two categories would let any cinematic crop trigger the label. + result = detect_screen_capture(_capture_frame(subtitles=False)) + + assert result == {} + + +def test_undecodable_content_is_not_flagged(): + assert detect_screen_capture(b"not an image") == {} + + +# --- evidence, scoring, round trip ------------------------------------------- + + +def test_provenance_evidence_scores_zero_but_still_surfaces_its_reason(): + evidence = provenance_evidence(_jpeg(exif=_exif({EXIF_COPYRIGHT: "(C) Example"}))) + + result = RiskScorer().score(evidence) + + assert result.score == 0 + assert result.band == "low" + assert any("권리 표기" in reason for reason in result.reasons) + + +def test_provenance_evidence_survives_the_payload_round_trip(): + evidence = provenance_evidence(_jpeg(exif=_exif({EXIF_COPYRIGHT: "(C) Example"})))[0] + + restored = _domain_evidence_from_ui(_evidence_payload("SUB-1", evidence)) + + assert restored.source == EvidenceSource.PROVENANCE + assert restored.data["provenance_kind"] == "rights_metadata" + assert restored.data["rights_fields"]["copyright"] == "(C) Example" + assert RiskScorer().score([restored]).score == 0 + + +@pytest.mark.parametrize("kind", ["rights_metadata", "screen_capture"]) +def test_provenance_payloads_carry_no_biometric_fields(kind): + content = _jpeg(exif=_exif({EXIF_COPYRIGHT: "(C) x"})) if kind == "rights_metadata" else _capture_frame() + for evidence in provenance_evidence(content): + assert_no_biometric_template(_evidence_payload("SUB-1", evidence)) diff --git a/tests/rights_filter/test_public_module_layout.py b/tests/rights_filter/test_public_module_layout.py index 2c9be0f..6c7455c 100644 --- a/tests/rights_filter/test_public_module_layout.py +++ b/tests/rights_filter/test_public_module_layout.py @@ -18,6 +18,7 @@ def test_planned_module_boundaries_are_importable(): "rights_filter.analysis.reason_builder", "rights_filter.analysis.llm_assistance", "rights_filter.analysis.llm_prompts", + "rights_filter.analysis.provenance", "rights_filter.analysis.search_query_generation", "rights_filter.analysis.search_result_promoter", "rights_filter.integrations.external_policy", diff --git a/web/operator-gui/app.js b/web/operator-gui/app.js index e7f6860..1559620 100644 --- a/web/operator-gui/app.js +++ b/web/operator-gui/app.js @@ -1596,6 +1596,7 @@ function evidenceGroupDefinitions() { { id: "page", label: "출처 페이지", description: "같은 이미지가 포함된 웹페이지", tone: "strong" }, { id: "naver", label: "네이버 검색", description: "텍스트 기반 이미지 검색 결과", tone: "search" }, { id: "face", label: "얼굴/인물 감지", description: "동일 인물 판정은 아니며 얼굴 또는 사람 존재만 표시합니다.", tone: "local" }, + { id: "provenance", label: "출처 표기", description: "파일에 들어 있던 권리 표기와 화면 캡처 신호입니다. 자기신고 정보이며 점수에 반영하지 않습니다.", tone: "local" }, { id: "visual", label: "유사 이미지", description: "시각적으로 유사한 참고 결과", tone: "weak" }, { id: "weak", label: "약한 라벨", description: "점수에 직접 반영하지 않는 힌트", tone: "weak" }, { id: "llm", label: "내부 요약", description: "출처 연결 요약 메모", tone: "note" }, diff --git a/web/operator-gui/operator-labels.js b/web/operator-gui/operator-labels.js index a9e899a..c8631b2 100644 --- a/web/operator-gui/operator-labels.js +++ b/web/operator-gui/operator-labels.js @@ -29,6 +29,7 @@ internal: "내부", fingerprint: "이미지 유사도", face: "얼굴/인물 감지", + provenance: "출처 표기", naver: "네이버", google: "구글", llm: "내부 요약",