import argparse
|
import csv
|
import hashlib
|
from collections import Counter, defaultdict
|
from datetime import datetime, timezone, timedelta
|
from pathlib import Path
|
|
|
CN_TZ = timezone(timedelta(hours=8))
|
|
|
def read_csv(path: Path):
|
with path.open("r", encoding="utf-8-sig", newline="") as f:
|
return list(csv.DictReader(f))
|
|
|
def write_csv(path: Path, fieldnames, rows):
|
path.parent.mkdir(parents=True, exist_ok=True)
|
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
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 split_values(value: str):
|
return [item.strip() for item in (value or "").split("|") if item.strip()]
|
|
|
def first_non_empty(values):
|
for value in values:
|
if value:
|
return value
|
return ""
|
|
|
def build_rows(company_cards, official_sources, recovery_rows, broker_rows, run_id, created_at):
|
official_by_company = defaultdict(list)
|
recovery_by_company = defaultdict(list)
|
broker_by_company = defaultdict(list)
|
for row in official_sources:
|
official_by_company[row.get("company_name", "")].append(row)
|
for row in recovery_rows:
|
recovery_by_company[row.get("company_name", "")].append(row)
|
for row in broker_rows:
|
broker_by_company[row.get("company_name_fragment", "")].append(row)
|
|
rows = []
|
for idx, card in enumerate(company_cards, start=1):
|
company = card.get("company_name", "")
|
official = official_by_company.get(company, [])
|
recovery = recovery_by_company.get(company, [])
|
broker = broker_by_company.get(company, [])
|
source_types = sorted({r.get("source_type", "") for r in official if r.get("source_type", "")})
|
recovery_types = sorted({r.get("archive_recovery_type", "") for r in recovery if r.get("archive_recovery_type", "")})
|
value_candidates = []
|
for row in official:
|
value_candidates.extend(split_values(row.get("value_unit_candidates", "")))
|
value_candidates = sorted(set(value_candidates))
|
has_official = bool(official)
|
has_recovery = bool(recovery)
|
has_broker_split = bool(broker)
|
rows.append(
|
{
|
"structured_company_card_id": f"YS-COMP-STRUCT-017-{idx:04d}",
|
"case_id": card.get("case_id", "ANA-YS-INDUSTRY-001"),
|
"batch_id": "BATCH-003",
|
"run_id": run_id,
|
"source_company_card_id": card.get("company_card_id", ""),
|
"company_name": company,
|
"mention_count": card.get("mention_count", ""),
|
"metal_tags": card.get("metal_tags_in_context", ""),
|
"stock_code": first_non_empty(row.get("stock_code", "") for row in official),
|
"broker_phrase_split_status": "SPLIT_LEDGER_LINKED" if has_broker_split else "NO_SPLIT_LEDGER_MATCH",
|
"broker_phrase_split_count": str(len(broker)),
|
"sample_context_policy": "DO_NOT_USE_AS_ANALYST_CONCLUSION" if has_broker_split else "RAW_SAMPLE_REVIEW_REQUIRED",
|
"official_source_status": "OFFICIAL_SOURCE_ENTRY_DRAFT" if has_official else "NEEDS_OFFICIAL_SOURCE",
|
"official_source_count": str(len(official)),
|
"official_source_types": "|".join(source_types),
|
"archive_recovery_status": "ARCHIVE_RECOVERY_QUEUE_LINKED" if has_recovery else "NO_ARCHIVE_RECOVERY_QUEUE",
|
"archive_recovery_count": str(len(recovery)),
|
"archive_recovery_types": "|".join(recovery_types),
|
"business_exposure_field": "TO_BE_EXTRACTED_FROM_OFFICIAL_SOURCE",
|
"resource_or_capacity_field": "TO_BE_EXTRACTED_FROM_OFFICIAL_SOURCE",
|
"profit_sensitivity_field": "TO_BE_MODELED_AFTER_VALUE_VERIFICATION",
|
"trigger_condition_field": "TO_BE_FILLED_AFTER_EVIDENCE_REVIEW",
|
"failure_condition_field": "TO_BE_FILLED_AFTER_EVIDENCE_REVIEW",
|
"risk_field": "TO_BE_FILLED_AFTER_EVIDENCE_REVIEW",
|
"value_unit_candidates": "|".join(value_candidates),
|
"official_archive_blockers": "archive_path_or_hash_missing; original_line_or_table_row_missing; value_consistency_not_verified" if has_recovery else "official_source_missing_or_not_prioritized",
|
"analyst_conclusion_status": "NOT_GENERATED_DRAFT_INPUT_ONLY",
|
"evidence_status": "STRUCTURED_DRAFT_FIELD_SHELL",
|
"next_action": "fill_official_archive_and_original_line_then_value_consistency" if has_official else "find_company_official_or_exchange_source",
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
)
|
return rows
|
|
|
def write_summary(path: Path, output_path: Path, manifest_path: Path, rows, digest: str, created_at: str):
|
by_official = Counter(r["official_source_status"] for r in rows)
|
by_split = Counter(r["broker_phrase_split_status"] for r in rows)
|
lines = [
|
"# 公司证据卡字段化 PASS-017 摘要",
|
"",
|
"状态：DRAFT_FOR_REVIEW",
|
f"生成时间：{created_at}",
|
"",
|
"## 输出",
|
"",
|
f"- 字段化公司证据卡：`{output_path.as_posix()}`",
|
f"- manifest：`{manifest_path.as_posix()}`",
|
f"- 记录数：{len(rows)}",
|
f"- sha256：`{digest}`",
|
"",
|
"## 官方源状态",
|
"",
|
"| 状态 | 数量 |",
|
"|---|---:|",
|
]
|
for key, count in by_official.items():
|
lines.append(f"| {key} | {count} |")
|
lines += ["", "## 券商原文分账链接", "", "| 状态 | 数量 |", "|---|---:|"]
|
for key, count in by_split.items():
|
lines.append(f"| {key} | {count} |")
|
lines += [
|
"",
|
"## 边界",
|
"",
|
"本轮只把公司候选卡字段化为证据卡草稿壳，并链接 PASS-016 分账清单、PASS-012 官方源入口和 PASS-014 归档恢复队列。所有公司字段仍为 DRAFT，不构成正式公司证据、正式投资读法、交易指令或收益承诺。",
|
"",
|
]
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text("\n".join(lines), encoding="utf-8")
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--company-cards", required=True)
|
parser.add_argument("--official-sources", required=True)
|
parser.add_argument("--archive-recovery", required=True)
|
parser.add_argument("--broker-split", required=True)
|
parser.add_argument("--output", 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()
|
created_at = datetime.now(CN_TZ).isoformat(timespec="seconds")
|
output_path = Path(args.output)
|
manifest_path = Path(args.manifest)
|
rows = build_rows(
|
read_csv(Path(args.company_cards)),
|
read_csv(Path(args.official_sources)),
|
read_csv(Path(args.archive_recovery)),
|
read_csv(Path(args.broker_split)),
|
args.run_id,
|
created_at,
|
)
|
fieldnames = list(rows[0].keys()) if rows else ["structured_company_card_id", "review_status"]
|
write_csv(output_path, fieldnames, rows)
|
digest = sha256_file(output_path)
|
manifest_rows = [
|
{
|
"case_id": "ANA-YS-INDUSTRY-001",
|
"batch_id": "BATCH-003",
|
"run_id": args.run_id,
|
"artifact_type": "structured_company_evidence_card",
|
"artifact_path": output_path.as_posix(),
|
"row_count": str(len(rows)),
|
"sha256": digest,
|
"review_status": "DRAFT_FOR_REVIEW",
|
"created_at": created_at,
|
}
|
]
|
write_csv(manifest_path, list(manifest_rows[0].keys()), manifest_rows)
|
write_summary(Path(args.summary), output_path, manifest_path, rows, digest, created_at)
|
print(f"OK rows={len(rows)} output={output_path} sha256={digest}")
|
|
|
if __name__ == "__main__":
|
main()
|
