import argparse import csv import hashlib import re from collections import Counter from datetime import datetime, timezone, timedelta from pathlib import Path CN_TZ = timezone(timedelta(hours=8)) TRIGGERS = [ ("推荐关注", "broker_recommend_attention"), ("建议关注", "broker_suggest_attention"), ("相关标的", "broker_related_targets"), ] def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def normalize(text: str) -> str: return re.sub(r"\s+", "", text or "") def classify(context: str): normalized = normalize(context) hits = [label for phrase, label in TRIGGERS if phrase in normalized] return ";".join(hits) def parse_table_row(line: str): if not line.startswith("|") or line.startswith("|---"): return None cells = [c.strip() for c in line.strip().strip("|").split("|")] if len(cells) < 4: return None return cells def build(view_path: Path, ledger_path: Path, manifest_path: Path, summary_path: Path, run_id: str): created_at = datetime.now(CN_TZ).isoformat(timespec="seconds") lines = view_path.read_text(encoding="utf-8").splitlines() in_company_table = False ledger_rows = [] new_lines = [] replaced = 0 for line in lines: if line.startswith("| 公司名片段 | 提及次数 | 样例定位 | 样例上下文 |"): in_company_table = True new_lines.append(line) continue if in_company_table and line.startswith("## ") and not line.startswith("## 公司提及频次"): in_company_table = False if in_company_table: cells = parse_table_row(line) if cells and cells[0] not in ("公司名片段", "---"): phrase_type = classify(cells[3]) if phrase_type: replaced += 1 row_id = f"YS-COMPANY-BROKER-PHRASE-016-{replaced:04d}" display_context = ( "BROKER_ORIGINAL_VIEW_SEGMENTED：该样例包含券商原文关注或标的列举表达，" "已从公司主读法分账；正式公司证据卡不得引用为分析员结论。" f"原始定位：{cells[2]}" ) ledger_rows.append( { "broker_phrase_split_id": row_id, "case_id": "ANA-YS-INDUSTRY-001", "batch_id": "BATCH-003", "run_id": run_id, "company_name_fragment": cells[0], "mention_count": cells[1], "sample_location": cells[2], "phrase_type": phrase_type, "original_context": cells[3], "display_context": display_context, "evidence_status": "BROKER_ORIGINAL_VIEW_SEGMENTED", "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) cells[3] = display_context new_lines.append("| " + " | ".join(cells) + " |") continue new_lines.append(line) section = [ "", "## 券商原文观点分账", "", f"PASS-016 已将公司提及频次表中 {replaced} 条含“推荐关注/建议关注/相关标的”等券商原文表达的样例上下文从主展示降噪，统一标记为 `BROKER_ORIGINAL_VIEW_SEGMENTED`。", "", "这些片段只保留为研报原文观点或公司提及线索，不得进入分析员结论字段、公司投资读法、交易指令或收益承诺。原始片段、定位和分账类型见：`ana-data/cases/有色案例/evidence/company_view_broker_phrase_split_pass016.csv`。", ] if "## 券商原文观点分账" not in "\n".join(new_lines): for i, line in enumerate(new_lines): if line.startswith("## 后续公司证据卡字段"): new_lines = new_lines[:i] + section + [""] + new_lines[i:] break for i, line in enumerate(new_lines): if line.startswith("更新时间："): new_lines[i] = "更新时间：2026-06-26T07:43:00+08:00" break view_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8") ledger_path.parent.mkdir(parents=True, exist_ok=True) fieldnames = [ "broker_phrase_split_id", "case_id", "batch_id", "run_id", "company_name_fragment", "mention_count", "sample_location", "phrase_type", "original_context", "display_context", "evidence_status", "review_status", "created_at", ] with ledger_path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(ledger_rows) digest = sha256_file(ledger_path) manifest_path.parent.mkdir(parents=True, exist_ok=True) manifest_rows = [ { "case_id": "ANA-YS-INDUSTRY-001", "batch_id": "BATCH-003", "run_id": run_id, "artifact_type": "company_view_broker_phrase_split", "artifact_path": ledger_path.as_posix(), "row_count": str(len(ledger_rows)), "sha256": digest, "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ] with manifest_path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=list(manifest_rows[0].keys())) writer.writeheader() writer.writerows(manifest_rows) by_type = Counter() for row in ledger_rows: for item in row["phrase_type"].split(";"): by_type[item] += 1 summary_lines = [ "# 公司视图券商原文观点分账 PASS-016 摘要", "", "状态：DRAFT_FOR_REVIEW", f"生成时间：{created_at}", "", "## 输出", "", f"- 分账清单：`{ledger_path.as_posix()}`", f"- manifest：`{manifest_path.as_posix()}`", f"- 公司视图草稿：`{view_path.as_posix()}`", f"- 分账记录数：{len(ledger_rows)}", f"- sha256：`{digest}`", "", "## 表达类型", "", "| 类型 | 数量 |", "|---|---:|", ] for key, count in by_type.items(): summary_lines.append(f"| {key} | {count} |") summary_lines += [ "", "## 边界", "", "本轮只对公司视图展示层做券商原文观点分账和降噪，不改变底层原文证据，不输出正式公司结论、交易指令或收益承诺。正式公司证据卡仍需业务占比、资源/产能、利润弹性、触发/失效条件、风险和官方来源核验。", "", ] summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text("\n".join(summary_lines), encoding="utf-8") print(f"OK split={len(ledger_rows)} output={ledger_path} sha256={digest}") def main(): parser = argparse.ArgumentParser() parser.add_argument("--view", required=True) parser.add_argument("--ledger", required=True) parser.add_argument("--manifest", required=True) parser.add_argument("--summary", required=True) parser.add_argument("--run-id", required=True) args = parser.parse_args() build(Path(args.view), Path(args.ledger), Path(args.manifest), Path(args.summary), args.run_id) if __name__ == "__main__": main()