feat: 출처 신호 2종 (EXIF/IPTC 권리 표기 + 방송캡처 포렌식), 0점 라벨
두 신호를 하나의 EvidenceSource.PROVENANCE 와 하나의 GUI 그룹으로 묶는다.
운영자 관점에서 둘 다 같은 질문("이 이미지가 어디서 왔나")에 답하고, 새
EvidenceSource 는 배관 12곳을 요구하므로 둘을 합쳐 한 번만 지불한다.
## EXIF/IPTC 권리 표기
저작권자가 파일에 직접 써 넣은 값을 지금까지 100% 버리고 있었다.
build_external_derivative 가 raw 픽셀로 재구성하며 메타를 전량 파기하는데,
그 전에 읽는 코드가 없었다. 추출 훅은 원본 바이트가 아직 살아 있는
InternalAnalyzer.analyze 진입부에 둔다.
읽는 것은 EXIF Copyright/Artist 와 IPTC CopyrightNotice/Credit/Source/By-line
뿐이다. 화이트리스트가 프라이버시 경계이므로 상수로 두고 테스트로 고정한다 —
GPS, 바디 시리얼, Make/Model 은 같은 EXIF 블록에 나란히 있어도 읽지 않는다.
메타데이터는 위조가 쉬우므로 reason 에 "자기신고" 를 명시한다. 이것이
클리어런스 증명으로 읽히면 안 된다.
XMP/C2PA 는 이번 범위에서 뺀다 — Pillow 의 getxmp() 는 defusedxml 이 없으면
경고와 함께 빈 dict 를 반환하고, defusedxml 은 requirements.txt 에 없다.
폐쇄망 타깃에서 조용히 no-op 이 되므로 별도 반입 심사 안건이다.
## 방송캡처 포렌식
결정적 영상처리만 사용한다(레터박스, 자막 밴드 엣지 밀도, 방송 프레임 크기).
모델 0, 신규 의존성 0.
레터박스와 방송 프레임 크기는 하나의 카테고리(geometry)로 묶는다. 레터박스된
1280x720 은 한 관측을 두 번 말한 것이고, 따로 세면 시네마틱 크롭 하나로
라벨이 붙는다. 서로 독립인 카테고리가 2개 이상일 때만 발화한다.
## 0점
둘 다 점수 0으로 출시한다(사용자 결정). 캘리브레이션 데이터가 없는 상태에서
정당화할 수 있는 가중치가 없다 — decision_stats 의 밴드x판정 교차표로 실제
반려와의 상관을 본 뒤 부여한다. reason 은 노출하므로 0점이어도 운영자가
바로 활용할 수 있다.
## 실측
실제 제출 이미지 6장: 캡처 오탐 0건, 권리 메타 히트 0건. 정밀도는 확인됐고
가치는 이 코퍼스에서 아직 입증되지 않았다 — 메신저/SNS 를 거친 이미지는 메타가
이미 제거돼 있어 히트율이 낮을 것이라는 사전 지적과 일치한다.
자막 밴드 절대 밀도 하한은 0.02 -> 0.01. 비율(밴드/프레임 = 2.5배)이 판별자이고
절대 하한은 "밴드에 뭔가 있긴 한가" 게이트일 뿐이며, 하한이 높으면 실제 한
줄짜리 자막이 걸러진다. 합성 픽스처 기반이므로 실제 캡처 이미지 확보 시
재확인이 필요하다.
pytest 531 passed.
This commit is contained in:
parent
6bbaaf6e23
commit
5a9fcc0bea
10 changed files with 508 additions and 1 deletions
|
|
@ -7,6 +7,7 @@ from rights_filter.analysis.face_person_detection import (
|
||||||
)
|
)
|
||||||
from rights_filter.analysis.fingerprints import FingerprintService
|
from rights_filter.analysis.fingerprints import FingerprintService
|
||||||
from rights_filter.analysis.preprocessing import ImagePayload
|
from rights_filter.analysis.preprocessing import ImagePayload
|
||||||
|
from rights_filter.analysis.provenance import provenance_evidence
|
||||||
from rights_filter.domain.records import (
|
from rights_filter.domain.records import (
|
||||||
Evidence,
|
Evidence,
|
||||||
EvidenceSource,
|
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 entry in self.repository.active_knowledge_entries():
|
||||||
for sample in entry.sample_fingerprints:
|
for sample in entry.sample_fingerprints:
|
||||||
similarity = self.fingerprint_service.similarity(
|
similarity = self.fingerprint_service.similarity(
|
||||||
|
|
|
||||||
279
src/rights_filter/analysis/provenance.py
Normal file
279
src/rights_filter/analysis/provenance.py
Normal file
|
|
@ -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)
|
||||||
|
|
@ -77,6 +77,15 @@ class RiskScorer:
|
||||||
reasons.append(item.reason)
|
reasons.append(item.reason)
|
||||||
if bucket:
|
if bucket:
|
||||||
counted_naver_buckets.add(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:
|
elif item.source == EvidenceSource.LLM_SUMMARY:
|
||||||
continue
|
continue
|
||||||
elif item.source in {
|
elif item.source in {
|
||||||
|
|
|
||||||
|
|
@ -16,6 +16,10 @@ def _new_id(prefix: str) -> str:
|
||||||
class EvidenceSource(StrEnum):
|
class EvidenceSource(StrEnum):
|
||||||
FINGERPRINT = "fingerprint"
|
FINGERPRINT = "fingerprint"
|
||||||
FACE_PERSON = "face_person"
|
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"
|
WEB_DETECTION = "web_detection"
|
||||||
NAVER_SEARCH = "naver_search"
|
NAVER_SEARCH = "naver_search"
|
||||||
LLM_SUMMARY = "llm_summary"
|
LLM_SUMMARY = "llm_summary"
|
||||||
|
|
@ -62,6 +66,10 @@ class DataClass(StrEnum):
|
||||||
LLM_SUMMARY = "llm_summary"
|
LLM_SUMMARY = "llm_summary"
|
||||||
PROVIDER_METADATA = "provider_metadata"
|
PROVIDER_METADATA = "provider_metadata"
|
||||||
OPERATOR_NOTE = "operator_note"
|
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
|
# Whole-image semantic vector used ONLY for image-to-image similarity
|
||||||
# (crop / retouch / re-encode detection). Not a biometric template: faces are
|
# (crop / retouch / re-encode detection). Not a biometric template: faces are
|
||||||
# masked out before the vector is computed and person-dominant frames are
|
# masked out before the vector is computed and person-dominant frames are
|
||||||
|
|
|
||||||
|
|
@ -76,6 +76,12 @@ class GovernancePolicyRegistry:
|
||||||
"delete with decision audit policy",
|
"delete with decision audit policy",
|
||||||
"append correction rather than overwriting audit trail",
|
"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(
|
DataClass.SCENE_EMBEDDING: DataClassPolicy(
|
||||||
# Operators never read raw vectors — they read the similarity
|
# Operators never read raw vectors — they read the similarity
|
||||||
# result, which is ordinary evidence. Keeping the vector out
|
# result, which is ordinary evidence. Keeping the vector out
|
||||||
|
|
|
||||||
|
|
@ -328,6 +328,10 @@ def _evidence_payload(submission_id: str, evidence: Evidence) -> dict[str, Any]:
|
||||||
"summary": str(evidence.data.get("summary", "")),
|
"summary": str(evidence.data.get("summary", "")),
|
||||||
"sourceUrls": list(evidence.data.get("source_urls", []) or []),
|
"sourceUrls": list(evidence.data.get("source_urls", []) or []),
|
||||||
"sourceEvidenceIds": evidence.data.get("source_evidence_ids", []),
|
"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
|
# Face signal (face rows only; empty elsewhere). faceTier drives the
|
||||||
# score, so losing it on the round trip would silently revert every
|
# score, so losing it on the round trip would silently revert every
|
||||||
# background-face row to the identifiable weight on the next rescore.
|
# 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_id": payload.get("promptId", ""),
|
||||||
"prompt_version": payload.get("promptVersion", ""),
|
"prompt_version": payload.get("promptVersion", ""),
|
||||||
"latency_ms": payload.get("latencyMs", ""),
|
"latency_ms": payload.get("latencyMs", ""),
|
||||||
|
"provenance_kind": payload.get("provenanceKind", ""),
|
||||||
|
"rights_fields": payload.get("rightsFields", {}),
|
||||||
|
"capture_categories": payload.get("captureCategories", []),
|
||||||
"tier": payload.get("faceTier", ""),
|
"tier": payload.get("faceTier", ""),
|
||||||
"detector": payload.get("faceDetector", ""),
|
"detector": payload.get("faceDetector", ""),
|
||||||
"face_area_ratio": payload.get("faceAreaRatio", ""),
|
"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:
|
def _domain_source_from_ui(source: str) -> EvidenceSource:
|
||||||
|
if source == "provenance":
|
||||||
|
return EvidenceSource.PROVENANCE
|
||||||
if source == "google":
|
if source == "google":
|
||||||
return EvidenceSource.WEB_DETECTION
|
return EvidenceSource.WEB_DETECTION
|
||||||
if source == "naver":
|
if source == "naver":
|
||||||
|
|
@ -486,12 +495,17 @@ def _ui_source(source: EvidenceSource) -> str:
|
||||||
return "failure"
|
return "failure"
|
||||||
if source == EvidenceSource.FACE_PERSON:
|
if source == EvidenceSource.FACE_PERSON:
|
||||||
return "face"
|
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"
|
return "fingerprint"
|
||||||
|
|
||||||
|
|
||||||
def _ui_group(source: EvidenceSource) -> str:
|
def _ui_group(source: EvidenceSource) -> str:
|
||||||
ui_source = _ui_source(source)
|
ui_source = _ui_source(source)
|
||||||
if ui_source in {"fingerprint", "face"}:
|
if ui_source in {"fingerprint", "face", "provenance"}:
|
||||||
return "internal"
|
return "internal"
|
||||||
return ui_source
|
return ui_source
|
||||||
|
|
||||||
|
|
|
||||||
182
tests/rights_filter/analysis/test_provenance.py
Normal file
182
tests/rights_filter/analysis/test_provenance.py
Normal file
|
|
@ -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"<svg xmlns='http://www.w3.org/2000/svg'/>") == {}
|
||||||
|
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))
|
||||||
|
|
@ -18,6 +18,7 @@ def test_planned_module_boundaries_are_importable():
|
||||||
"rights_filter.analysis.reason_builder",
|
"rights_filter.analysis.reason_builder",
|
||||||
"rights_filter.analysis.llm_assistance",
|
"rights_filter.analysis.llm_assistance",
|
||||||
"rights_filter.analysis.llm_prompts",
|
"rights_filter.analysis.llm_prompts",
|
||||||
|
"rights_filter.analysis.provenance",
|
||||||
"rights_filter.analysis.search_query_generation",
|
"rights_filter.analysis.search_query_generation",
|
||||||
"rights_filter.analysis.search_result_promoter",
|
"rights_filter.analysis.search_result_promoter",
|
||||||
"rights_filter.integrations.external_policy",
|
"rights_filter.integrations.external_policy",
|
||||||
|
|
|
||||||
|
|
@ -1596,6 +1596,7 @@ function evidenceGroupDefinitions() {
|
||||||
{ id: "page", label: "출처 페이지", description: "같은 이미지가 포함된 웹페이지", tone: "strong" },
|
{ id: "page", label: "출처 페이지", description: "같은 이미지가 포함된 웹페이지", tone: "strong" },
|
||||||
{ id: "naver", label: "네이버 검색", description: "텍스트 기반 이미지 검색 결과", tone: "search" },
|
{ id: "naver", label: "네이버 검색", description: "텍스트 기반 이미지 검색 결과", tone: "search" },
|
||||||
{ id: "face", label: "얼굴/인물 감지", description: "동일 인물 판정은 아니며 얼굴 또는 사람 존재만 표시합니다.", tone: "local" },
|
{ id: "face", label: "얼굴/인물 감지", description: "동일 인물 판정은 아니며 얼굴 또는 사람 존재만 표시합니다.", tone: "local" },
|
||||||
|
{ id: "provenance", label: "출처 표기", description: "파일에 들어 있던 권리 표기와 화면 캡처 신호입니다. 자기신고 정보이며 점수에 반영하지 않습니다.", tone: "local" },
|
||||||
{ id: "visual", label: "유사 이미지", description: "시각적으로 유사한 참고 결과", tone: "weak" },
|
{ id: "visual", label: "유사 이미지", description: "시각적으로 유사한 참고 결과", tone: "weak" },
|
||||||
{ id: "weak", label: "약한 라벨", description: "점수에 직접 반영하지 않는 힌트", tone: "weak" },
|
{ id: "weak", label: "약한 라벨", description: "점수에 직접 반영하지 않는 힌트", tone: "weak" },
|
||||||
{ id: "llm", label: "내부 요약", description: "출처 연결 요약 메모", tone: "note" },
|
{ id: "llm", label: "내부 요약", description: "출처 연결 요약 메모", tone: "note" },
|
||||||
|
|
|
||||||
|
|
@ -29,6 +29,7 @@
|
||||||
internal: "내부",
|
internal: "내부",
|
||||||
fingerprint: "이미지 유사도",
|
fingerprint: "이미지 유사도",
|
||||||
face: "얼굴/인물 감지",
|
face: "얼굴/인물 감지",
|
||||||
|
provenance: "출처 표기",
|
||||||
naver: "네이버",
|
naver: "네이버",
|
||||||
google: "구글",
|
google: "구글",
|
||||||
llm: "내부 요약",
|
llm: "내부 요약",
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue