import argparse import csv import hashlib import re from collections import Counter, defaultdict from datetime import datetime, timezone, timedelta from pathlib import Path STOPWORDS = { "有色金属", "能源金属", "工业金属", "基本金属", "贵金属", "小金属", "金属新材料", "中重稀土", "矿产资源", "自然资源", "稀土资源", "国家能源", "中国新能源", "基础化工", "稀土冶炼", "年全球稀土", "全球稀土", "COMEX黄金", "LME铜", "LME铝", } NOISE_PATTERNS = ["本报告", "所有材料", "免责声明", "证券研究报告", "请务必阅读", "评级", "走势图"] OPINION_WORDS = ["推荐", "建议关注", "相关标的", "有望", "预计", "看好", "维持"] 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 split_tags(value): return [tag for tag in (value or "").split("|") if tag] def clean_text(value): return re.sub(r"\s+", " ", (value or "").strip()) def is_company_candidate(name, context): if not name or name in STOPWORDS or name.isdigit() or len(name) > 14: return False if any(pattern in name for pattern in NOISE_PATTERNS): return False if any(pattern in context for pattern in ["版权均属", "不得以任何形式", "目录", "分析师承诺"]): return False return True def expression_type(text): if any(word in text for word in OPINION_WORDS): return "BROKER_VIEW_OR_ORIGINAL_RECOMMENDATION" if re.search(r"\d", text): return "NUMERIC_OR_FACT_CONTEXT" return "TEXT_CONTEXT" def card_priority(count, metals, examples): score = count score += len(metals) * 3 if any(re.search(r"\d", row.get("context_snippet", "")) for row in examples): score += 5 if any("BROKER_VIEW" in expression_type(row.get("context_snippet", "")) for row in examples): score += 1 return score def sha256_file(path): return hashlib.sha256(path.read_bytes()).hexdigest() def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--company-files", nargs="+", required=True) parser.add_argument("--cards-out", required=True) parser.add_argument("--manifest-out", required=True) parser.add_argument("--summary-out", required=True) parser.add_argument("--limit", type=int, default=80) args = parser.parse_args() project_root = Path(args.project_root).resolve() rows = [] for file_name in args.company_files: path = project_root / file_name if path.exists(): rows.extend(read_csv(path)) grouped = defaultdict(list) for row in rows: name = row.get("company_name", "") context = clean_text(row.get("context_snippet", "")) if is_company_candidate(name, context): grouped[name].append(row) ranked = [] for name, examples in grouped.items(): metal_counter = Counter() for row in examples: metal_counter.update(split_tags(row.get("metal_tags"))) ranked.append((card_priority(len(examples), metal_counter, examples), name, examples, metal_counter)) ranked.sort(reverse=True) created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") cards = [] for _, name, examples, metal_counter in ranked[: args.limit]: example = examples[0] numeric_example = next((row for row in examples if re.search(r"\d", row.get("context_snippet", ""))), example) broker_flag = any(expression_type(row.get("context_snippet", "")) == "BROKER_VIEW_OR_ORIGINAL_RECOMMENDATION" for row in examples[:10]) cards.append( { "company_card_id": f"YS-COMP-CARD-001-{len(cards) + 1:04d}", "case_id": "ANA-YS-INDUSTRY-001", "batch_id": "BATCH-001+BATCH-003", "run_id": "RUN-ANA-YS-COMPANY-CARD-001", "company_name": name, "mention_count": len(examples), "metal_tags_in_context": "|".join([metal for metal, _ in metal_counter.most_common(8)]), "source_expression_type": "HAS_BROKER_VIEW_OR_ORIGINAL_RECOMMENDATION" if broker_flag else expression_type(numeric_example.get("context_snippet", "")), "sample_doc_id": numeric_example.get("doc_id", ""), "sample_page_no": numeric_example.get("page_no", ""), "sample_source_location": numeric_example.get("source_location", ""), "sample_converted_text_path": numeric_example.get("converted_text_path", ""), "sample_raw_file_path": numeric_example.get("raw_file_path", ""), "sample_context": clean_text(numeric_example.get("context_snippet", ""))[:500], "business_exposure_status": "NEEDS_EXTERNAL_COMPANY_SOURCE", "resource_or_capacity_status": "NEEDS_EXTERNAL_COMPANY_SOURCE", "profit_sensitivity_status": "NEEDS_MODEL_OR_DISCLOSURE", "trigger_condition_status": "DRAFT_TO_BE_FILLED", "failure_condition_status": "DRAFT_TO_BE_FILLED", "risk_status": "DRAFT_TO_BE_FILLED", "next_action": "BUILD_COMPANY_SOURCE_CARD_FROM_REPORT_AND_ANNOUNCEMENT", "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) cards_path = project_root / args.cards_out manifest_path = project_root / args.manifest_out summary_path = project_root / args.summary_out fields = [ "company_card_id", "case_id", "batch_id", "run_id", "company_name", "mention_count", "metal_tags_in_context", "source_expression_type", "sample_doc_id", "sample_page_no", "sample_source_location", "sample_converted_text_path", "sample_raw_file_path", "sample_context", "business_exposure_status", "resource_or_capacity_status", "profit_sensitivity_status", "trigger_condition_status", "failure_condition_status", "risk_status", "next_action", "review_status", "created_at", ] write_csv(cards_path, fields, cards) manifest_rows = [ { "case_id": "ANA-YS-INDUSTRY-001", "batch_id": "BATCH-001+BATCH-003", "run_id": "RUN-ANA-YS-COMPANY-CARD-001", "artifact_type": "company_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, ) expression_counts = Counter(row["source_expression_type"] for row in cards) metal_counts = Counter() for row in cards: metal_counts.update(split_tags(row.get("metal_tags_in_context"))) lines = [ "# 公司证据卡候选 PASS-001 摘要", "", "状态：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)}", "", "## 表达类型", "", "| 类型 | 数量 |", "|---|---:|", ] for key, count in expression_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## 金属上下文覆盖", "", "| 金属 | 数量 |", "|---|---:|"]) for key, count in metal_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()