import argparse
|
import csv
|
import hashlib
|
import re
|
from collections import Counter
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
|
VALUE_UNIT_RE = re.compile(
|
r"([-+]?\d+(?:\.\d+)?)\s*(%|万吨|吨|亿元|元/吨|美元/吨|美元/盎司|万元/吨|万\w*吨|GWh|万吨/年|吨/年|万股|亿元/年|万吨金属量)"
|
)
|
DATE_RE = re.compile(r"(20\d{2}[年/-]\d{1,2}(?:[月/-]\d{1,2}日?)?|20\d{2}年|Q[1-4]|[一二三四]季度|本周|上周|本月)")
|
|
|
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 sha256_file(path):
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
def extract_values(text):
|
pairs = []
|
for value, unit in VALUE_UNIT_RE.findall(text or ""):
|
pairs.append(f"{value}{unit}")
|
return "|".join(pairs[:8])
|
|
|
def extract_dates(text):
|
return "|".join(DATE_RE.findall(text or "")[:6])
|
|
|
def evidence_kind(theme_tags):
|
themes = set((theme_tags or "").split("|"))
|
if "price" in themes:
|
return "PRICE_OR_MARKET_DATA"
|
if "inventory" in themes:
|
return "INVENTORY"
|
if "supply" in themes:
|
return "SUPPLY_CAPACITY_OR_OUTPUT"
|
if "demand" in themes:
|
return "DEMAND"
|
if "project" in themes:
|
return "PROJECT_OR_ASSET"
|
if "cost" in themes:
|
return "COST_OR_MARGIN"
|
if "policy" in themes:
|
return "POLICY"
|
return "FACT_CANDIDATE"
|
|
|
def next_action(row, values):
|
if not values:
|
return "NEEDS_MANUAL_NUMERIC_OR_TABLE_REVIEW"
|
if "price" in row.get("theme_tags", "") or "inventory" in row.get("theme_tags", ""):
|
return "ADD_TABLE_LOCATION_AND_EXTERNAL_CROSS_CHECK"
|
return "ADD_PARAGRAPH_LOCATION_AND_SOURCE_CONTEXT"
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--review", required=True)
|
parser.add_argument("--cards-out", required=True)
|
parser.add_argument("--manifest-out", required=True)
|
parser.add_argument("--summary-out", required=True)
|
args = parser.parse_args()
|
|
project_root = Path(args.project_root).resolve()
|
rows = read_csv(project_root / args.review)
|
candidates = [
|
row
|
for row in rows
|
if row.get("evidence_upgrade_status") == "CANDIDATE_FOR_EVIDENCE_CARD"
|
and row.get("scope_gate") == "CORE_INDUSTRY_CANDIDATE"
|
]
|
created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
|
cards = []
|
for row in candidates:
|
text = row.get("evidence_text", "")
|
context = row.get("page_context", "")
|
values = extract_values(text + " " + context)
|
dates = extract_dates(text + " " + context)
|
cards.append(
|
{
|
"evidence_card_id": f"YS-FACT-CARD-002-{len(cards) + 1:04d}",
|
"case_id": row.get("case_id", "ANA-YS-INDUSTRY-001"),
|
"batch_id": row.get("batch_id", ""),
|
"run_id": "RUN-ANA-YS-FACT-CARD-002",
|
"fact_id": row.get("fact_id", ""),
|
"doc_id": row.get("doc_id", ""),
|
"metal_tags": row.get("metal_tags", ""),
|
"theme_tags": row.get("theme_tags", ""),
|
"evidence_kind": evidence_kind(row.get("theme_tags", "")),
|
"source_expression_type": row.get("source_expression_type", ""),
|
"declared_page_no": row.get("declared_page_no", ""),
|
"matched_page_no": row.get("matched_page_no", ""),
|
"converted_text_path": row.get("converted_text_path", ""),
|
"raw_file_path": row.get("raw_file_path", ""),
|
"raw_sha256": row.get("raw_sha256", ""),
|
"evidence_text": text,
|
"value_unit_candidates": values,
|
"date_candidates": dates,
|
"page_context": context[:700],
|
"table_location_status": "NEEDS_TABLE_OR_PARAGRAPH_LOCATION",
|
"external_cross_check_status": "NEEDS_EXTERNAL_CROSS_CHECK",
|
"next_action": next_action(row, values),
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
)
|
|
cards_path = project_root / args.cards_out
|
fields = [
|
"evidence_card_id",
|
"case_id",
|
"batch_id",
|
"run_id",
|
"fact_id",
|
"doc_id",
|
"metal_tags",
|
"theme_tags",
|
"evidence_kind",
|
"source_expression_type",
|
"declared_page_no",
|
"matched_page_no",
|
"converted_text_path",
|
"raw_file_path",
|
"raw_sha256",
|
"evidence_text",
|
"value_unit_candidates",
|
"date_candidates",
|
"page_context",
|
"table_location_status",
|
"external_cross_check_status",
|
"next_action",
|
"review_status",
|
"created_at",
|
]
|
write_csv(cards_path, fields, cards)
|
|
manifest_path = project_root / args.manifest_out
|
manifest_rows = [
|
{
|
"case_id": "ANA-YS-INDUSTRY-001",
|
"batch_id": "BATCH-001+BATCH-003",
|
"run_id": "RUN-ANA-YS-FACT-CARD-002",
|
"artifact_type": "key_fact_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,
|
)
|
|
kind_counts = Counter(card["evidence_kind"] for card in cards)
|
action_counts = Counter(card["next_action"] for card in cards)
|
with_values = sum(1 for card in cards if card["value_unit_candidates"])
|
summary_path = project_root / args.summary_out
|
lines = [
|
"# 关键事实证据卡 PASS-002 摘要",
|
"",
|
"状态：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)}",
|
f"- 含数值/单位候选：{with_values}",
|
"",
|
"## 证据类型",
|
"",
|
"| 类型 | 数量 |",
|
"|---|---:|",
|
]
|
for key, count in kind_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(
|
[
|
"",
|
"## 边界",
|
"",
|
"本轮只是把关键事实候选拆成证据卡草表。数值/单位仍为候选，正式证据升级前必须补段落或表格定位，并完成外部交叉验证。",
|
]
|
)
|
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()
|
