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 source_candidates(row): fact_id = row.get("fact_id", "") if fact_id == "YS-B3-F-02056": return [ { "company_name": "洛阳钼业", "stock_code": "603993.SH / 03993.HK", "source_type": "company_official_report_entry", "source_name": "洛阳钼业业绩与报告入口", "source_url": "https://www.cmoc.com/html/InvestorMedia/PerformanceReport/Quarterly/", "source_period_or_date": "2025/2026", "source_access_status": "SOURCE_ENTRY_LOCATED", "source_relevance": "official report entry; annual report row and tungsten detail still need extraction", "next_action": "open official annual report entry and archive annual report page or PDF", }, { "company_name": "洛阳钼业", "stock_code": "603993.SH / 03993.HK", "source_type": "company_official_news", "source_name": "洛阳钼业2025年盈利203亿元新闻发布", "source_url": "https://www.cmoc.com/html/2026/News_0327/398.html", "source_period_or_date": "2026-03-27", "source_access_status": "SOURCE_ENTRY_LOCATED", "source_relevance": "official company news confirms 2025 performance context; tungsten segment detail still needs annual report/table row", "next_action": "use as company context only; verify tungsten business value against annual report or exchange filing", }, { "company_name": "洛阳钼业", "stock_code": "603993.SH / 03993.HK", "source_type": "disclosure_mirror_annual_report_pdf", "source_name": "洛阳钼业2025年年度报告披露镜像", "source_url": "https://pdf.dfcfw.com/pdf/H2_AN202603271820812761_1.pdf", "source_period_or_date": "2025", "source_access_status": "SOURCE_ENTRY_LOCATED_NEEDS_ARCHIVE", "source_relevance": "annual report mirror candidate; must archive and extract exact tungsten row before formal use", "next_action": "archive PDF and search tungsten segment/production guidance row", }, ] if fact_id in {"YS-B3-F-00805", "YS-B3-F-02391"}: return [ { "company_name": "湖南黄金", "stock_code": "002155.SZ", "source_type": "disclosure_pdf_project_announcement", "source_name": "湖南黄金关于控股子公司新邵四维取得采矿许可证的公告", "source_url": "https://xinpi.zqrb.cn/pdf/SZ/002155/2025/1128/f3d81db0a446b7501b6b0e4e00fbc0de.pdf", "source_period_or_date": "2025-11-28", "source_access_status": "SOURCE_ENTRY_LOCATED_NEEDS_ARCHIVE", "source_relevance": "project announcement candidate for Tanxi tungsten mine mining license and 990kt/year scale", "next_action": "archive PDF and extract license name, mineral type, production scale and validity period", }, { "company_name": "湖南黄金", "stock_code": "002155.SZ", "source_type": "company_annual_report_disclosure_mirror", "source_name": "湖南黄金2025年年度报告披露镜像", "source_url": "https://money.finance.sina.com.cn/corp/view/vCB_AllBulletinDetail.php?id=12073043&stockid=002155", "source_period_or_date": "2025", "source_access_status": "SOURCE_ENTRY_LOCATED_NEEDS_ARCHIVE", "source_relevance": "annual report candidate for company resource, tungsten output and business context", "next_action": "archive annual report and extract tungsten resource/output table rows", }, { "company_name": "湖南黄金", "stock_code": "002155.SZ", "source_type": "disclosure_media_entry", "source_name": "上证报湖南黄金控股子公司新邵四维取得采矿许可证", "source_url": "https://www.cnstock.com/commonDetail/594295", "source_period_or_date": "2025-11-27", "source_access_status": "SOURCE_ENTRY_LOCATED", "source_relevance": "disclosure media confirms announcement summary; use only as secondary pointer before formal evidence upgrade", "next_action": "use as pointer; formal evidence should rely on announcement PDF or annual report", }, ] return [] def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--gap-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-COMPANY-OFFICIAL-SOURCE-012") args = parser.parse_args() project_root = Path(args.project_root).resolve() gaps = read_csv(project_root / args.gap_input) target = [row for row in gaps if row.get("gap_type") == "COMPANY_OFFICIAL_SOURCE_GAP"] created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") rows = [] for gap in target: for source in source_candidates(gap): rows.append( { "company_official_source_id": f"YS-COMPANY-OFFICIAL-SRC-012-{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", ""), "preferred_source_type": gap.get("preferred_source_type", ""), "company_name": source["company_name"], "stock_code": source["stock_code"], "source_type": source["source_type"], "source_name": source["source_name"], "source_url": source["source_url"], "source_period_or_date": source["source_period_or_date"], "source_access_status": source["source_access_status"], "source_relevance": source["source_relevance"], "value_unit_candidates": gap.get("value_unit_candidates", ""), "date_candidates": gap.get("date_candidates", ""), "evidence_status": "SOURCE_ENTRY_DRAFT_NOT_VALUE_VERIFIED", "next_action": source["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 ["company_official_source_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-003", "run_id": args.run_id, "artifact_type": "company_official_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, } ], ) company_counts = Counter(row["company_name"] for row in rows) source_type_counts = Counter(row["source_type"] for row in rows) status_counts = Counter(row["source_access_status"] for row in rows) summary_path = project_root / args.summary lines = [ "# 公司官方源补证 PASS-012 摘要", "", "状态：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(target)}", f"- 来源候选记录：{len(rows)}", f"- sha256：`{output_sha}`", "", "## 公司分布", "", "| 公司 | 数量 |", "|---|---:|", ] for key, count in company_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## 来源类型", "", "| 来源类型 | 数量 |", "|---|---:|"]) for key, count in source_type_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## 来源状态", "", "| 状态 | 数量 |", "|---|---:|"]) for key, count in status_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend( [ "", "## 边界", "", "本轮只补充公司官方源或披露入口候选，尚未完成网页/PDF 归档、原文表格行抽取、数值一致性核验或正式证据升级；不作为正式公司结论或正式指标。", ] ) 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(company_counts)) print(dict(source_type_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()