fix: restore scoring signals lost in the evidence DB round-trip

Production scores come from _rescore_submission, which re-reads evidence via
_domain_evidence_from_ui and re-scores. Serialization dropped the promoted /
entity fields and the precise EvidenceSource, so on every rescore Naver
promotion (+40) and Google web-entity (+40) silently scored 0 and skip
evidence was mis-restored as FAILURE (+30 phantom).

- store_serialization: persist domainSource / promoted / promotionReason /
  entity / category in _evidence_payload and restore them in
  _domain_evidence_from_ui; resolve the source from the persisted enum
  (keyword heuristic kept only as a legacy fallback).
- search_result_promoter: structural no-result detection (covers image/blog/
  web variants), scan only result fields not the query, add Korean rights
  markers (excluding over-generic substrings).
- risk_scoring: dedup promoted Naver hits by query_signature; drop zero-point
  skip reasons from the operator-facing reasons list.

Tests: round-trip regression (promoted naver / web entity / two distinct
entities score > 0; skip adds no phantom points) + skip-reason exclusion.
This commit is contained in:
changukyu 2026-07-11 16:06:05 +09:00
parent 89c1fe4788
commit ac47f0c5d0
4 changed files with 134 additions and 29 deletions

View file

@ -8,6 +8,7 @@ class RiskScorer:
score = 0 score = 0
reasons: list[str] = [] reasons: list[str] = []
counted_web_buckets: set[str] = set() counted_web_buckets: set[str] = set()
counted_naver_buckets: set[str] = set()
for item in evidence: for item in evidence:
if item.data.get("operator_status") in {"irrelevant", "false_positive"}: if item.data.get("operator_status") in {"irrelevant", "false_positive"}:
@ -37,9 +38,18 @@ class RiskScorer:
counted_web_buckets.add(bucket) counted_web_buckets.add(bucket)
elif item.source == EvidenceSource.NAVER_SEARCH: elif item.source == EvidenceSource.NAVER_SEARCH:
points = _naver_search_points(item) points = _naver_search_points(item)
if not points:
continue
# Count one promoted hit per executed query. Multiple promoted
# results from the same search are one signal, not N stacked +40s
# (which would push a single query straight to the high band).
bucket = _naver_search_bucket(item)
if bucket and bucket in counted_naver_buckets:
continue
score += points score += points
if points:
reasons.append(item.reason) reasons.append(item.reason)
if bucket:
counted_naver_buckets.add(bucket)
elif item.source == EvidenceSource.LLM_SUMMARY: elif item.source == EvidenceSource.LLM_SUMMARY:
continue continue
elif item.source in { elif item.source in {
@ -54,13 +64,12 @@ class RiskScorer:
# e.g. "External API failed") still contribute the +30 signal. # e.g. "External API failed") still contribute the +30 signal.
if item.reason.startswith("LLM assistance failed"): if item.reason.startswith("LLM assistance failed"):
continue continue
score += ( if item.source in {EvidenceSource.FAILURE, EvidenceSource.ENRICHMENT_FAILURE}:
30 score += 30
if item.source
in {EvidenceSource.FAILURE, EvidenceSource.ENRICHMENT_FAILURE}
else 0
)
reasons.append(item.reason) reasons.append(item.reason)
# SEARCH_SKIPPED / EXTERNAL_SKIPPED contribute zero points and are
# operational status ("quota reached", "provider disabled"), not
# risk signals — keep them out of the operator-facing reasons list.
score = min(100, score) score = min(100, score)
if score >= 70: if score >= 70:
@ -113,6 +122,16 @@ def _naver_search_points(item: Evidence) -> int:
return 0 return 0
def _naver_search_bucket(item: Evidence) -> str:
# Dedup promoted Naver hits by the query that produced them. Distinct queries
# remain distinct signals; repeated hits from one query collapse to one.
signature = str(item.data.get("query_signature", "")).strip()
if signature:
return f"naver:{signature}"
query = str(item.data.get("query", "")).strip().casefold()
return f"naver-query:{query}" if query else ""
def _unique(values: list[str]) -> list[str]: def _unique(values: list[str]) -> list[str]:
result: list[str] = [] result: list[str] = []
for value in values: for value in values:

View file

@ -14,12 +14,24 @@ class SearchResultPromoter:
if item.source != EvidenceSource.NAVER_SEARCH: if item.source != EvidenceSource.NAVER_SEARCH:
return item return item
if item.reason == "Naver search returned no results": # Structural no-result detection covers the image/blog/web variants
return replace(item, confidence=0.0, data={**item.data, "promoted": False}) # (each emits a different "... returned no results" reason). A real hit
# always carries at least one result field; a no-result item carries
# none, so it must never be promoted even if its query text contained a
# rights marker.
if not _has_result_content(item.data):
return replace(
item,
confidence=0.0,
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.
text = " ".join( text = " ".join(
str(item.data.get(key, "")) str(item.data.get(key, ""))
for key in ("title", "description", "query", "result_url") for key in ("title", "description", "result_url")
).lower() ).lower()
promoted = _has_named_rights_signal(text) promoted = _has_named_rights_signal(text)
data = { data = {
@ -32,9 +44,15 @@ class SearchResultPromoter:
return replace(item, confidence=0.8 if promoted else 0.2, data=data) return replace(item, confidence=0.8 if promoted else 0.2, data=data)
def _has_named_rights_signal(text: str) -> bool: def _has_result_content(data: dict) -> bool:
markers = { return any(data.get(key) for key in ("title", "description", "result_url", "image_url"))
"iu",
# Markers that indicate a named person / work / rights context. English markers
# plus the Korean equivalents actually emitted by Naver results and by this
# project's own query templates, since Naver is a Korean-language provider and an
# English-only list produces systematic false negatives.
_RIGHTS_MARKERS = {
"official", "official",
"album", "album",
"cover", "cover",
@ -45,8 +63,33 @@ def _has_named_rights_signal(text: str) -> bool:
"drama", "drama",
"copyright", "copyright",
"celebrity", "celebrity",
"goods",
"공식",
"앨범",
"커버",
"캐릭터",
"웹툰",
"화보",
"저작권",
"연예인",
"아이돌",
"굿즈",
"포스터",
# NOTE: intentionally NOT including bare '게임'/'방송' — as CJK substrings they
# over-match benign Korean text ('방송통신위원회', casual game blogs) and
# over-promote. Their English counterparts (game/broadcast) stay, matched on
# word boundaries.
} }
# Match whole words only. Plain `in` matching let short markers like "iu"
# or "cover" hit inside unrelated words/URLs ("premium", "discover"),
# falsely promoting context-only results. def _has_named_rights_signal(text: str) -> bool:
return any(re.search(rf"\b{re.escape(marker)}\b", text) for marker in markers) # ASCII markers match whole words only (so short ones like "cover" don't hit
# inside "discover"); CJK markers have no word boundaries in regex, so match
# them as substrings (Korean has no spaces between a marker and its context).
for marker in _RIGHTS_MARKERS:
if marker.isascii():
if re.search(rf"\b{re.escape(marker)}\b", text):
return True
elif marker in text:
return True
return False

View file

@ -261,6 +261,16 @@ def _evidence_payload(submission_id: str, evidence: Evidence) -> dict[str, Any]:
"id": _evidence_id(submission_id, evidence), "id": _evidence_id(submission_id, evidence),
"group": "watchlist" if knowledge_entry_status == "watchlist" else "face_web" if face_crop_search else _ui_group(evidence.source), "group": "watchlist" if knowledge_entry_status == "watchlist" else "face_web" if face_crop_search else _ui_group(evidence.source),
"source": source, "source": source,
# Persist the precise domain source and the scoring-relevant data fields
# so the production round-trip (_domain_evidence_from_ui -> RiskScorer in
# _rescore_submission) reproduces the original score. Without these the
# coarse "source" string collapses SEARCH_SKIPPED/FAILURE together and the
# promoted/entity/category signals vanish, silently zeroing risk points.
"domainSource": evidence.source.value,
"promoted": bool(evidence.data.get("promoted", False)),
"promotionReason": str(evidence.data.get("promotion_reason", "")),
"entity": str(evidence.data.get("entity", "")),
"category": str(evidence.data.get("category", "")),
"title": evidence.reason, "title": evidence.reason,
"confidence": evidence.confidence, "confidence": evidence.confidence,
"query": str(evidence.data.get("query", "")), "query": str(evidence.data.get("query", "")),
@ -327,6 +337,10 @@ def _domain_evidence_from_ui(payload: dict[str, Any]) -> Evidence:
"match": payload.get("matchType", ""), "match": payload.get("matchType", ""),
"rank": payload.get("rank", ""), "rank": payload.get("rank", ""),
"score": payload.get("providerScore", ""), "score": payload.get("providerScore", ""),
"promoted": bool(payload.get("promoted", False)),
"promotion_reason": payload.get("promotionReason", ""),
"entity": payload.get("entity", ""),
"category": payload.get("category", ""),
"contributed": payload.get("contributed", True), "contributed": payload.get("contributed", True),
"status": payload.get("status", ""), "status": payload.get("status", ""),
"weak_hint": bool(payload.get("faceCropSearch", False)) or _is_google_weak_label_payload(payload), "weak_hint": bool(payload.get("faceCropSearch", False)) or _is_google_weak_label_payload(payload),
@ -344,6 +358,17 @@ def _domain_evidence_from_ui(payload: dict[str, Any]) -> Evidence:
def _domain_source_from_ui_payload(payload: dict[str, Any]) -> EvidenceSource: def _domain_source_from_ui_payload(payload: dict[str, Any]) -> EvidenceSource:
# Prefer the persisted precise source. This distinguishes SEARCH_SKIPPED
# (0 points) from FAILURE (+30) instead of guessing from title keywords,
# which previously misclassified skip evidence as failure on every rescore.
domain_source = str(payload.get("domainSource", ""))
if domain_source:
try:
return EvidenceSource(domain_source)
except ValueError:
pass
# Legacy fallback for rows written before domainSource was persisted.
source = str(payload.get("source", "")) source = str(payload.get("source", ""))
title = str(payload.get("title", "")).lower() title = str(payload.get("title", "")).lower()
if source == "failure" and ( if source == "failure" and (

View file

@ -280,3 +280,21 @@ def test_same_web_entity_repeated_does_not_double_count():
assert result.score == 45 assert result.score == 45
assert result.band == "medium" assert result.band == "medium"
def test_search_skipped_reason_is_not_listed_as_a_risk_reason():
# 2026-07-11 #16: a zero-point skip ("quota reached") is operational status,
# not a risk signal, and must not pollute the operator-facing reasons list.
result = RiskScorer().score(
[
Evidence(
source=EvidenceSource.SEARCH_SKIPPED,
reason="search API usage limit reached",
confidence=1.0,
data={},
),
]
)
assert result.score == 0
assert result.reasons == []