import argparse import csv import hashlib from collections import Counter from datetime import datetime, timedelta, timezone from pathlib import Path SOURCE_LIBRARY = { "precious_price_global": { "source_name": "LBMA Precious Metal Prices", "source_url": "https://www.lbma.org.uk/prices-and-data/precious-metal-prices", "source_scope": "gold and silver global benchmark prices", "source_date_policy": "use quoted benchmark date on LBMA page or LBMA/IBA dataset date", "source_status": "SOURCE_ENTRY_READY", }, "precious_price_china": { "source_name": "Shanghai Gold Exchange benchmark price data", "source_url": "https://en.sge.com.cn/", "source_scope": "Shanghai gold and silver benchmark price entry", "source_date_policy": "use SGE daily benchmark date for SHAU/SHAG", "source_status": "SOURCE_ENTRY_READY", }, "precious_inventory": { "source_name": "CME COMEX & NYMEX Delivery Notices and Warehouse Stocks", "source_url": "https://www.cmegroup.com/solutions/clearing/operations-and-deliveries/nymex-delivery-notices.html", "source_scope": "COMEX gold and silver warehouse/depository stocks", "source_date_policy": "use CME daily warehouse stock report date", "source_status": "SOURCE_ENTRY_READY", }, "antimony_price": { "source_name": "SMM China Antimony 99.70% Sb min", "source_url": "https://www.metal.com/Antimony/201102250546", "source_scope": "China antimony ingot price entry and specification", "source_date_policy": "use SMM displayed quote date; numerical history may require access", "source_status": "SOURCE_ENTRY_PARTIAL_ACCESS", }, } 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 split_tags(value): return [part.strip() for part in (value or "").split("|") if part.strip()] def choose_sources(metals, themes, preferred): metal_set = set(metals) keys = [] if {"\u91d1", "\u9ec4\u91d1", "\u94f6", "\u767d\u94f6"} & metal_set: if "inventory" in themes or "inventory" in preferred: keys.append("precious_inventory") keys.extend(["precious_price_global", "precious_price_china"]) if "\u9511" in metal_set: keys.append("antimony_price") if not keys: keys.append("precious_price_global") # Keep stable ordering and no duplicates. out = [] for key in keys: if key not in out: out.append(key) return out def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--gap-priority", 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-HIGH-GAP-SOURCE-008") args = parser.parse_args() project_root = Path(args.project_root).resolve() gaps = [ row for row in read_csv(project_root / args.gap_priority) if row.get("priority") == "HIGH" and row.get("next_action") == "supplement_now" ] created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") rows = [] for gap in gaps: metals = split_tags(gap.get("metal_tags", "")) themes = split_tags(gap.get("theme_tags", "")) source_keys = choose_sources(metals, themes, gap.get("preferred_source_type", "")) for source_key in source_keys: source = SOURCE_LIBRARY[source_key] rows.append( { "source_supplement_id": f"YS-HIGH-GAP-SRC-008-{len(rows) + 1:04d}", "case_id": gap.get("case_id", ""), "batch_id": gap.get("batch_id", ""), "run_id": args.run_id, "gap_priority_id": gap.get("gap_priority_id", ""), "evidence_card_id": gap.get("evidence_card_id", ""), "fact_id": gap.get("fact_id", ""), "doc_id": gap.get("doc_id", ""), "metal_tags": gap.get("metal_tags", ""), "theme_tags": gap.get("theme_tags", ""), "gap_type": gap.get("gap_type", ""), "priority": gap.get("priority", ""), "preferred_source_type": gap.get("preferred_source_type", ""), "source_name": source["source_name"], "source_url": source["source_url"], "source_scope": source["source_scope"], "source_date_policy": source["source_date_policy"], "source_status": source["source_status"], "value_unit_candidates": gap.get("value_unit_candidates", ""), "date_candidates": gap.get("date_candidates", ""), "next_action": "extract_date_unit_value_or_confirm_source_gap", "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) output_path = project_root / args.output fields = list(rows[0].keys()) if rows else ["source_supplement_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_high_gap_source_supplement", "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) gap_counts = Counter(row["gap_priority_id"] for row in rows) status_counts = Counter(row["source_status"] for row in rows) summary_path = project_root / args.summary lines = [ "# \u9ad8\u4f18\u5148\u7ea7\u5173\u952e\u4e8b\u5b9e\u5916\u90e8\u6765\u6e90\u8865\u5145 PASS-008 \u6458\u8981", "", "\u72b6\u6001\uff1aDRAFT_FOR_REVIEW", f"\u751f\u6210\u65f6\u95f4\uff1a{created_at}", "", "## \u8f93\u51fa", "", f"- \u8865\u6e90\u8868\uff1a`{output_path.relative_to(project_root).as_posix()}`", f"- manifest\uff1a`{manifest_path.relative_to(project_root).as_posix()}`", f"- \u8986\u76d6 HIGH \u7f3a\u53e3\uff1a{len(gap_counts)}", f"- \u6765\u6e90\u5019\u9009\u8bb0\u5f55\uff1a{len(rows)}", f"- sha256\uff1a`{output_sha}`", "", "## \u6765\u6e90\u5206\u5e03", "", "| \u6765\u6e90 | \u6570\u91cf |", "|---|---:|", ] for key, count in source_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## \u6765\u6e90\u72b6\u6001", "", "| \u72b6\u6001 | \u6570\u91cf |", "|---|---:|"]) for key, count in status_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend( [ "", "## \u8fb9\u754c", "", "\u672c\u8f6e\u53ea\u8865\u5145\u5916\u90e8\u6765\u6e90\u5165\u53e3\u548c\u53e3\u5f84\u8bf4\u660e\uff0c\u4e0d\u8868\u793a\u6570\u503c\u3001\u5355\u4f4d\u3001\u65e5\u671f\u6216\u8868\u683c\u884c\u5df2\u4eba\u5de5\u6838\u5b9e\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"high_gaps={len(gap_counts)}") print(f"source_rows={len(rows)}") print(dict(source_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()