import argparse
|
import csv
|
import hashlib
|
import re
|
from collections import Counter, defaultdict
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
|
METAL_ORDER = ["铜", "铝", "金", "银", "锂", "镍", "钴", "稀土", "锡", "钨", "钼", "锑"]
|
THEME_PRIORITY = ["price", "inventory", "supply", "demand", "cost", "project", "policy", "risk", "company"]
|
OPINION_WORDS = ["推荐", "建议关注", "相关标的", "有望", "预计", "或将", "看好", "维持", "评级"]
|
TABLE_WORDS = ["图", "表", "数据来源", "来源：", "根据", "库存", "价格", "产量", "产能", "同比", "环比"]
|
ADJACENT_OR_NOISE_WORDS = [
|
"凤凰光学",
|
"骏鼎达",
|
"宏达电子",
|
"银禧科技",
|
"一汽解放",
|
"佰维存储",
|
"纳芯微",
|
"AI服务器",
|
"光通信",
|
"半固态电池",
|
"智能物联",
|
]
|
|
|
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 short_hash(path):
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
def score_fact(row):
|
metals = split_tags(row.get("metal_tags"))
|
themes = split_tags(row.get("theme_tags"))
|
text = row.get("evidence_text", "")
|
score = 0
|
if row.get("fact_type") == "indicator_sentence":
|
score += 6
|
if row.get("fact_type") == "risk_sentence":
|
score += 3
|
if re.search(r"\d", text):
|
score += 4
|
for theme in themes:
|
if theme in THEME_PRIORITY:
|
score += max(1, 10 - THEME_PRIORITY.index(theme))
|
for metal in metals:
|
if metal in METAL_ORDER:
|
score += 2
|
if any(word in text for word in TABLE_WORDS):
|
score += 2
|
if len(text) < 25:
|
score -= 2
|
return score
|
|
|
def load_pages(path):
|
if not path.exists():
|
return {}
|
pages = defaultdict(list)
|
current_page = "unknown"
|
for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
match = re.match(r"\[\[PAGE\s+(\d+)\]\]", line.strip())
|
if match:
|
current_page = match.group(1)
|
continue
|
cleaned = clean_text(line)
|
if cleaned:
|
pages[current_page].append(cleaned)
|
return {page: " ".join(lines) for page, lines in pages.items()}
|
|
|
def find_context(row, project_root):
|
converted = project_root / row.get("converted_text_path", "")
|
pages = load_pages(converted)
|
if not pages:
|
return "NO_TEXT_FILE", "", ""
|
page_no = row.get("page_no") or "unknown"
|
text = clean_text(row.get("evidence_text", ""))
|
candidates = []
|
if page_no in pages:
|
candidates.append((page_no, pages[page_no]))
|
candidates.extend((page, body) for page, body in pages.items() if page != page_no)
|
needle = text[:80]
|
for page, body in candidates:
|
idx = body.find(needle)
|
if idx >= 0:
|
start = max(0, idx - 180)
|
end = min(len(body), idx + len(text) + 220)
|
status = "TEXT_MATCH_CONTEXT_OK" if page == page_no else "TEXT_MATCH_DIFFERENT_PAGE"
|
return status, page, body[start:end]
|
tokens = [token for token in re.split(r"\W+", text) if len(token) >= 2]
|
hit_count = sum(1 for token in tokens[:12] if token in pages.get(page_no, ""))
|
if hit_count >= max(2, min(5, len(tokens) // 2)):
|
return "TEXT_MATCH_PARTIAL_CONTEXT", page_no, pages.get(page_no, "")[:700]
|
return "NO_TEXT_MATCH", page_no, pages.get(page_no, "")[:700]
|
|
|
def source_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_FACT_CANDIDATE"
|
return "TEXT_FACT_OR_VIEW_CANDIDATE"
|
|
|
def scope_gate(text):
|
compact = clean_text(text)
|
if "........" in compact or re.search(r"^\d+(?:\.\d+)+\s+.*\.{5,}\s*\d+", compact):
|
return "TOC_OR_INDEX_NOISE"
|
if "公司公告" in compact or any(word in compact for word in ADJACENT_OR_NOISE_WORDS):
|
return "MARKET_NEWS_OR_ADJACENT_REVIEW"
|
return "CORE_INDUSTRY_CANDIDATE"
|
|
|
def next_action(row, review_status, expression_type):
|
text = row.get("evidence_text", "")
|
scope = scope_gate(text)
|
if scope == "TOC_OR_INDEX_NOISE":
|
return "EXCLUDE_FROM_CORE_EVIDENCE_AS_TOC_NOISE"
|
if scope == "MARKET_NEWS_OR_ADJACENT_REVIEW":
|
return "ROUTE_TO_MARKET_OR_ADJACENT_REVIEW_BEFORE_USE"
|
themes = set(split_tags(row.get("theme_tags")))
|
if review_status in {"NO_TEXT_FILE", "NO_TEXT_MATCH"}:
|
return "CHECK_CONVERTED_OR_RAW_SOURCE"
|
if expression_type == "BROKER_VIEW_OR_ORIGINAL_RECOMMENDATION":
|
return "KEEP_AS_BROKER_VIEW_DO_NOT_REWRITE_AS_ANALYST_RECOMMENDATION"
|
if {"price", "inventory", "supply", "demand", "cost", "project"} & themes or re.search(r"\d", text):
|
return "ADD_PARAGRAPH_TABLE_OR_EXTERNAL_CROSS_CHECK"
|
return "KEEP_DRAFT_OR_USE_AS_CONTEXT"
|
|
|
def upgrade_status(row, review_status, expression_type):
|
scope = scope_gate(row.get("evidence_text", ""))
|
if scope == "TOC_OR_INDEX_NOISE":
|
return "EXCLUDE_FROM_CORE_EVIDENCE"
|
if scope == "MARKET_NEWS_OR_ADJACENT_REVIEW":
|
return "KEEP_DRAFT_SCOPE_REVIEW"
|
if review_status not in {"TEXT_MATCH_CONTEXT_OK", "TEXT_MATCH_PARTIAL_CONTEXT", "TEXT_MATCH_DIFFERENT_PAGE"}:
|
return "DATA_GAP_REVIEW"
|
if expression_type == "BROKER_VIEW_OR_ORIGINAL_RECOMMENDATION":
|
return "KEEP_DRAFT_BROKER_VIEW"
|
if row.get("fact_type") == "indicator_sentence":
|
return "CANDIDATE_FOR_EVIDENCE_CARD"
|
return "KEEP_DRAFT"
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--fact-files", nargs="+", required=True)
|
parser.add_argument("--review-out", required=True)
|
parser.add_argument("--manifest-out", required=True)
|
parser.add_argument("--summary-out", required=True)
|
parser.add_argument("--per-metal", type=int, default=8)
|
parser.add_argument("--per-theme", type=int, default=12)
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
facts = []
|
for fact_file in args.fact_files:
|
path = project_root / fact_file
|
if path.exists():
|
facts.extend(read_csv(path))
|
|
selected = {}
|
for metal in METAL_ORDER:
|
rows = [row for row in facts if metal in split_tags(row.get("metal_tags"))]
|
for row in sorted(rows, key=score_fact, reverse=True)[: args.per_metal]:
|
selected[row.get("fact_id")] = row
|
for theme in THEME_PRIORITY:
|
rows = [row for row in facts if theme in split_tags(row.get("theme_tags"))]
|
for row in sorted(rows, key=score_fact, reverse=True)[: args.per_theme]:
|
selected[row.get("fact_id")] = row
|
|
selected_rows = sorted(selected.values(), key=score_fact, reverse=True)
|
created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
review_rows = []
|
for row in selected_rows:
|
review_status, matched_page, context = find_context(row, project_root)
|
expression_type = source_expression_type(row.get("evidence_text", ""))
|
scope = scope_gate(row.get("evidence_text", ""))
|
review_rows.append(
|
{
|
"review_id": f"YS-FACT-REV-001-{len(review_rows) + 1:04d}",
|
"case_id": row.get("case_id", "ANA-YS-INDUSTRY-001"),
|
"batch_id": row.get("batch_id", ""),
|
"sub_batch_id": row.get("sub_batch_id", ""),
|
"run_id": "RUN-ANA-YS-FACT-REVIEW-001",
|
"fact_id": row.get("fact_id", ""),
|
"doc_id": row.get("doc_id", ""),
|
"fact_type": row.get("fact_type", ""),
|
"metal_tags": row.get("metal_tags", ""),
|
"theme_tags": row.get("theme_tags", ""),
|
"evidence_text": row.get("evidence_text", ""),
|
"declared_page_no": row.get("page_no", ""),
|
"matched_page_no": matched_page,
|
"source_location": row.get("source_location", ""),
|
"converted_text_path": row.get("converted_text_path", ""),
|
"raw_file_path": row.get("raw_file_path", ""),
|
"raw_sha256": row.get("raw_sha256", ""),
|
"source_expression_type": expression_type,
|
"scope_gate": scope,
|
"analyst_review_status": review_status,
|
"evidence_upgrade_status": upgrade_status(row, review_status, expression_type),
|
"next_action": next_action(row, review_status, expression_type),
|
"page_context": context[:900],
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
)
|
|
review_path = project_root / args.review_out
|
manifest_path = project_root / args.manifest_out
|
summary_path = project_root / args.summary_out
|
fields = [
|
"review_id",
|
"case_id",
|
"batch_id",
|
"sub_batch_id",
|
"run_id",
|
"fact_id",
|
"doc_id",
|
"fact_type",
|
"metal_tags",
|
"theme_tags",
|
"evidence_text",
|
"declared_page_no",
|
"matched_page_no",
|
"source_location",
|
"converted_text_path",
|
"raw_file_path",
|
"raw_sha256",
|
"source_expression_type",
|
"scope_gate",
|
"analyst_review_status",
|
"evidence_upgrade_status",
|
"next_action",
|
"page_context",
|
"review_status",
|
"created_at",
|
]
|
write_csv(review_path, fields, review_rows)
|
|
manifest_rows = [
|
{
|
"case_id": "ANA-YS-INDUSTRY-001",
|
"batch_id": "BATCH-001+BATCH-003",
|
"run_id": "RUN-ANA-YS-FACT-REVIEW-001",
|
"artifact_type": "key_fact_review",
|
"artifact_path": review_path.relative_to(project_root).as_posix(),
|
"row_count": len(review_rows),
|
"sha256": short_hash(review_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,
|
)
|
|
metal_counts = Counter()
|
theme_counts = Counter()
|
review_counts = Counter()
|
upgrade_counts = Counter()
|
action_counts = Counter()
|
scope_counts = Counter()
|
for row in review_rows:
|
for metal in split_tags(row.get("metal_tags")):
|
metal_counts[metal] += 1
|
for theme in split_tags(row.get("theme_tags")):
|
theme_counts[theme] += 1
|
review_counts[row.get("analyst_review_status")] += 1
|
upgrade_counts[row.get("evidence_upgrade_status")] += 1
|
action_counts[row.get("next_action")] += 1
|
scope_counts[row.get("scope_gate")] += 1
|
|
lines = [
|
"# 关键事实复核包 PASS-001 摘要",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{created_at}",
|
"",
|
"## 输入",
|
"",
|
"- BATCH-001 事实句草表",
|
"- BATCH-003 事实句草表",
|
"- converted text 页码标记和上下文",
|
"",
|
"## 输出",
|
"",
|
f"- 复核表：`{review_path.relative_to(project_root).as_posix()}`",
|
f"- manifest：`{manifest_path.relative_to(project_root).as_posix()}`",
|
f"- 复核记录数：{len(review_rows)}",
|
"",
|
"## 复核状态",
|
"",
|
"| 状态 | 数量 |",
|
"|---|---:|",
|
]
|
for key, count in review_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## 证据升级状态", "", "| 状态 | 数量 |", "|---|---:|"])
|
for key, count in upgrade_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## 下一步动作", "", "| 动作 | 数量 |", "|---|---:|"])
|
for key, count in action_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(["", "## 范围闸门", "", "| 范围 | 数量 |", "|---|---:|"])
|
for key, count in scope_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(["", "## 主题覆盖", "", "| 主题 | 数量 |", "|---|---:|"])
|
for key, count in theme_counts.most_common():
|
lines.append(f"| {key} | {count} |")
|
lines.extend(
|
[
|
"",
|
"## 边界",
|
"",
|
"本轮只做 converted text 页码上下文核对和证据升级候选标注。`TEXT_MATCH_CONTEXT_OK` 不等于正式证据通过；关键指标仍需段落、表格定位和必要外部交叉验证。",
|
]
|
)
|
summary_path.parent.mkdir(parents=True, exist_ok=True)
|
summary_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
print(f"review_rows={len(review_rows)}")
|
print(f"review={review_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()
|
