import argparse import csv import hashlib import re from collections import Counter from datetime import datetime, timezone, timedelta from pathlib import Path VALUE_UNIT_RE = re.compile( r"([-+]?\d+(?:\.\d+)?)\s*(%|万吨|吨|亿元|元/吨|美元/吨|美元/盎司|万元/吨|万\w*吨|GWh|万吨/年|吨/年|万股|亿元/年|万吨金属量)" ) DATE_RE = re.compile(r"(20\d{2}[年/-]\d{1,2}(?:[月/-]\d{1,2}日?)?|20\d{2}年|Q[1-4]|[一二三四]季度|本周|上周|本月)") 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 extract_values(text): pairs = [] for value, unit in VALUE_UNIT_RE.findall(text or ""): pairs.append(f"{value}{unit}") return "|".join(pairs[:8]) def extract_dates(text): return "|".join(DATE_RE.findall(text or "")[:6]) def evidence_kind(theme_tags): themes = set((theme_tags or "").split("|")) if "price" in themes: return "PRICE_OR_MARKET_DATA" if "inventory" in themes: return "INVENTORY" if "supply" in themes: return "SUPPLY_CAPACITY_OR_OUTPUT" if "demand" in themes: return "DEMAND" if "project" in themes: return "PROJECT_OR_ASSET" if "cost" in themes: return "COST_OR_MARGIN" if "policy" in themes: return "POLICY" return "FACT_CANDIDATE" def next_action(row, values): if not values: return "NEEDS_MANUAL_NUMERIC_OR_TABLE_REVIEW" if "price" in row.get("theme_tags", "") or "inventory" in row.get("theme_tags", ""): return "ADD_TABLE_LOCATION_AND_EXTERNAL_CROSS_CHECK" return "ADD_PARAGRAPH_LOCATION_AND_SOURCE_CONTEXT" def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--review", required=True) parser.add_argument("--cards-out", required=True) parser.add_argument("--manifest-out", required=True) parser.add_argument("--summary-out", required=True) args = parser.parse_args() project_root = Path(args.project_root).resolve() rows = read_csv(project_root / args.review) candidates = [ row for row in rows if row.get("evidence_upgrade_status") == "CANDIDATE_FOR_EVIDENCE_CARD" and row.get("scope_gate") == "CORE_INDUSTRY_CANDIDATE" ] created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") cards = [] for row in candidates: text = row.get("evidence_text", "") context = row.get("page_context", "") values = extract_values(text + " " + context) dates = extract_dates(text + " " + context) cards.append( { "evidence_card_id": f"YS-FACT-CARD-002-{len(cards) + 1:04d}", "case_id": row.get("case_id", "ANA-YS-INDUSTRY-001"), "batch_id": row.get("batch_id", ""), "run_id": "RUN-ANA-YS-FACT-CARD-002", "fact_id": row.get("fact_id", ""), "doc_id": row.get("doc_id", ""), "metal_tags": row.get("metal_tags", ""), "theme_tags": row.get("theme_tags", ""), "evidence_kind": evidence_kind(row.get("theme_tags", "")), "source_expression_type": row.get("source_expression_type", ""), "declared_page_no": row.get("declared_page_no", ""), "matched_page_no": row.get("matched_page_no", ""), "converted_text_path": row.get("converted_text_path", ""), "raw_file_path": row.get("raw_file_path", ""), "raw_sha256": row.get("raw_sha256", ""), "evidence_text": text, "value_unit_candidates": values, "date_candidates": dates, "page_context": context[:700], "table_location_status": "NEEDS_TABLE_OR_PARAGRAPH_LOCATION", "external_cross_check_status": "NEEDS_EXTERNAL_CROSS_CHECK", "next_action": next_action(row, values), "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) cards_path = project_root / args.cards_out fields = [ "evidence_card_id", "case_id", "batch_id", "run_id", "fact_id", "doc_id", "metal_tags", "theme_tags", "evidence_kind", "source_expression_type", "declared_page_no", "matched_page_no", "converted_text_path", "raw_file_path", "raw_sha256", "evidence_text", "value_unit_candidates", "date_candidates", "page_context", "table_location_status", "external_cross_check_status", "next_action", "review_status", "created_at", ] write_csv(cards_path, fields, cards) manifest_path = project_root / args.manifest_out manifest_rows = [ { "case_id": "ANA-YS-INDUSTRY-001", "batch_id": "BATCH-001+BATCH-003", "run_id": "RUN-ANA-YS-FACT-CARD-002", "artifact_type": "key_fact_evidence_card", "artifact_path": cards_path.relative_to(project_root).as_posix(), "row_count": len(cards), "sha256": sha256_file(cards_path), "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ] write_csv( manifest_path, ["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"], manifest_rows, ) kind_counts = Counter(card["evidence_kind"] for card in cards) action_counts = Counter(card["next_action"] for card in cards) with_values = sum(1 for card in cards if card["value_unit_candidates"]) summary_path = project_root / args.summary_out lines = [ "# 关键事实证据卡 PASS-002 摘要", "", "状态：DRAFT_FOR_REVIEW", f"生成时间：{created_at}", "", "## 输出", "", f"- 证据卡：`{cards_path.relative_to(project_root).as_posix()}`", f"- manifest：`{manifest_path.relative_to(project_root).as_posix()}`", f"- 证据卡数量：{len(cards)}", f"- 含数值/单位候选：{with_values}", "", "## 证据类型", "", "| 类型 | 数量 |", "|---|---:|", ] for key, count in kind_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## 下一步动作", "", "| 动作 | 数量 |", "|---|---:|"]) for key, count in action_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend( [ "", "## 边界", "", "本轮只是把关键事实候选拆成证据卡草表。数值/单位仍为候选，正式证据升级前必须补段落或表格定位，并完成外部交叉验证。", ] ) summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"cards={len(cards)}") print(f"cards_out={cards_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()