fix: remediate 2026-07-24 multi-agent review findings

HIGH:
- face detection: guard OpenCV builds lacking CascadeClassifier and isolate
  per-submission analysis failures in BatchAnalyzer, so a broken cv2 or one bad
  image no longer takes down every seed/reload/import/upload/rerun with no HTTP
  response
- risk scoring: gate Google custom image search (match=search_image) to 0
  points so the persistence round-trip (result_url aliased to url) can no longer
  flip the same evidence to +45 and inflate bands
- serialization: exclude SEARCH_SKIPPED/ENRICHMENT_FAILURE rows from the auto
  search dedup signatures so a quota-skipped query stays retryable instead of
  being treated as already done forever
- external_policy: add a daily-window reset + lock (mirrors SearchApiPolicy) so
  Vision is not permanently disabled once the daily limit is hit
- image_store: skip a malformed manifest row (and coerce bad width/height)
  instead of letting one ValueError paralyse loading/re-analysis for the queue

MEDIUM/LOW:
- enrichment: only count a face crop as transmitted when it actually left the
  system, removing false usage debits and a false "personal image sent to
  external provider" audit record; pick lastFailure from the failing item;
  preserve the collection set when every search was quota-skipped; dedup
  candidate counts; attribute Vision failures to the google provider badge
- store: search coverage requires a real result, not a failure row
- scoring: FACE_PERSON contributes once regardless of how many rows exist
- fingerprints: refuse near-matches on degenerate (near-uniform) average
  hashes while keeping identical-hash matches
- preprocessing: apply EXIF orientation for external derivatives, face crops,
  pHash and the PIL decode fallback
- promoter: scan title/description only, never the URL path/host tokens
- integrations: map Vision responses inside the adapter try; cap HTTPError
  bodies; skip the connect probe when a proxy applies; reservation-style quota
  accounting (try_reserve/release) to make the check-then-act atomic
- governance/admin: enforce the LLM-summary rule on the real persistence path;
  reject unknown manual-entry fields; upsert the rejected_image knowledge entry
  on repeat rejections; split policy skips out of the failures group
- env_file: strip a trailing inline comment from unquoted values

Environment drift fixed out-of-tree: removed the user-site OpenCV 5.0.0 that
shadowed the pinned opencv-python-headless==4.13.0.92.

Regression tests added; pytest tests/rights_filter is green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01E6GD9gxUtfUiVB3JHejGpU
This commit is contained in:
changukyu 2026-07-24 14:17:45 +09:00
parent 2c607e0df6
commit db9ee9a7ca
26 changed files with 946 additions and 142 deletions

3
.gitignore vendored
View file

@ -38,3 +38,6 @@ tmp_dbg_img/
# OS
.DS_Store
Thumbs.db
# PowerShell module-analysis cache written into cwd by some automation
Microsoft/

View file

@ -1,3 +1,13 @@
"""In-memory reference implementation of the operator review flow.
NOT wired into the running server: the live operator console persists through
``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
of truth for decision semantics, audit records and knowledge-entry side effects.
This module backs unit tests and documents the intended domain shape; changing
behavior here does NOT change what operators see. Keep the two in step, or fold
this into the store path, rather than treating this file as production logic.
"""
from __future__ import annotations
from rights_filter.domain.records import (

View file

@ -1,3 +1,13 @@
"""In-memory reference implementation of the operator review flow.
NOT wired into the running server: the live operator console persists through
``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
of truth for decision semantics, audit records and knowledge-entry side effects.
This module backs unit tests and documents the intended domain shape; changing
behavior here does NOT change what operators see. Keep the two in step, or fold
this into the store path, rather than treating this file as production logic.
"""
from __future__ import annotations
from typing import Any
@ -54,6 +64,7 @@ def _group_evidence(evidence: list[Evidence]) -> dict[str, list[dict[str, Any]]]
"naver": [],
"google": [],
"llm": [],
"skipped": [],
"failures": [],
}
for item in evidence:
@ -70,6 +81,11 @@ def _group_name(item: Evidence) -> str:
return "google"
if item.source == EvidenceSource.LLM_SUMMARY:
return "llm"
# A provider that was deliberately not called (quota reached, disabled) is
# operational status, not a failure — grouping the two together made a
# normally-configured case look like it had errored.
if item.source in {EvidenceSource.EXTERNAL_SKIPPED, EvidenceSource.SEARCH_SKIPPED}:
return "skipped"
return "failures"

View file

@ -1,3 +1,13 @@
"""In-memory reference implementation of the operator review flow.
NOT wired into the running server: the live operator console persists through
``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
of truth for decision semantics, audit records and knowledge-entry side effects.
This module backs unit tests and documents the intended domain shape; changing
behavior here does NOT change what operators see. Keep the two in step, or fold
this into the store path, rather than treating this file as production logic.
"""
from __future__ import annotations
from typing import Any
@ -22,7 +32,14 @@ def register_manual_entry(
sample_fingerprints: list[str] | None = None,
**payload: Any,
) -> KnowledgeBaseEntry:
# Extra keys were screened for biometric templates and then silently dropped,
# so a caller passing a field this handler does not persist got a success
# response and no entry data. Reject instead of losing the input quietly.
assert_no_biometric_template(payload)
if payload:
raise ValueError(
"unsupported knowledge entry fields: " + ", ".join(sorted(payload))
)
entry = create_manual_knowledge_entry(
entry_type=entry_type,
name=name,

View file

@ -1,3 +1,13 @@
"""In-memory reference implementation of the operator review flow.
NOT wired into the running server: the live operator console persists through
``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
of truth for decision semantics, audit records and knowledge-entry side effects.
This module backs unit tests and documents the intended domain shape; changing
behavior here does NOT change what operators see. Keep the two in step, or fold
this into the store path, rather than treating this file as production logic.
"""
from __future__ import annotations
from typing import Any

View file

@ -1,10 +1,17 @@
from __future__ import annotations
import logging
from io import BytesIO
from dataclasses import dataclass
from rights_filter.analysis.preprocessing import ImagePayload
logger = logging.getLogger(__name__)
# Emit the "OpenCV build lacks CascadeClassifier" warning at most once per
# process instead of on every image so operator logs stay readable.
_warned_missing_cascade = False
FaceBox = tuple[int, int, int, int]
@ -43,12 +50,29 @@ class HeuristicFacePersonDetector:
if decoded is None:
return FacePersonSignal(face_count=0, person_count=0)
# Some OpenCV builds (e.g. the 5.x preview wheels) drop the legacy Haar
# cascade API. Detect that up front and degrade to a presence-negative
# signal instead of letting an AttributeError escape into the seed/
# analysis pipeline and take down every ingest/re-analysis request.
if not (hasattr(cv2, "CascadeClassifier") and hasattr(cv2, "data")):
global _warned_missing_cascade
if not _warned_missing_cascade:
_warned_missing_cascade = True
logger.warning(
"OpenCV build %s lacks CascadeClassifier/data — face "
"presence detection disabled (install "
"opencv-python-headless==4.13.0.92).",
getattr(cv2, "__version__", "?"),
)
return FacePersonSignal(face_count=0, person_count=0)
try:
gray = cv2.cvtColor(decoded, cv2.COLOR_BGR2GRAY)
except Exception:
return FacePersonSignal(face_count=0, person_count=0)
face_boxes: list[FaceBox] = []
try:
for cascade_name in (
"haarcascade_frontalface_default.xml",
"haarcascade_profileface.xml",
@ -68,6 +92,10 @@ class HeuristicFacePersonDetector:
_boxes_overlap(box, existing) for existing in face_boxes
):
face_boxes.append(box)
except Exception:
# Any cascade/runtime failure degrades to a presence-negative signal
# rather than crashing the caller.
return FacePersonSignal(face_count=0, person_count=0)
face_count = len(face_boxes)
return FacePersonSignal(
@ -99,8 +127,10 @@ def _decode_with_pillow(content: bytes, cv2: object, np: object) -> object | Non
try:
from PIL import Image
with Image.open(BytesIO(content)) as image:
rgb = image.convert("RGB")
from rights_filter.analysis.preprocessing import apply_exif_orientation
with Image.open(BytesIO(content)) as opened:
rgb = apply_exif_orientation(opened).convert("RGB")
return cv2.cvtColor(np.array(rgb), cv2.COLOR_RGB2BGR)
except Exception:
return None

View file

@ -26,6 +26,19 @@ class FingerprintService:
if left_value.startswith("unavailable:") or right_value.startswith("unavailable:"):
return 0.0
if _looks_like_64bit_hash(left_value) and _looks_like_64bit_hash(right_value):
# A near-uniform image (blank, solid colour, very low detail) makes
# this 8x8 average hash collapse to nearly-all-ones or nearly-all-
# zeros. Two such hashes that differ by a bit or two would score
# ~0.98 and be reported as a knowledge-base match worth +80, even
# though neither hash carries usable information. Refuse to infer
# similarity from an uninformative hash.
#
# Identical hashes are handled above and still match: this cannot
# separate "same picture" from "two different blank images" — that
# needs the exact (SHA-256) fingerprint, which callers compare
# separately — but it does remove the near-miss false positives.
if _is_degenerate_hash(left_value) or _is_degenerate_hash(right_value):
return 0.0
distance = (int(left_value, 16) ^ int(right_value, 16)).bit_count()
return 1 - (distance / 64)
return 0.0
@ -35,8 +48,13 @@ def _perceptual_hash(content: bytes, exact_digest: str) -> str:
try:
from PIL import Image
with Image.open(BytesIO(content)) as image:
thumbnail = image.convert("L").resize((8, 8))
from rights_filter.analysis.preprocessing import apply_exif_orientation
with Image.open(BytesIO(content)) as opened:
# Normalize orientation first: the same photo saved rotated with an
# EXIF tag would otherwise hash differently from its upright copy and
# never match as a duplicate.
thumbnail = apply_exif_orientation(opened).convert("L").resize((8, 8))
if hasattr(thumbnail, "get_flattened_data"):
pixels = list(thumbnail.get_flattened_data())
else:
@ -51,6 +69,18 @@ def _perceptual_hash(content: bytes, exact_digest: str) -> str:
return f"{bits:016x}"
# A 64-bit average hash is degenerate when nearly every bit agrees: the image had
# almost no variation around its mean, so the bits describe the threshold, not the
# picture. 4 of 64 bits is the practical cut-off — below it, unrelated images
# routinely collide.
_MIN_INFORMATIVE_BITS = 4
def _is_degenerate_hash(value: str) -> bool:
set_bits = int(value, 16).bit_count()
return set_bits <= _MIN_INFORMATIVE_BITS or set_bits >= 64 - _MIN_INFORMATIVE_BITS
def _looks_like_64bit_hash(value: str) -> bool:
if len(value) != 16:
return False

View file

@ -32,23 +32,19 @@ def build_external_derivative(
if original.width <= 0 or original.height <= 0:
raise PreprocessingError("image dimensions must be positive")
longest_side = max(original.width, original.height)
if longest_side > max_side:
scale = max_side / longest_side
width = max(1, round(original.width * scale))
height = max(1, round(original.height * scale))
else:
width = original.width
height = original.height
if original.metadata.get("format", "").upper() == "AVIF":
content = _jpeg_derivative(original.content, width, height)
else:
# Re-encode through PIL to drop embedded EXIF/GPS metadata from the image
# bytes. Clearing only the metadata dict (below) left camera serials and
# GPS coordinates inside the bytes that get shipped to external APIs.
content = _reencode_without_metadata(original.content, width, height)
if content is None:
# bytes and to apply the EXIF orientation. Clearing only the metadata dict
# (below) left camera serials and GPS coordinates inside the bytes shipped to
# external APIs; skipping the orientation shipped rotated phone photos
# sideways (and hid faces from the detector, which has no EXIF awareness).
# The output size is derived from the UPRIGHT image, since a 90° rotation
# swaps width/height relative to the stored dimensions.
derived = _upright_derivative(
original.content,
max_side,
force_jpeg=original.metadata.get("format", "").upper() == "AVIF",
)
if derived is None:
# Re-encoding failed (Pillow missing or undecodable bytes): fall back to
# the original bytes, but warn — those bytes may still carry EXIF/GPS
# metadata that is about to be sent to an external API. On an air-gapped
@ -58,6 +54,9 @@ def build_external_derivative(
"metadata may not have been stripped before external submission"
)
content = original.content
width, height = _scaled_size(original.width, original.height, max_side)
else:
content, width, height = derived
return ImagePayload(
content=content,
@ -81,8 +80,10 @@ def build_face_crop_derivatives(
try:
from PIL import Image
with Image.open(BytesIO(original.content)) as image:
image = image.convert("RGB")
with Image.open(BytesIO(original.content)) as opened:
# Face boxes come from the detector, which sees the upright image —
# crop from the same orientation or the boxes address the wrong pixels.
image = apply_exif_orientation(opened).convert("RGB")
crops: list[ImagePayload] = []
image_width, image_height = image.size
for box in face_boxes[:max_crops]:
@ -111,46 +112,69 @@ def build_face_crop_derivatives(
return []
def _jpeg_derivative(content: bytes, width: int, height: int) -> bytes | None:
try:
from PIL import Image
with Image.open(BytesIO(content)) as image:
image = image.convert("RGB")
if image.size != (width, height):
image = image.resize((width, height))
output = BytesIO()
image.save(output, format="JPEG", quality=90, optimize=True)
return output.getvalue()
except Exception:
return None
_REENCODABLE_FORMATS = {"JPEG", "PNG", "WEBP", "GIF", "BMP", "TIFF"}
def _reencode_without_metadata(content: bytes, width: int, height: int) -> bytes | None:
"""Re-save the image (preserving format where possible) so embedded EXIF/GPS
metadata is dropped. Returns None when PIL is unavailable or the bytes are
not a decodable image, so the caller can fall back to the original content."""
def _scaled_size(width: int, height: int, max_side: int) -> tuple[int, int]:
longest_side = max(width, height)
if longest_side <= max_side:
return width, height
scale = max_side / longest_side
return max(1, round(width * scale)), max(1, round(height * scale))
def apply_exif_orientation(image: object) -> object:
"""Rotate/flip an opened PIL image per its EXIF orientation tag.
Every consumer that reasons about pixels (external derivative, face crops,
perceptual hash, face detection) must call this: a phone photo stored
rotated with orientation=6 otherwise reaches the detector on its side, where
upright-trained cascades miss the face entirely.
"""
try:
from PIL import ImageOps
return ImageOps.exif_transpose(image) or image
except Exception:
return image
def _upright_derivative(
content: bytes, max_side: int, force_jpeg: bool = False
) -> tuple[bytes, int, int] | None:
"""Re-save the image upright and metadata-free, capped to `max_side`.
Returns (bytes, width, height) using the orientation-corrected dimensions, or
None when PIL is unavailable / the bytes are not a decodable image so the
caller can fall back to the original content.
"""
try:
from PIL import Image
with Image.open(BytesIO(content)) as image:
source_format = (image.format or "").upper()
save_format = source_format if source_format in _REENCODABLE_FORMATS else "JPEG"
with Image.open(BytesIO(content)) as opened:
source_format = (opened.format or "").upper()
image = apply_exif_orientation(opened)
save_format = (
"JPEG"
if force_jpeg or source_format not in _REENCODABLE_FORMATS
else source_format
)
if save_format in {"JPEG", "BMP"} and image.mode not in {"RGB", "L"}:
image = image.convert("RGB")
elif image.mode == "P":
image = image.convert("RGBA")
width, height = _scaled_size(int(image.width), int(image.height), max_side)
if image.size != (width, height):
image = image.resize((width, height))
# Rebuild from raw pixels so NO metadata (EXIF/GPS/ICC) survives —
# Pillow can otherwise re-emit image.info["exif"] when re-saving.
clean = Image.frombytes(image.mode, image.size, image.tobytes())
output = BytesIO()
if save_format == "JPEG":
clean.save(output, format="JPEG", quality=90, optimize=True)
else:
clean.save(output, format=save_format)
return output.getvalue()
return output.getvalue(), width, height
except Exception:
return None

View file

@ -9,6 +9,7 @@ class RiskScorer:
reasons: list[str] = []
counted_web_buckets: set[str] = set()
counted_naver_buckets: set[str] = set()
counted_face_person = False
for item in evidence:
if item.data.get("operator_status") in {"irrelevant", "false_positive"}:
@ -24,6 +25,13 @@ class RiskScorer:
score += 30
reasons.append(item.reason)
elif item.source == EvidenceSource.FACE_PERSON:
# "A person is present" is one signal no matter how many rows
# assert it. Re-analysis appends a fresh FACE_PERSON row while the
# previous one remains, so without this bucket the same image
# accrued +35 per rerun and drifted into the high band on its own.
if counted_face_person:
continue
counted_face_person = True
score += 35
reasons.append(item.reason)
elif item.source == EvidenceSource.WEB_DETECTION:
@ -85,6 +93,16 @@ class RiskScorer:
def _web_detection_points(item: Evidence) -> int:
if item.data.get("weak_hint"):
return 0
# Google custom IMAGE search results (match="search_image") are context-only:
# existence of a text-query image hit is not itself a rights signal (the
# matching pHash FINGERPRINT evidence carries the score). Fresh evidence has
# no bare "url" key so it scores 0, but the persistence round-trip aliases
# result_url -> url; without this explicit gate the round-trip would flip the
# same evidence to +45 via the url fallback below, violating the
# score-preservation invariant and inflating bands on any query that returns
# anything.
if item.data.get("match") == "search_image":
return 0
if item.data.get("match") == "full":
return 45
if item.data.get("match") in {"partial", "page"}:

View file

@ -26,12 +26,14 @@ class SearchResultPromoter:
data={**item.data, "promoted": False, "promotion_reason": "no search results"},
)
# Scan only the actual result fields — never the query. The query is our
# own generated input; including it self-promotes any result whose query
# happened to contain a marker word.
# Scan only the human-readable result fields — never the query and never
# the URL. The query is our own generated input (it would self-promote any
# result whose query contained a marker word), and a URL's host/path
# routinely contains marker substrings for unrelated reasons
# ("/game/", "album" in a CDN path, a "cover" directory), which promoted
# irrelevant results to 0.8 confidence and +40 points.
text = " ".join(
str(item.data.get(key, ""))
for key in ("title", "description", "result_url")
str(item.data.get(key, "")) for key in ("title", "description")
).lower()
promoted = _has_named_rights_signal(text)
data = {

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from dataclasses import dataclass, field
from dataclasses import dataclass, field, replace
from enum import StrEnum
from itertools import count
from typing import Any
@ -242,9 +242,36 @@ class InMemoryRightsFilterRepository:
def create_rejected_image_entry(
self, decision_id: str, submission_id: str, fingerprints: list[str]
) -> KnowledgeBaseEntry:
name = f"rejected:{submission_id}"
# One submission has one rejection reference. Re-deciding the same case
# previously appended another entry each time, so a submission rejected
# N times seeded N duplicate reference images into every later similarity
# check. Refresh the existing entry in place instead.
incumbent = next(
(
entry
for entry in self._knowledge_entries.values()
if entry.active
and entry.entry_type == KnowledgeEntryType.REJECTED_IMAGE
and entry.name == name
),
None,
)
if incumbent is not None:
merged = list(
dict.fromkeys([*incumbent.sample_fingerprints, *fingerprints])
)
entry = replace(
incumbent,
source_decision_id=decision_id,
sample_fingerprints=merged,
)
self.save_knowledge_entry(entry)
return entry
entry = KnowledgeBaseEntry.create(
entry_type=KnowledgeEntryType.REJECTED_IMAGE,
name=f"rejected:{submission_id}",
name=name,
provenance=KnowledgeProvenance.AUTOMATIC_REJECTION,
source_decision_id=decision_id,
sample_fingerprints=fingerprints,

View file

@ -112,6 +112,10 @@ class CloudVisionWebDetectionAdapter:
source=EvidenceSource.EXTERNAL_SKIPPED,
reason=reason or "external API skipped",
confidence=1.0,
# Deliberately NOT provider-tagged: a policy refusal means the
# call never ran, and the provider badge derives "failed" from
# any attributed failure-source row. Leaving skips unattributed
# keeps the badge at "not run", which is what actually happened.
data={"submission_id": submission_id},
)
]
@ -124,14 +128,27 @@ class CloudVisionWebDetectionAdapter:
source=EvidenceSource.FAILURE,
reason=f"External API failed: {exc}",
confidence=1.0,
data={"submission_id": submission_id},
data={"submission_id": submission_id, "provider": "google"},
)
]
# Record the quota debit only after the request actually succeeds, so a
# failed/timed-out call does not permanently consume the daily limit.
policy.record_call()
try:
return _map_response(submission_id, response)
except Exception as exc:
# A malformed/unexpected response shape must not abort the caller
# after the quota was already spent — surface it as failure evidence
# like a transport error does.
return [
Evidence(
source=EvidenceSource.FAILURE,
reason=f"External API response could not be parsed: {exc}",
confidence=1.0,
data={"submission_id": submission_id, "provider": "google"},
)
]
def _map_response(submission_id: str, response: dict[str, Any]) -> list[Evidence]:
@ -208,6 +225,9 @@ def _map_response(submission_id: str, response: dict[str, Any]) -> list[Evidence
)
for label in response.get("best_guess_labels", []):
# Tolerate a bare string / non-mapping entry instead of raising on .get.
if not isinstance(label, dict):
continue
text = label.get("label", "")
evidence.append(
Evidence(

View file

@ -1,6 +1,8 @@
from __future__ import annotations
from dataclasses import dataclass
import threading
from dataclasses import dataclass, field
from datetime import date
@dataclass
@ -11,8 +13,24 @@ class ExternalApiPolicy:
allow_online_sync: bool = False
daily_limit: int | None = None
calls_made: int = 0
window_day: str = ""
_lock: threading.Lock = field(
default_factory=threading.Lock, repr=False, compare=False
)
def _roll_window_locked(self) -> None:
# daily_limit is a per-day budget. Reset the counter when the calendar
# day changes; otherwise the count only ever grows and, once the limit is
# hit, web detection stays disabled until the process restarts (it never
# re-windows for the next day). Mirrors SearchApiPolicy.
today = date.today().isoformat()
if self.window_day != today:
self.window_day = today
self.calls_made = 0
def can_call(self) -> tuple[bool, str | None]:
with self._lock:
self._roll_window_locked()
if self.disabled:
return False, "external API disabled"
if not self.compliance_approved:
@ -26,4 +44,6 @@ class ExternalApiPolicy:
return True, None
def record_call(self) -> None:
with self._lock:
self._roll_window_locked()
self.calls_made += 1

View file

@ -102,7 +102,7 @@ class GoogleCustomSearchAdapter:
raise ValueError("Google custom search requires a text query")
requested_calls = _requested_calls(self.client, "image_pages")
allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@ -119,9 +119,15 @@ class GoogleCustomSearchAdapter:
)
]
try:
try:
response = self.client.search_image(query)
policy.record_call(requested_calls)
except Exception:
# The request never reached the provider — refund the
# reservation. A failure AFTER a successful fetch keeps the
# debit, since the provider did serve the call.
policy.release(requested_calls)
raise
# Map inside the try so a malformed payload becomes ENRICHMENT_FAILURE
# evidence instead of an uncaught exception that aborts the request.
return _map_image_response(submission_id, query, response)
@ -148,7 +154,7 @@ class GoogleCustomSearchAdapter:
raise ValueError("Google custom search requires a text query")
requested_calls = _requested_calls(self.client, "web_pages")
allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@ -165,9 +171,15 @@ class GoogleCustomSearchAdapter:
)
]
try:
try:
response = self.client.search_web(query)
policy.record_call(requested_calls)
except Exception:
# The request never reached the provider — refund the
# reservation. A failure AFTER a successful fetch keeps the
# debit, since the provider did serve the call.
policy.release(requested_calls)
raise
return _map_web_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [

View file

@ -5,7 +5,17 @@ import socket
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen
from urllib.request import Request, getproxies, proxy_bypass, urlopen
def _proxy_applies(scheme: str, host: str) -> bool:
"""True when the system proxy configuration would route `host` for `scheme`."""
try:
if scheme not in getproxies():
return False
return not proxy_bypass(host)
except Exception:
return False
class HttpJsonError(RuntimeError):
@ -16,6 +26,10 @@ class HttpJsonError(RuntimeError):
# cannot exhaust memory. JSON API responses are small; 10 MiB is generous.
_MAX_RESPONSE_BYTES = 10 * 1024 * 1024
# An HTTP error body only ever becomes part of an exception message, so keep it
# small — the full 10 MiB cap would put a whole error page into a log line.
_MAX_ERROR_DETAIL_BYTES = 4096
class UrllibJsonTransport:
def __init__(self, connect_timeout: float | None = None) -> None:
@ -52,7 +66,9 @@ class UrllibJsonTransport:
raw = raw_bytes.decode("utf-8", errors="replace")
except HTTPError as exc:
try:
detail = exc.read().decode("utf-8", errors="replace")
# Apply the same cap on the error path: an error page can be far
# larger than a JSON reply, and it is only used for a message.
detail = exc.read(_MAX_ERROR_DETAIL_BYTES).decode("utf-8", errors="replace")
except Exception:
detail = ""
raise HttpJsonError(f"HTTP {exc.code}: {detail}") from exc
@ -73,6 +89,12 @@ class UrllibJsonTransport:
host = parts.hostname
if not host:
return
# A direct TCP probe bypasses the system proxy. On a proxy-only network
# the probe always fails while the actual (proxied) request would have
# succeeded, so skip the fast-fail check when a proxy applies to this host
# and let urlopen route through it.
if _proxy_applies(parts.scheme, host):
return
port = parts.port or (443 if parts.scheme == "https" else 80)
try:
with socket.create_connection((host, port), timeout=self.connect_timeout):

View file

@ -162,7 +162,7 @@ class NaverSearchAdapter:
raise ValueError("Naver search requires a text query")
requested_calls = _requested_calls(self.client, "image_pages")
allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@ -178,9 +178,15 @@ class NaverSearchAdapter:
)
]
try:
try:
response = self.client.search_image(query)
policy.record_call(requested_calls)
except Exception:
# The request never reached the provider — refund the
# reservation. A failure AFTER a successful fetch keeps the
# debit, since the provider did serve the call.
policy.release(requested_calls)
raise
return _map_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
@ -204,7 +210,7 @@ class NaverSearchAdapter:
raise ValueError("Naver search requires a text query")
requested_calls = _requested_calls(self.client, "blog_pages")
allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@ -221,9 +227,15 @@ class NaverSearchAdapter:
)
]
try:
try:
response = self.client.search_blog(query)
policy.record_call(requested_calls)
except Exception:
# The request never reached the provider — refund the
# reservation. A failure AFTER a successful fetch keeps the
# debit, since the provider did serve the call.
policy.release(requested_calls)
raise
return _map_blog_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
@ -248,7 +260,7 @@ class NaverSearchAdapter:
raise ValueError("Naver search requires a text query")
requested_calls = _requested_calls(self.client, "web_pages")
allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@ -265,9 +277,15 @@ class NaverSearchAdapter:
)
]
try:
try:
response = self.client.search_web(query)
policy.record_call(requested_calls)
except Exception:
# The request never reached the provider — refund the
# reservation. A failure AFTER a successful fetch keeps the
# debit, since the provider did serve the call.
policy.release(requested_calls)
raise
return _map_web_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [

View file

@ -27,7 +27,36 @@ class SearchApiPolicy:
self.window_day = today
self.calls_made = 0
def try_reserve(self, provider: str, requested_calls: int = 1) -> tuple[bool, str | None]:
"""Atomically check the budget and debit it in one locked step.
Checking and debiting separately let concurrent operator requests both
pass the check before either recorded, overshooting daily_limit by the
number of racing callers. Callers refund with release() when the request
never actually reached the provider.
"""
with self._lock:
self._roll_window_locked()
if self.disabled:
return False, "search API disabled"
if not self.compliance_approved:
return False, "search API compliance not approved"
if provider not in self.allowed_providers:
return False, "search provider not allowed"
calls = max(1, int(requested_calls or 1))
if self.daily_limit is not None and self.calls_made + calls > self.daily_limit:
return False, "search API usage limit reached"
self.calls_made += calls
return True, None
def release(self, count: int = 1) -> None:
"""Return an unused reservation (the request never reached the provider)."""
with self._lock:
self._roll_window_locked()
self.calls_made = max(0, self.calls_made - max(1, int(count or 1)))
def can_call(self, provider: str, requested_calls: int = 1) -> tuple[bool, str | None]:
"""Non-reserving budget check, for callers that only need to look."""
with self._lock:
self._roll_window_locked()
if self.disabled:

View file

@ -1,5 +1,6 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from rights_filter.analysis.internal_analyzer import InternalAnalyzer
@ -14,6 +15,8 @@ from rights_filter.domain.records import (
from rights_filter.integrations.cloud_vision_web_detection import CloudVisionWebDetectionAdapter
from rights_filter.integrations.external_policy import ExternalApiPolicy
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class SubmissionImage:
@ -86,6 +89,17 @@ class BatchAnalyzer:
# has_analysis_run() skip the submission permanently on re-runs.
failed += 1
continue
except Exception:
# Isolate any unexpected per-submission failure (e.g. an OpenCV
# build missing CascadeClassifier, a decoder edge case) so one
# bad image cannot abort the whole batch/seed. Log with the
# submission id for diagnosis instead of swallowing silently.
logger.exception(
"Unexpected analysis failure for submission %s; skipping.",
submission.submission_id,
)
failed += 1
continue
external_skipped += sum(
1 for item in evidence if item.source == EvidenceSource.EXTERNAL_SKIPPED

View file

@ -24,10 +24,25 @@ def load_env_file(path: Path | str, environ: MutableMapping[str, str]) -> None:
key = key.strip()
if not key or key in environ:
continue
environ[key] = _strip_quotes(value.strip())
environ[key] = _parse_value(value.strip())
def _strip_quotes(value: str) -> str:
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
def _parse_value(value: str) -> str:
if _is_quoted(value):
# Quoted values are verbatim — a '#' inside quotes is part of the secret.
return value[1:-1]
return _strip_inline_comment(value)
def _is_quoted(value: str) -> bool:
return len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}
def _strip_inline_comment(value: str) -> str:
# ` #` starts a trailing comment on an unquoted value (dotenv convention).
# Without this, `TOKEN=abc123 # staging key` silently loaded the comment as
# part of the token and every authenticated request failed with no clue why.
marker = value.find(" #")
if marker == -1:
return value
return value[:marker].rstrip()

View file

@ -1,11 +1,14 @@
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Any
from rights_filter.analysis.preprocessing import ImagePayload
logger = logging.getLogger(__name__)
SUPPORTED_IMAGE_SUFFIXES = {
".avif",
".bmp",
@ -30,13 +33,23 @@ class LocalSubmissionImageStore:
manifest = self.root / "submissions.json"
if manifest.exists():
records = json.loads(manifest.read_text(encoding="utf-8"))
normalized = [
self._normalize_manifest_record(record)
for record in records
if isinstance(record, dict)
and record.get("file")
and self._safe_path(str(record["file"])).exists()
]
normalized: list[dict[str, Any]] = []
for record in records:
if not isinstance(record, dict) or not record.get("file"):
continue
# Skip a malformed row (e.g. a path escaping the image root)
# instead of letting its ValueError abort the whole listing —
# one bad manifest line previously paralyzed loading and
# re-analysis for the entire queue.
try:
if not self._safe_path(str(record["file"])).exists():
continue
normalized.append(self._normalize_manifest_record(record))
except (ValueError, OSError):
logger.warning(
"Skipping malformed submissions.json record: file=%r",
record.get("file"),
)
return self._with_unlisted_files(normalized)
return self._scan_records()
@ -45,8 +58,8 @@ class LocalSubmissionImageStore:
path = self._safe_path(record["file"])
return ImagePayload(
content=path.read_bytes(),
width=int(record["width"]),
height=int(record["height"]),
width=_safe_dimension(record.get("width")),
height=_safe_dimension(record.get("height")),
metadata={
"submission_id": submission_id,
"title": str(record.get("title", submission_id)),
@ -127,6 +140,15 @@ class LocalSubmissionImageStore:
return path
def _safe_dimension(value: Any) -> int:
# A manifest may carry a non-numeric or missing width/height; fall back to
# the same placeholder the normalizer uses rather than raising mid-analysis.
try:
return int(value)
except (TypeError, ValueError):
return 1
def _image_size(path: Path) -> tuple[int, int]:
try:
from PIL import Image

View file

@ -395,12 +395,21 @@ class StoreEnrichmentMixin:
if not crops:
continue
crop = crops[0]
call_count += 1
crop_evidence = self.provider_runtime.google_adapter.detect(
f"{submission_id}:face-crop-{crop_index}",
crop,
self.provider_runtime.external_policy,
)
# Only count a crop as transmitted when the adapter actually issued a
# request. A policy refusal returns EXTERNAL_SKIPPED without sending
# anything, so counting it inflated provider usage and — worse —
# produced a false "personal image sent to external provider" audit
# record for biometric data that never left the system. A FAILURE
# still counts: the request may have been sent before it failed.
if any(
item.source != EvidenceSource.EXTERNAL_SKIPPED for item in crop_evidence
):
call_count += 1
for item in crop_evidence:
face_web_evidence = _face_crop_web_evidence(submission_id, crop_index, item)
evidence.append(face_web_evidence)

View file

@ -10,12 +10,14 @@ and self.* attributes provided by the host class. Behavior unchanged.
from __future__ import annotations
import json
import logging
import sqlite3
from typing import Any
from rights_filter.analysis.fingerprints import FingerprintService
from rights_filter.analysis.search_result_promoter import SearchResultPromoter
from rights_filter.domain.records import Evidence, EvidenceSource
from rights_filter.domain.records import DataClass, Evidence, EvidenceSource
from rights_filter.governance.policies import assert_operator_evidence_payload_allowed
from rights_filter.server.image_store import LocalSubmissionImageStore
from rights_filter.server.store_serialization import (
_domain_evidence_from_ui,
@ -32,6 +34,8 @@ from rights_filter.server.store_serialization import (
)
from rights_filter.server.store_text import _text_list, _unique_texts
logger = logging.getLogger(__name__)
class StoreOperationsMixin:
def promote_watchlist_entry(self, entry_id: str) -> dict[str, Any]:
@ -228,6 +232,8 @@ class StoreOperationsMixin:
}
self._put("evidence", evidence["id"], {**evidence, "submission_id": submission_id})
rerun_error = ""
try:
if image_store is not None:
self._rerun_internal_analysis(submission_id, image_store)
google_evidence = self._rerun_google_image_search(submission_id, image_store)
@ -237,6 +243,20 @@ class StoreOperationsMixin:
]
self._auto_naver_search(submission_id, query_source_evidence, image_store)
self._auto_google_custom_search(submission_id, query_source_evidence, image_store)
except Exception as exc:
# A submission whose image file went missing (or any other enrichment
# failure) must not escape as an unhandled exception: the handler
# would close the connection with no response while this method has
# already written lastAnalysis and the "queued" marker, leaving the
# case permanently marked as re-analyzing. Record the failure, keep
# the rescore/return path intact, and let the marker cleanup below run.
rerun_error = str(exc) or exc.__class__.__name__
logger.exception("Rerun enrichment failed for submission %s", submission_id)
finally:
# The rerun is synchronous, so by the time we return there is nothing
# "queued" — drop the marker instead of accumulating one dead row per
# rerun (previously they piled up in the operator evidence list).
self._delete_evidence_by_id(evidence["id"])
self._ensure_llm_summary(submission_id)
@ -268,6 +288,7 @@ class StoreOperationsMixin:
refreshed = self._get("submissions", submission_id)
refreshed["lastRerunDiff"] = {
"at": _now_label(),
"error": rerun_error,
"scoreBefore": score_before,
"scoreAfter": int(refreshed.get("riskScore", 0) or 0),
"addedEvidenceIds": added_ids,
@ -356,7 +377,18 @@ class StoreOperationsMixin:
item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
for item in domain_evidence
):
provider_payload["lastFailure"] = domain_evidence[0].reason if domain_evidence else f"{provider} search failed"
# Report the reason of the item that actually failed. Indexing
# [0] picked whatever came first — usually a successful result —
# so the operator saw a success string labelled as the failure.
provider_payload["lastFailure"] = next(
(
item.reason
for item in domain_evidence
if item.source
in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
),
f"{provider} search failed",
)
else:
provider_payload["lastSuccess"] = _now_label()
provider_payload["lastFailure"] = "없음"
@ -403,6 +435,25 @@ class StoreOperationsMixin:
submission_id,
source_evidence,
)
# Enforce the LLM-summary governance rule on the real persistence path.
# The policy existed but no production caller invoked it, so an
# ungrounded summary (no source URLs and no source evidence ids) could be
# stored and shown as if it were source-linked. A summary that fails the
# check is dropped rather than persisted.
if llm_evidence.source == EvidenceSource.LLM_SUMMARY:
try:
assert_operator_evidence_payload_allowed(
DataClass.LLM_SUMMARY, llm_evidence.data
)
except ValueError as exc:
logger.warning(
"Discarding ungrounded LLM summary for submission %s: %s",
submission_id,
exc,
)
llm_provider["lastFailure"] = f"LLM assistance failed: {exc}"
self._apply_provider_usage_delta("llm", 1, llm_provider)
return False
self._delete_llm_summary_evidence(submission_id)
self._put(
"evidence",
@ -417,6 +468,10 @@ class StoreOperationsMixin:
self._apply_provider_usage_delta("llm", 1, llm_provider)
return True
def _delete_evidence_by_id(self, evidence_id: str) -> None:
with self._connect() as conn:
conn.execute("delete from evidence where id = ?", (evidence_id,))
def _delete_llm_summary_evidence(self, submission_id: str) -> None:
with self._connect() as conn:
conn.execute(
@ -657,15 +712,42 @@ class StoreOperationsMixin:
item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
for item in domain_evidence
):
provider_payload["lastFailure"] = domain_evidence[0].reason if domain_evidence else f"{provider} collection failed"
provider_payload["lastFailure"] = next(
(
item.reason
for item in domain_evidence
if item.source
in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
),
f"{provider} collection failed",
)
else:
provider_payload["lastSuccess"] = _now_label()
provider_payload["lastFailure"] = "없음"
# The cascade stages can surface the same candidate id more than once
# (e.g. an image result that also appears in the page sweep). Persisting
# is keyed by id, so counting every append reported more "collected"
# candidates than rows actually stored.
deduped_candidates = list({candidate["id"]: candidate for candidate in pending}.values())
collected = len(deduped_candidates)
# Every search was refused before it ran (quota/disabled provider), so we
# learned nothing. Wiping the previous collection here would destroy the
# operator's working set as punishment for exhausted quota.
if not deduped_candidates and provider_call_count == 0:
# Still persist why it was refused (zero delta merges lastFailure
# without charging quota) so the provider panel explains the no-op.
self._apply_provider_usage_delta(provider, 0, provider_payload)
payload = self.bootstrap()
payload["collected"] = 0
payload["collectionSkipped"] = True
return payload
with self._write_lock:
with self._transaction() as conn:
conn.execute("delete from collection_candidates")
for candidate in pending:
for candidate in deduped_candidates:
self._put("collection_candidates", candidate["id"], candidate, conn=conn)
self.add_audit_event(
"rights.ops",

View file

@ -186,6 +186,13 @@ def _evidence_matches_provider(evidence: dict[str, Any], provider_id: str) -> bo
if provider_id == "naver":
return source == "naver" or domain in {"naver", "naver_blog", "naver_web"}
if provider_id == "google":
# Vision failure/skip rows serialize with source "failure", so matching on
# source == "google" alone left them unattributed: the badge showed "not
# run" for a provider that had actually failed, while the score still
# carried the failure reason. Rows written before the adapter tagged the
# provider fall back to domain "internal" and are matched by title.
if source == "failure" and domain == "google":
return True
return source == "google" and domain != "google_custom_search"
if provider_id == "google_search":
return domain == "google_custom_search"
@ -478,11 +485,29 @@ def _validate_payload(table: str, id_value: str, payload: dict[str, Any]) -> Non
raise ValueError(f"{table} payload id does not match storage id")
# A quota-skip (SEARCH_SKIPPED) or transient failure (ENRICHMENT_FAILURE) means
# the query never actually executed / produced no evidence, so it must NOT enter
# the dedup signature set — otherwise auto-search treats it as "already done" and
# never retries it, even after quota resets or the network recovers. Genuine
# "returned no results" rows keep their real provider source (naver/google) and
# stay deduped correctly (they did run).
_RETRYABLE_SKIP_DOMAIN_SOURCES = {
EvidenceSource.SEARCH_SKIPPED.value,
EvidenceSource.ENRICHMENT_FAILURE.value,
}
def _is_retryable_skip_evidence(item: dict[str, Any]) -> bool:
return str(item.get("domainSource", "")) in _RETRYABLE_SKIP_DOMAIN_SOURCES
def _existing_naver_query_signatures(evidence: list[dict[str, Any]]) -> set[str]:
return {
str(item.get("querySignature") or _naver_query_signature(str(item.get("query", ""))))
for item in evidence
if item.get("query") and item.get("source") in {"naver", "failure"}
if item.get("query")
and item.get("source") in {"naver", "failure"}
and not _is_retryable_skip_evidence(item)
}
@ -494,6 +519,7 @@ def _existing_google_custom_query_signatures(evidence: list[dict[str, Any]]) ->
and item.get("source") in {"google", "failure"}
and str(item.get("domain", "")) == "google_custom_search"
and item.get("querySignature")
and not _is_retryable_skip_evidence(item)
}

View file

@ -65,7 +65,10 @@ def test_detailed_review_groups_operator_evidence_and_actions():
assert len(review["evidence_groups"]["naver"]) == 1
assert len(review["evidence_groups"]["google"]) == 1
assert len(review["evidence_groups"]["llm"]) == 1
assert len(review["evidence_groups"]["failures"]) == 1
# A provider that was skipped by policy is operational status, not an error:
# it must not be presented alongside genuine failures.
assert len(review["evidence_groups"]["skipped"]) == 1
assert "failures" not in review["evidence_groups"]
def test_missing_analysis_returns_unavailable_review_model():

View file

@ -101,7 +101,16 @@ def test_local_image_store_skips_manifest_records_with_missing_files(tmp_path: P
assert [record["id"] for record in records] == ["copied"]
def test_local_image_store_rejects_manifest_paths_outside_root(tmp_path: Path):
def test_local_image_store_skips_manifest_paths_outside_root_without_losing_valid_records(
tmp_path: Path,
):
# An out-of-root path must never be surfaced as a submission, but it also must
# not abort the whole listing: a single malformed manifest row previously
# raised ValueError and paralyzed loading/re-analysis for the entire queue.
image_dir = tmp_path / "images"
image_dir.mkdir()
(image_dir / "good.svg").write_text("<svg></svg>", encoding="utf-8")
(tmp_path.parent / "secret.svg").write_text("<svg>secret</svg>", encoding="utf-8")
(tmp_path / "submissions.json").write_text(
"""
[
@ -111,17 +120,22 @@ def test_local_image_store_rejects_manifest_paths_outside_root(tmp_path: Path):
"file": "../secret.svg",
"width": 100,
"height": 100
},
{
"id": "SUB-GOOD",
"title": "Good sample",
"file": "images/good.svg",
"width": 640,
"height": 480
}
]
""",
encoding="utf-8",
)
store = LocalSubmissionImageStore(tmp_path)
records = LocalSubmissionImageStore(tmp_path).submission_records()
try:
store.submission_records()
except ValueError as exc:
assert "outside image store" in str(exc)
else:
raise AssertionError("expected unsafe manifest path to be rejected")
ids = [record["id"] for record in records]
assert "SUB-BAD" not in ids
assert "SUB-GOOD" in ids
assert all("secret" not in str(record["file"]) for record in records)

View file

@ -0,0 +1,311 @@
"""Regression tests for the 2026-07-24 multi-agent review remediation.
One test per confirmed finding, named after the behaviour it locks in.
"""
from pathlib import Path
import pytest
from rights_filter.analysis.fingerprints import FingerprintService
from rights_filter.analysis.risk_scoring import RiskScorer
from rights_filter.analysis.search_result_promoter import SearchResultPromoter
from rights_filter.domain.records import Evidence, EvidenceSource
from rights_filter.integrations.external_policy import ExternalApiPolicy
from rights_filter.server.env_file import load_env_file
from rights_filter.server.image_store import LocalSubmissionImageStore
from rights_filter.server.store_serialization import (
_domain_evidence_from_ui,
_evidence_matches_provider,
_evidence_payload,
_existing_naver_query_signatures,
)
# --- H1: an OpenCV build without CascadeClassifier must not crash analysis ----
def test_face_detection_degrades_when_opencv_lacks_cascade_classifier(monkeypatch):
import rights_filter.analysis.face_person_detection as detection
class _Cv2WithoutCascades:
IMREAD_COLOR = 1
COLOR_BGR2GRAY = 6
def imdecode(self, *_args):
return [[0]]
def cvtColor(self, *_args):
return [[0]]
monkeypatch.setitem(__import__("sys").modules, "cv2", _Cv2WithoutCascades())
detection._warned_missing_cascade = False
signal = detection.HeuristicFacePersonDetector()._detect_with_opencv(b"bytes")
assert signal.face_count == 0
assert signal.present is False
def test_batch_analyzer_isolates_an_unexpected_per_submission_failure():
from rights_filter.analysis.preprocessing import ImagePayload
from rights_filter.domain.records import InMemoryRightsFilterRepository
from rights_filter.integrations.cloud_vision_web_detection import (
CloudVisionWebDetectionAdapter,
FakeWebDetectionClient,
)
from rights_filter.jobs.batch_analyzer import BatchAnalyzer, SubmissionImage
class _ExplodingAnalyzer:
def analyze(self, submission_id, image):
if submission_id == "BAD":
raise AttributeError("module 'cv2' has no attribute 'CascadeClassifier'")
return []
batch = BatchAnalyzer(
repository=InMemoryRightsFilterRepository(),
internal_analyzer=_ExplodingAnalyzer(),
external_adapter=CloudVisionWebDetectionAdapter(FakeWebDetectionClient()),
external_policy=ExternalApiPolicy(disabled=True),
scorer=RiskScorer(),
)
image = ImagePayload(content=b"x", width=10, height=10, metadata={})
summary = batch.run(
[SubmissionImage("BAD", image), SubmissionImage("GOOD", image)]
)
# The bad submission is counted, and the good one still gets analysed.
assert summary.failed == 1
assert summary.processed == 1
# --- H2: the persistence round-trip must not invent risk points --------------
def test_custom_image_search_result_scores_the_same_before_and_after_round_trip():
fresh = Evidence(
source=EvidenceSource.WEB_DETECTION,
reason="Google custom image search result found",
confidence=0.45,
data={
"submission_id": "SUB-1",
"provider": "google_custom_search",
"search_type": "image",
"query": "test",
"result_url": "https://example.test/page",
"match": "search_image",
},
)
scorer = RiskScorer()
before = scorer.score([fresh]).score
restored = _domain_evidence_from_ui(_evidence_payload("SUB-1", fresh))
after = scorer.score([restored]).score
assert before == 0
assert after == before
# --- H3: a quota-skipped query must stay retryable ---------------------------
def test_quota_skipped_query_is_not_registered_as_already_searched():
skipped = Evidence(
source=EvidenceSource.SEARCH_SKIPPED,
reason="search API usage limit reached",
confidence=1.0,
data={
"submission_id": "SUB-1",
"provider": "naver",
"query": "IU album",
"query_signature": "naver:iu album",
},
)
executed = Evidence(
source=EvidenceSource.NAVER_SEARCH,
reason="Naver search returned no results",
confidence=0.0,
data={
"submission_id": "SUB-1",
"provider": "naver",
"query": "other query",
"query_signature": "naver:other query",
},
)
payloads = [
_evidence_payload("SUB-1", skipped),
_evidence_payload("SUB-1", executed),
]
signatures = _existing_naver_query_signatures(payloads)
assert "naver:iu album" not in signatures
# A query that DID run (even with zero results) stays deduped.
assert "naver:other query" in signatures
# --- H5: the external daily limit must re-window each day --------------------
def test_external_policy_resets_its_daily_budget_on_a_new_day():
policy = ExternalApiPolicy(
compliance_approved=True,
metadata_logging_accepted=True,
allow_online_sync=True,
daily_limit=1,
)
policy.record_call()
assert policy.can_call()[0] is False
policy.window_day = "2000-01-01" # simulate the calendar day rolling over
assert policy.can_call()[0] is True
# --- H6: one malformed manifest row must not paralyse the queue --------------
def test_manifest_with_a_bad_row_still_lists_the_valid_submissions(tmp_path: Path):
(tmp_path / "images").mkdir()
(tmp_path / "images" / "ok.svg").write_text("<svg></svg>", encoding="utf-8")
(tmp_path / "submissions.json").write_text(
'[{"id":"BAD","file":"../outside.svg","width":1,"height":1},'
'{"id":"OK","file":"images/ok.svg","width":1,"height":1}]',
encoding="utf-8",
)
records = LocalSubmissionImageStore(tmp_path).submission_records()
assert [record["id"] for record in records] == ["OK"]
def test_non_numeric_manifest_dimensions_do_not_raise(tmp_path: Path):
(tmp_path / "images").mkdir()
(tmp_path / "images" / "ok.svg").write_text("<svg></svg>", encoding="utf-8")
(tmp_path / "submissions.json").write_text(
'[{"id":"OK","file":"images/ok.svg","width":"wide","height":null}]',
encoding="utf-8",
)
payload = LocalSubmissionImageStore(tmp_path).image_payload("OK")
assert payload.width == 1
assert payload.height == 1
# --- MEDIUM/LOW --------------------------------------------------------------
def test_vision_failure_is_attributed_to_the_google_provider():
failure = Evidence(
source=EvidenceSource.FAILURE,
reason="External API failed: timeout",
confidence=1.0,
data={"submission_id": "SUB-1", "provider": "google"},
)
payload = _evidence_payload("SUB-1", failure)
assert _evidence_matches_provider(payload, "google") is True
def test_face_person_evidence_counts_once_however_many_rows_exist():
rows = [
Evidence(
source=EvidenceSource.FACE_PERSON,
reason="Face or person detected",
confidence=1.0,
data={"submission_id": "SUB-1"},
)
for _ in range(3)
]
assert RiskScorer().score(rows).score == 35
def test_degenerate_perceptual_hashes_do_not_produce_a_near_match():
service = FingerprintService()
# Two "blank image" hashes differing by one bit: 0.98 similarity under a raw
# Hamming comparison, which used to register as a knowledge-base match.
assert service.similarity("phash:ffffffffffffffff", "phash:fffffffffffffffe") == 0.0
# Identical fingerprints still match, and informative hashes still compare.
assert service.similarity("phash:ffffffffffffffff", "phash:ffffffffffffffff") == 1.0
assert service.similarity("phash:0f0f0f0f0f0f0f0f", "phash:0f0f0f0f0f0f0f0f") == 1.0
def test_promoter_ignores_marker_words_that_only_appear_in_the_url():
promoter = SearchResultPromoter()
item = Evidence(
source=EvidenceSource.NAVER_SEARCH,
reason="Naver search result found",
confidence=0.5,
data={
"submission_id": "SUB-1",
"title": "동네 산책 사진",
"description": "주말 나들이 기록",
"result_url": "https://cdn.example.test/game/album/cover/img.jpg",
},
)
promoted = promoter.promote([item])[0]
assert promoted.data["promoted"] is False
def test_env_file_strips_a_trailing_inline_comment(tmp_path: Path):
env_path = tmp_path / ".env"
env_path.write_text(
'COPYRIGHTER_AUTH_TOKEN=abc123 # staging key\nQUOTED="keep # this"\n',
encoding="utf-8",
)
environ: dict[str, str] = {}
load_env_file(env_path, environ)
assert environ["COPYRIGHTER_AUTH_TOKEN"] == "abc123"
assert environ["QUOTED"] == "keep # this"
def test_search_policy_reservation_is_atomic_and_refundable():
from rights_filter.integrations.search_policy import SearchApiPolicy
policy = SearchApiPolicy(
compliance_approved=True, allowed_providers={"naver"}, daily_limit=1
)
assert policy.try_reserve("naver")[0] is True
# The budget is consumed at reservation time, so a racing caller is refused.
assert policy.try_reserve("naver")[0] is False
policy.release()
assert policy.try_reserve("naver")[0] is True
def test_rejected_image_knowledge_entry_is_reused_across_repeat_rejections():
from rights_filter.domain.records import InMemoryRightsFilterRepository
repository = InMemoryRightsFilterRepository()
first = repository.create_rejected_image_entry("dec-1", "SUB-1", ["phash:aaaa"])
second = repository.create_rejected_image_entry("dec-2", "SUB-1", ["phash:bbbb"])
assert first.id == second.id
assert second.source_decision_id == "dec-2"
assert set(second.sample_fingerprints) == {"phash:aaaa", "phash:bbbb"}
def test_manual_knowledge_registration_rejects_unknown_fields():
from rights_filter.admin.knowledge_base_handlers import register_manual_entry
from rights_filter.domain.records import (
InMemoryRightsFilterRepository,
KnowledgeEntryType,
)
with pytest.raises(ValueError, match="unsupported"):
register_manual_entry(
InMemoryRightsFilterRepository(),
KnowledgeEntryType.CELEBRITY,
"name",
surprise="dropped silently before",
)