125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
||
"""细纲 detector 的机器可读输入输出合同。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Mapping
|
||
|
||
|
||
DETECTOR_PROTOCOL = "fine_outline_detector_v0"
|
||
ALLOWED_SEVERITIES = frozenset({"high", "medium", "low"})
|
||
ALLOWED_FINDING_CATEGORIES = frozenset(
|
||
{
|
||
"candidate_structure",
|
||
"causal_chain",
|
||
"entity_state",
|
||
"foreshadowing_action",
|
||
"source_reference",
|
||
"unknowns_discipline",
|
||
}
|
||
)
|
||
ALLOWED_COVERAGE_CATEGORIES = frozenset({"frozen_context_gap"})
|
||
FINDING_FIELDS = frozenset({"category", "severity", "location", "evidenceSummary"})
|
||
COVERAGE_FINDING_FIELDS = frozenset({"category", "summary"})
|
||
|
||
|
||
def build_detector_request(
|
||
*,
|
||
candidate_id: str,
|
||
candidate: Mapping[str, Any],
|
||
as_of_chapter: int,
|
||
frozen_snapshot: Mapping[str, Any],
|
||
common_context: Mapping[str, Any],
|
||
sources: list[Any],
|
||
) -> dict[str, Any]:
|
||
"""只用公共冻结事实和匿名候选构造真正不识别评测臂的盲检输入。"""
|
||
|
||
return {
|
||
"protocol": DETECTOR_PROTOCOL,
|
||
"candidateId": candidate_id,
|
||
"asOfChapter": as_of_chapter,
|
||
"candidate": dict(candidate),
|
||
"planningContext": {
|
||
"frozenSnapshot": dict(frozen_snapshot),
|
||
"commonContext": dict(common_context),
|
||
"sources": list(sources),
|
||
},
|
||
"rules": {
|
||
"referenceProxyVisible": False,
|
||
"mayJudgeTargetRoleCardCoverage": False,
|
||
"highSeverityBlocksGroup": True,
|
||
},
|
||
}
|
||
|
||
|
||
def validate_detector_report(
|
||
report: Mapping[str, Any], expected_candidate_id: str
|
||
) -> dict[str, Any]:
|
||
"""机械校验 detector JSON；合同不完整时失败关闭。"""
|
||
|
||
errors: list[str] = []
|
||
if not isinstance(report, Mapping):
|
||
return {"ok": False, "errors": ["detector 报告必须是对象"], "highSeverityCount": 0}
|
||
if report.get("protocol") != DETECTOR_PROTOCOL:
|
||
errors.append(f"protocol 必须是 {DETECTOR_PROTOCOL}")
|
||
if report.get("candidateId") != expected_candidate_id:
|
||
errors.append("candidateId 与盲检输入不一致")
|
||
unexpected = sorted(
|
||
set(report) - {"protocol", "candidateId", "findings", "coverageFindings"}
|
||
)
|
||
if unexpected:
|
||
errors.append(f"detector 报告包含未登记字段: {','.join(unexpected)}")
|
||
|
||
findings = report.get("findings")
|
||
coverage_findings = report.get("coverageFindings")
|
||
if not isinstance(findings, list):
|
||
errors.append("findings 必须是数组")
|
||
findings = []
|
||
if not isinstance(coverage_findings, list):
|
||
errors.append("coverageFindings 必须是数组")
|
||
coverage_findings = []
|
||
|
||
sections = (
|
||
("findings", findings, ALLOWED_FINDING_CATEGORIES),
|
||
("coverageFindings", coverage_findings, ALLOWED_COVERAGE_CATEGORIES),
|
||
)
|
||
for section, items, allowed_categories in sections:
|
||
for index, finding in enumerate(items):
|
||
if not isinstance(finding, Mapping):
|
||
errors.append(f"{section}[{index}] 必须是对象")
|
||
continue
|
||
allowed_fields = FINDING_FIELDS if section == "findings" else COVERAGE_FINDING_FIELDS
|
||
unexpected_fields = sorted(set(finding) - allowed_fields)
|
||
if unexpected_fields:
|
||
errors.append(
|
||
f"{section}[{index}] 包含未登记字段: {','.join(unexpected_fields)}"
|
||
)
|
||
category = finding.get("category")
|
||
if not isinstance(category, str) or category not in allowed_categories:
|
||
errors.append(f"{section}[{index}].category 未登记")
|
||
if section == "findings":
|
||
severity = finding.get("severity")
|
||
if not isinstance(severity, str) or severity not in ALLOWED_SEVERITIES:
|
||
errors.append(f"findings[{index}].severity 无效")
|
||
for field in ("category", "location", "evidenceSummary"):
|
||
value = finding.get(field)
|
||
if not isinstance(value, str) or not value.strip():
|
||
errors.append(f"findings[{index}].{field} 必须是非空字符串")
|
||
else:
|
||
summary = finding.get("summary")
|
||
if not isinstance(summary, str) or not summary.strip():
|
||
errors.append(f"coverageFindings[{index}].summary 必须是非空字符串")
|
||
|
||
high_count = sum(
|
||
1
|
||
for finding in findings
|
||
if isinstance(finding, Mapping) and finding.get("severity") == "high"
|
||
)
|
||
return {
|
||
"ok": not errors,
|
||
"errors": errors,
|
||
"findingCount": len(findings),
|
||
"coverageFindingCount": len(coverage_findings),
|
||
"highSeverityCount": high_count,
|
||
}
|