import argparse import csv import hashlib import re from collections import Counter from datetime import datetime, timezone, timedelta from pathlib import Path METAL_KEYWORDS = { "铜": ["铜", "铜矿", "电解铜", "阴极铜", "沪铜", "伦铜"], "铝": ["铝", "氧化铝", "电解铝", "沪铝", "铝土矿"], "金": ["黄金", "金价", "贵金属", "COMEX金", "伦敦金"], "银": ["白银", "银价", "银浆"], "锂": ["锂", "碳酸锂", "氢氧化锂", "锂矿", "盐湖"], "钴": ["钴"], "镍": ["镍"], "稀土": ["稀土", "氧化镨钕", "镨钕", "钕铁硼", "磁材"], "锡": ["锡"], "钨": ["钨"], "钼": ["钼"], "锑": ["锑"], } THEME_KEYWORDS = { "price": ["价格", "价差", "涨", "跌", "均价", "报价", "沪", "LME", "COMEX"], "inventory": ["库存", "累库", "去库", "仓单"], "supply": ["供给", "供应", "产量", "产能", "矿山", "冶炼", "开工", "进口"], "demand": ["需求", "消费", "订单", "出货", "下游", "终端"], "cost": ["成本", "利润", "毛利", "加工费", "TC", "电价"], "policy": ["政策", "关税", "配额", "制裁", "出口管制", "收储"], "risk": ["风险", "扰动", "不及预期", "下滑", "压力", "亏损"], "company": ["公司", "股份", "集团", "矿业", "锂业", "铝业", "铜业", "黄金", "稀土"], "project": ["项目", "扩建", "投产", "建设", "收购", "资源量", "储量"], } COMPANY_PATTERN = re.compile( r"([\u4e00-\u9fa5A-Za-z0-9]{2,18}(?:股份|集团|矿业|锂业|铝业|铜业|黄金|稀土|钴业|能源|材料|资源|科技|新材|金属|冶炼|化工))" ) STOCK_PATTERN = re.compile(r"(?:股票代码|代码|证券代码)?[:： ]?([036]\d{5})(?:\.[A-Z]{2})?") 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 norm_path(value): return (value or "").replace("\\", "/").strip() def text_hash(value): return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] def split_sentences(text): text = re.sub(r"[ \t]+", " ", text) parts = re.split(r"(?<=[。！？；;])|\n+", text) return [part.strip() for part in parts if len(part.strip()) >= 18] def scan_text(path): current_page = "" for line in path.read_text(encoding="utf-8", errors="ignore").splitlines(): page_match = re.match(r"\[\[PAGE\s+(\d+)\]\]", line.strip()) if page_match: current_page = page_match.group(1) continue for sentence in split_sentences(line): yield current_page, sentence def tags_for(sentence, keyword_map): tags = [] for tag, keywords in keyword_map.items(): if any(keyword in sentence for keyword in keywords): tags.append(tag) return tags def fact_type(sentence, theme_tags): if "risk" in theme_tags: return "risk_sentence" if re.search(r"\d+(?:\.\d+)?\s*(?:%|万吨|吨|亿元|元/吨|美元/吨|万亿元|GWh|万台)", sentence): return "indicator_sentence" return "fact_or_view_sentence" def confidence(sentence, metal_tags, theme_tags): if not metal_tags: return "LOW" if len(theme_tags) >= 2 or re.search(r"\d", sentence): return "MEDIUM" return "LOW" def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--conversion-status", required=True) parser.add_argument("--extracted-dir", required=True) parser.add_argument("--manifest-dir", required=True) parser.add_argument("--outputs-dir", required=True) parser.add_argument("--case-id", default="ANA-YS-INDUSTRY-001") parser.add_argument("--batch-id", default="BATCH-003") parser.add_argument("--run-id", default="RUN-ANA-YS-INDUSTRY-001-BATCH-003-EXTRACT-001") parser.add_argument("--max-facts-per-doc", type=int, default=12) args = parser.parse_args() project_root = Path(args.project_root).resolve() status_rows = read_csv(project_root / args.conversion_status) extracted_dir = project_root / args.extracted_dir manifest_dir = project_root / args.manifest_dir outputs_dir = project_root / args.outputs_dir created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") fact_rows = [] class_rows = [] company_rows = [] converted_rows = [row for row in status_rows if row.get("conversion_status") == "TEXT_CONVERTED"] for doc_index, row in enumerate(converted_rows, start=1): converted_path = project_root / row.get("converted_text_path", "") if not converted_path.exists(): continue doc_id = row.get("source_doc_id") or f"YS-B3-DOC-{doc_index:03d}" doc_metal_counter = Counter() doc_theme_counter = Counter() doc_fact_count = 0 company_seen = set() for page, sentence in scan_text(converted_path): metal_tags = tags_for(sentence, METAL_KEYWORDS) theme_tags = tags_for(sentence, THEME_KEYWORDS) if metal_tags: doc_metal_counter.update(metal_tags) if theme_tags: doc_theme_counter.update(theme_tags) if metal_tags and theme_tags and doc_fact_count < args.max_facts_per_doc: doc_fact_count += 1 fact_rows.append( { "fact_id": f"YS-B3-F-{len(fact_rows) + 1:05d}", "case_id": args.case_id, "batch_id": args.batch_id, "sub_batch_id": row.get("sub_batch_id", ""), "run_id": args.run_id, "doc_id": doc_id, "fact_type": fact_type(sentence, theme_tags), "metal_tags": "|".join(metal_tags), "theme_tags": "|".join(theme_tags), "evidence_text": sentence[:220], "source_location": f"converted_text_page_{page or 'unknown'}_sentence_scan", "page_no": page, "converted_text_path": row.get("converted_text_path", ""), "raw_file_path": row.get("raw_file_path", ""), "raw_sha256": row.get("raw_sha256", ""), "confidence": confidence(sentence, metal_tags, theme_tags), "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) for company_match in COMPANY_PATTERN.findall(sentence): if company_match in company_seen: continue company_seen.add(company_match) stock_match = STOCK_PATTERN.search(sentence) company_rows.append( { "company_map_id": f"YS-B3-CM-{len(company_rows) + 1:05d}", "case_id": args.case_id, "batch_id": args.batch_id, "sub_batch_id": row.get("sub_batch_id", ""), "run_id": args.run_id, "doc_id": doc_id, "company_name": company_match, "stock_code": stock_match.group(1) if stock_match else "", "metal_tags": "|".join(tags_for(sentence, METAL_KEYWORDS)), "context_snippet": sentence[:220], "source_location": f"converted_text_page_{page or 'unknown'}_company_scan", "page_no": page, "converted_text_path": row.get("converted_text_path", ""), "raw_file_path": row.get("raw_file_path", ""), "confidence": "LOW", "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) class_rows.append( { "case_id": args.case_id, "batch_id": args.batch_id, "sub_batch_id": row.get("sub_batch_id", ""), "run_id": args.run_id, "doc_id": doc_id, "raw_file_path": row.get("raw_file_path", ""), "converted_text_path": row.get("converted_text_path", ""), "top_metal_tags": "|".join([tag for tag, _ in doc_metal_counter.most_common(6)]), "top_theme_tags": "|".join([tag for tag, _ in doc_theme_counter.most_common(8)]), "fact_count": doc_fact_count, "company_mention_count": len(company_seen), "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) fact_path = extracted_dir / "evidence_fact_table_batch003_extract001.csv" class_path = extracted_dir / "classification_summary_batch003_extract001.csv" company_path = extracted_dir / "company_exposure_batch003_extract001.csv" write_csv( fact_path, [ "fact_id", "case_id", "batch_id", "sub_batch_id", "run_id", "doc_id", "fact_type", "metal_tags", "theme_tags", "evidence_text", "source_location", "page_no", "converted_text_path", "raw_file_path", "raw_sha256", "confidence", "review_status", "created_at", ], fact_rows, ) write_csv( class_path, [ "case_id", "batch_id", "sub_batch_id", "run_id", "doc_id", "raw_file_path", "converted_text_path", "top_metal_tags", "top_theme_tags", "fact_count", "company_mention_count", "review_status", "created_at", ], class_rows, ) write_csv( company_path, [ "company_map_id", "case_id", "batch_id", "sub_batch_id", "run_id", "doc_id", "company_name", "stock_code", "metal_tags", "context_snippet", "source_location", "page_no", "converted_text_path", "raw_file_path", "confidence", "review_status", "created_at", ], company_rows, ) manifest_rows = [] for artifact_path, artifact_type, row_count in [ (fact_path, "evidence_fact_table", len(fact_rows)), (class_path, "classification_summary", len(class_rows)), (company_path, "company_exposure", len(company_rows)), ]: data = artifact_path.read_bytes() manifest_rows.append( { "case_id": args.case_id, "batch_id": args.batch_id, "run_id": args.run_id, "artifact_type": artifact_type, "artifact_path": artifact_path.relative_to(project_root).as_posix(), "row_count": row_count, "sha256": hashlib.sha256(data).hexdigest(), "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) manifest_path = manifest_dir / "batch003_extract001_manifest.csv" write_csv( manifest_path, ["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"], manifest_rows, ) by_sub = Counter(row.get("sub_batch_id", "") for row in fact_rows) md_lines = [ "# ANA-YS-INDUSTRY-001 BATCH-003 EXTRACT-001 批次摘要", "", "状态：DRAFT_FOR_REVIEW", f"生成时间：{created_at}", f"case_id：{args.case_id}", f"batch_id：{args.batch_id}", f"run_id：{args.run_id}", "", "## 本轮动作", "", "对 BATCH-003 已转换文本执行规则扫描，生成事实句、分类汇总和公司映射草表。事实句保留 converted text 页码级定位；该定位仍需人工复核，不等于正式引用页码。", "", "## 输出", "", "| 文件 | 行数 | 用途 |", "|---|---:|---|", f"| `{fact_path.relative_to(project_root).as_posix()}` | {len(fact_rows)} | 事实句、指标句、风险句草表 |", f"| `{class_path.relative_to(project_root).as_posix()}` | {len(class_rows)} | 每篇研报金属和主题标签 |", f"| `{company_path.relative_to(project_root).as_posix()}` | {len(company_rows)} | 公司映射和上下文片段草表 |", f"| `{manifest_path.relative_to(project_root).as_posix()}` | {len(manifest_rows)} | 本轮产物清单和 hash |", "", "## sub_batch 事实句数量", "", "| sub_batch_id | 事实句 |", "|---|---:|", ] for sub_batch_id in sorted(by_sub): md_lines.append(f"| {sub_batch_id} | {by_sub[sub_batch_id]} |") md_lines.extend( [ "", "## 边界", "", "本轮仍是规则扫描。`source_location` 为 converted text 页码和句子扫描定位，后续必须补段落、表格定位和人工校验后，才能升级为正式证据。", ] ) summary_path = outputs_dir / "batch003_extract001_summary.md" summary_path.write_text("\n".join(md_lines) + "\n", encoding="utf-8") print(f"facts={len(fact_rows)}") print(f"classification={len(class_rows)}") print(f"companies={len(company_rows)}") 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()