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 method_for_source(source_name, source_status): if "LBMA" in source_name: return { "verify_date_field": "LBMA benchmark quote date", "verify_instrument_field": "gold/silver benchmark price type", "verify_unit_field": "USD per troy ounce or published LBMA unit", "verify_table_field": "LBMA price table/download row", "verify_value_method": "compare candidate value/date against LBMA published benchmark row", "verify_status": "VERIFY_QUEUE_READY", } if "Shanghai Gold Exchange" in source_name: return { "verify_date_field": "SGE trading date", "verify_instrument_field": "Au/Ag benchmark, contract or quote variety", "verify_unit_field": "CNY per gram or SGE published unit", "verify_table_field": "SGE benchmark quote table row", "verify_value_method": "compare candidate China gold/silver quote against SGE daily benchmark row", "verify_status": "VERIFY_QUEUE_READY", } if "CME" in source_name or "COMEX" in source_name: return { "verify_date_field": "CME warehouse stock report date", "verify_instrument_field": "COMEX gold or silver depository/warehouse category", "verify_unit_field": "troy ounce or CME published inventory unit", "verify_table_field": "CME warehouse stock report row", "verify_value_method": "compare candidate inventory/stock wording against CME daily warehouse report", "verify_status": "VERIFY_QUEUE_READY", } if "SMM" in source_name: status = "VERIFY_HELD_BY_PARTIAL_ACCESS" if source_status == "SOURCE_ENTRY_PARTIAL_ACCESS" else "VERIFY_QUEUE_READY" return { "verify_date_field": "SMM quote date", "verify_instrument_field": "antimony specification", "verify_unit_field": "SMM published price unit", "verify_table_field": "SMM price page quote row", "verify_value_method": "confirm quote date/value access or replace with alternative public source", "verify_status": status, } return { "verify_date_field": "source date", "verify_instrument_field": "source instrument", "verify_unit_field": "source unit", "verify_table_field": "source table row", "verify_value_method": "manual verification required", "verify_status": "VERIFY_METHOD_REVIEW", } def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--source-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-VERIFY-009") args = parser.parse_args() project_root = Path(args.project_root).resolve() sources = read_csv(project_root / args.source_input) created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") rows = [] for source in sources: method = method_for_source(source.get("source_name", ""), source.get("source_status", "")) rows.append( { "source_verify_id": f"YS-SOURCE-VERIFY-009-{len(rows) + 1:04d}", "case_id": source.get("case_id", ""), "batch_id": source.get("batch_id", ""), "run_id": args.run_id, "source_supplement_id": source.get("source_supplement_id", ""), "gap_priority_id": source.get("gap_priority_id", ""), "evidence_card_id": source.get("evidence_card_id", ""), "fact_id": source.get("fact_id", ""), "doc_id": source.get("doc_id", ""), "metal_tags": source.get("metal_tags", ""), "theme_tags": source.get("theme_tags", ""), "source_name": source.get("source_name", ""), "source_url": source.get("source_url", ""), "source_status": source.get("source_status", ""), "verify_date_field": method["verify_date_field"], "verify_instrument_field": method["verify_instrument_field"], "verify_unit_field": method["verify_unit_field"], "verify_table_field": method["verify_table_field"], "verify_value_method": method["verify_value_method"], "verify_status": method["verify_status"], "value_unit_candidates": source.get("value_unit_candidates", ""), "date_candidates": source.get("date_candidates", ""), "next_action": "run_or_manual_verify_source_row" if method["verify_status"] == "VERIFY_QUEUE_READY" else "resolve_source_access_or_replace_source", "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) output_path = project_root / args.output fields = list(rows[0].keys()) if rows else ["source_verify_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": "key_fact_source_verify_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, } ], ) source_counts = Counter(row["source_name"] for row in rows) status_counts = Counter(row["verify_status"] for row in rows) summary_path = project_root / args.summary lines = [ "# \u9ad8\u4f18\u5148\u7ea7\u6765\u6e90\u6838\u9a8c\u961f\u5217 PASS-009 \u6458\u8981", "", "\u72b6\u6001\uff1aDRAFT_FOR_REVIEW", f"\u751f\u6210\u65f6\u95f4\uff1a{created_at}", "", "## \u8f93\u51fa", "", f"- \u6838\u9a8c\u961f\u5217\uff1a`{output_path.relative_to(project_root).as_posix()}`", f"- manifest\uff1a`{manifest_path.relative_to(project_root).as_posix()}`", f"- \u8bb0\u5f55\u6570\uff1a{len(rows)}", f"- sha256\uff1a`{output_sha}`", "", "## \u6838\u9a8c\u72b6\u6001", "", "| \u72b6\u6001 | \u6570\u91cf |", "|---|---:|", ] for key, count in status_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## \u6765\u6e90\u5206\u5e03", "", "| \u6765\u6e90 | \u6570\u91cf |", "|---|---:|"]) for key, count in source_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend( [ "", "## \u8fb9\u754c", "", "\u672c\u8f6e\u53ea\u751f\u6210\u6765\u6e90\u6838\u9a8c\u961f\u5217\uff0c\u5c1a\u672a\u5b8c\u6210\u65e5\u671f\u3001\u54c1\u79cd/\u5408\u7ea6\u3001\u5355\u4f4d\u3001\u539f\u59cb\u8868\u683c\u884c\u548c\u6570\u503c\u4e00\u81f4\u6027\u6838\u9a8c\uff1b\u4e0d\u4f5c\u4e3a\u6b63\u5f0f\u8bc1\u636e\u6216\u6b63\u5f0f\u6307\u6807\u3002", ] ) 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(status_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()