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