import argparse
|
import csv
|
import hashlib
|
from collections import Counter
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
|
|
def read_csv(path):
|
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
return list(csv.DictReader(handle))
|
|
|
def write_csv(path, fieldnames, rows):
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=fieldnames)
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def sha256_file(path):
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
def classify_gap(row):
|
value_status = row.get("value_consistency_result", "")
|
access_status = row.get("source_access_status", "")
|
source_name = row.get("source_name", "")
|
if value_status == "NOT_VERIFIED_CANDIDATE_DATE_AMBIGUOUS":
|
return {
|
"gap_type": "REPORT_DATE_AND_TABLE_ROW_GAP",
|
"gap_reason": "candidate fact uses relative date wording; source page is locatable but exact report week/date and original table row are unresolved",
|
"required_fix": "resolve report publication date or report week, then map to source table/download row and compare value/unit",
|
"queue_priority": "HIGH",
|
"queue_status": "QUEUE_REPORT_DATE_RESOLUTION",
|
"next_action": "resolve_report_date_then_map_source_row",
|
}
|
if value_status == "NOT_VERIFIED_SOURCE_ACCESS_HELD" or access_status == "HELD_BY_WEB_VERIFICATION_PAGE":
|
return {
|
"gap_type": "SOURCE_ACCESS_OR_ARCHIVE_GAP",
|
"gap_reason": "source entry is official but current access is blocked by verification page or missing archived table/download",
|
"required_fix": "retry source download/archive through approved path, capture table/download evidence, or replace with auditable equivalent source",
|
"queue_priority": "HIGH",
|
"queue_status": "QUEUE_SOURCE_ARCHIVE_OR_REPLACEMENT",
|
"next_action": "retry_archive_or_find_auditable_replacement_source",
|
}
|
if value_status == "NOT_VERIFIED_VALUE_ACCESS_HELD" or access_status == "PARTIAL_ACCESS_VALUE_SIGNIN_REQUIRED":
|
return {
|
"gap_type": "VALUE_ACCESS_HELD_GAP",
|
"gap_reason": "source page/specification is locatable but value requires login, screenshot, or replacement source",
|
"required_fix": "obtain access confirmation, archive screenshot/table row, or replace with public auditable source",
|
"queue_priority": "HIGH",
|
"queue_status": "QUEUE_VALUE_ACCESS_CONFIRMATION",
|
"next_action": "confirm_value_access_or_replace_source",
|
}
|
return {
|
"gap_type": "SOURCE_VERIFY_REVIEW_GAP",
|
"gap_reason": f"unhandled source verification state for {source_name}",
|
"required_fix": "manual review required",
|
"queue_priority": row.get("priority", "MEDIUM") or "MEDIUM",
|
"queue_status": "QUEUE_MANUAL_REVIEW",
|
"next_action": "manual_review_source_verify_result",
|
}
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--verify-result-input", required=True)
|
parser.add_argument("--output", required=True)
|
parser.add_argument("--manifest", required=True)
|
parser.add_argument("--summary", required=True)
|
parser.add_argument("--run-id", default="RUN-ANA-YS-SOURCE-GAP-CONFLICT-011")
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
result_rows = read_csv(project_root / args.verify_result_input)
|
created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
rows = []
|
for row in result_rows:
|
gap = classify_gap(row)
|
rows.append(
|
{
|
"source_gap_conflict_id": f"YS-SOURCE-GAP-011-{len(rows) + 1:04d}",
|
"case_id": row.get("case_id", ""),
|
"batch_id": row.get("batch_id", ""),
|
"run_id": args.run_id,
|
"source_verify_result_id": row.get("source_verify_result_id", ""),
|
"source_verify_id": row.get("source_verify_id", ""),
|
"source_supplement_id": row.get("source_supplement_id", ""),
|
"gap_priority_id": row.get("gap_priority_id", ""),
|
"evidence_card_id": row.get("evidence_card_id", ""),
|
"fact_id": row.get("fact_id", ""),
|
"doc_id": row.get("doc_id", ""),
|
"metal_tags": row.get("metal_tags", ""),
|
"theme_tags": row.get("theme_tags", ""),
|
"source_name": row.get("source_name", ""),
|
"source_url": row.get("source_url", ""),
|
"verified_source_url": row.get("verified_source_url", ""),
|
"source_access_status": row.get("source_access_status", ""),
|
"source_row_status": row.get("source_row_status", ""),
|
"value_consistency_result": row.get("value_consistency_result", ""),
|
"gap_type": gap["gap_type"],
|
"gap_reason": gap["gap_reason"],
|
"required_fix": gap["required_fix"],
|
"queue_priority": gap["queue_priority"],
|
"queue_status": gap["queue_status"],
|
"value_unit_candidates": row.get("value_unit_candidates", ""),
|
"date_candidates": row.get("date_candidates", ""),
|
"next_action": gap["next_action"],
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
)
|
|
output_path = project_root / args.output
|
fields = list(rows[0].keys()) if rows else ["source_gap_conflict_id", "case_id", "run_id", "review_status"]
|
write_csv(output_path, fields, rows)
|
output_sha = sha256_file(output_path)
|
manifest_path = project_root / args.manifest
|
write_csv(
|
manifest_path,
|
["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"],
|
[
|
{
|
"case_id": "ANA-YS-INDUSTRY-001",
|
"batch_id": "BATCH-001+BATCH-003",
|
"run_id": args.run_id,
|
"artifact_type": "source_gap_conflict_queue",
|
"artifact_path": output_path.relative_to(project_root).as_posix(),
|
"row_count": str(len(rows)),
|
"sha256": output_sha,
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
],
|
)
|
|
gap_counts = Counter(row["gap_type"] for row in rows)
|
queue_counts = Counter(row["queue_status"] for row in rows)
|
source_counts = Counter(row["source_name"] for row in rows)
|
summary_path = project_root / args.summary
|
lines = [
|
"# 来源缺口/冲突队列 PASS-011 摘要",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{created_at}",
|
"",
|
"## 输出",
|
"",
|
f"- 缺口/冲突队列：`{output_path.relative_to(project_root).as_posix()}`",
|
f"- manifest：`{manifest_path.relative_to(project_root).as_posix()}`",
|
f"- 记录数：{len(rows)}",
|
f"- sha256：`{output_sha}`",
|
"",
|
"## 缺口类型",
|
"",
|
"| 类型 | 数量 |",
|
"|---|---:|",
|
]
|
for key, count in gap_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## 队列状态", "", "| 状态 | 数量 |", "|---|---:|"])
|
for key, count in queue_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## 来源分布", "", "| 来源 | 数量 |", "|---|---:|"])
|
for key, count in source_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(
|
[
|
"",
|
"## 边界",
|
"",
|
"本轮只把 PASS-010 的 held 项拆成后续补证队列，未完成报告日期解析、原始表格行定位、访问归档或数值一致性核验；不作为正式指标、正式证据或正式结论。",
|
]
|
)
|
summary_path.parent.mkdir(parents=True, exist_ok=True)
|
summary_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
print(f"rows={len(rows)}")
|
print(dict(gap_counts))
|
print(dict(queue_counts))
|
print(f"output={output_path.relative_to(project_root).as_posix()}")
|
print(f"manifest={manifest_path.relative_to(project_root).as_posix()}")
|
print(f"summary={summary_path.relative_to(project_root).as_posix()}")
|
|
|
if __name__ == "__main__":
|
main()
|
