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()