122 lines
3.7 KiB
Python
122 lines
3.7 KiB
Python
"""Evidence validation for Diagnosis (PRD §31.2.1 / §35.2).
|
|
|
|
Evidence must appear in the corresponding Original Text. Unmatched
|
|
evidence is discarded so the frontend never paints fictional anchors.
|
|
Patterns without any verified evidence are kept in the diagnosis table
|
|
but lose their evidence list (no fake highlights).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
from schemas import Annotation, DiagnoseResponse, Pattern
|
|
|
|
_WS = re.compile(r"\s+")
|
|
|
|
PATTERN_CATEGORY: dict[str, str] = {
|
|
"P01": "growth",
|
|
"P02": "growth",
|
|
"P03": "rhetoric",
|
|
"P04": "repeat",
|
|
"P05": "rhetoric",
|
|
"P06": "repeat",
|
|
"P07": "growth",
|
|
"P08": "structure",
|
|
"P09": "growth",
|
|
"P10": "growth",
|
|
"P11": "repeat",
|
|
"P12": "rhetoric",
|
|
}
|
|
|
|
KIND_LABEL: dict[str, str] = {
|
|
"rhetoric": "修辞包装",
|
|
"repeat": "重复 / 工整",
|
|
"growth": "抽象 / 总结",
|
|
"structure": "结构 / 路标",
|
|
}
|
|
|
|
VALID_CATEGORIES = set(KIND_LABEL)
|
|
|
|
|
|
def normalize(s: str) -> str:
|
|
return _WS.sub(" ", s).strip().lower()
|
|
|
|
|
|
def evidence_in_text(evidence: str, text: str) -> bool:
|
|
ev = normalize(evidence)
|
|
if not ev:
|
|
return False
|
|
return ev in normalize(text)
|
|
|
|
|
|
def verified_evidence(items: list[str], texts: list[str]) -> list[str]:
|
|
haystacks = [normalize(t) for t in texts]
|
|
hits: list[str] = []
|
|
seen: set[str] = set()
|
|
for raw in items:
|
|
ev = normalize(raw)
|
|
if not ev or ev in seen:
|
|
continue
|
|
if any(ev in h for h in haystacks):
|
|
hits.append(raw)
|
|
seen.add(ev)
|
|
return hits
|
|
|
|
|
|
def category_of(pattern_id: str, fallback: str = "") -> str:
|
|
cat = (fallback or "").strip().lower()
|
|
if cat in VALID_CATEGORIES:
|
|
return cat
|
|
return PATTERN_CATEGORY.get((pattern_id or "").upper(), "rhetoric")
|
|
|
|
|
|
def sanitize_patterns(patterns: list[Pattern], paragraphs: list[str]) -> list[Pattern]:
|
|
"""Keep patterns, but drop evidence that cannot be found in the original."""
|
|
out: list[Pattern] = []
|
|
for pat in patterns:
|
|
hits = verified_evidence(pat.evidence, paragraphs) if pat.evidence else []
|
|
cat = category_of(pat.pattern_id, pat.category)
|
|
out.append(pat.model_copy(update={"evidence": hits, "category": cat}))
|
|
return out
|
|
|
|
|
|
def sanitize_annotation(note: Annotation, paragraph_text: str, index: int) -> Annotation | None:
|
|
hits = verified_evidence(note.evidence, [paragraph_text]) if note.evidence else []
|
|
if not hits:
|
|
return None
|
|
cat = category_of(note.pattern_id, note.category)
|
|
return note.model_copy(
|
|
update={
|
|
"annotation_id": note.annotation_id or f"a{index + 1}",
|
|
"evidence": hits,
|
|
"category": cat,
|
|
"kind_label": note.kind_label or KIND_LABEL.get(cat, "修辞包装"),
|
|
}
|
|
)
|
|
|
|
|
|
def sanitize_diagnosis(resp: DiagnoseResponse, paragraphs: list[str]) -> DiagnoseResponse:
|
|
"""Filter fictional evidence and cap per-paragraph annotations at 3."""
|
|
patterns = sanitize_patterns(resp.patterns, paragraphs)
|
|
briefs = []
|
|
for brief in resp.paragraph_briefs:
|
|
idx = _paragraph_index(brief.paragraph_id)
|
|
text = paragraphs[idx] if 0 <= idx < len(paragraphs) else ""
|
|
notes: list[Annotation] = []
|
|
for i, note in enumerate(brief.annotations):
|
|
cleaned = sanitize_annotation(note, text, i)
|
|
if cleaned:
|
|
notes.append(cleaned)
|
|
if len(notes) >= 3:
|
|
break
|
|
briefs.append(brief.model_copy(update={"annotations": notes}))
|
|
return resp.model_copy(update={"patterns": patterns, "paragraph_briefs": briefs})
|
|
|
|
|
|
def _paragraph_index(pid: str) -> int:
|
|
digits = "".join(ch for ch in (pid or "") if ch.isdigit())
|
|
if not digits:
|
|
return -1
|
|
return int(digits) - 1
|