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()
|
