import argparse import csv import hashlib from collections import Counter from datetime import datetime, timedelta, timezone from pathlib import Path 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 main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--source-input", required=True) parser.add_argument("--archive-manifest", 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", default="RUN-ANA-YS-COMPANY-SOURCE-ARCHIVE-013") args = parser.parse_args() project_root = Path(args.project_root).resolve() sources = read_csv(project_root / args.source_input) archive_rows = read_csv(project_root / args.archive_manifest) archive_by_url = {} for row in archive_rows: archive_by_url.setdefault(row.get("source_url", ""), []).append(row) created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") rows = [] for source in sources: url = source.get("source_url", "") matches = archive_by_url.get(url, []) match = matches.pop(0) if matches else {} fetch_status = match.get("fetch_status", "NOT_ATTEMPTED") if fetch_status == "FETCHED": archive_status = "ARCHIVED_DRAFT" evidence_status = "ARCHIVED_NOT_VALUE_VERIFIED" next_action = "extract_original_page_or_table_row_and_compare_value" else: archive_status = "HELD_BY_ENV_NETWORK" evidence_status = "SOURCE_ENTRY_DRAFT_ARCHIVE_FAILED" next_action = "retry_archive_with_browser_or_approved_network_path" rows.append( { "company_source_archive_review_id": f"YS-COMPANY-ARCHIVE-013-{len(rows) + 1:04d}", "case_id": source.get("case_id", ""), "batch_id": source.get("batch_id", ""), "run_id": args.run_id, "company_official_source_id": source.get("company_official_source_id", ""), "gap_priority_id": source.get("gap_priority_id", ""), "evidence_card_id": source.get("evidence_card_id", ""), "fact_id": source.get("fact_id", ""), "doc_id": source.get("doc_id", ""), "company_name": source.get("company_name", ""), "stock_code": source.get("stock_code", ""), "source_type": source.get("source_type", ""), "source_name": source.get("source_name", ""), "source_url": url, "source_period_or_date": source.get("source_period_or_date", ""), "archive_source_id": match.get("source_id", ""), "archive_fetch_status": fetch_status, "archive_http_status": match.get("http_status", ""), "archive_content_type": match.get("content_type", ""), "archive_bytes": match.get("bytes", ""), "archive_sha256": match.get("sha256", ""), "archive_relative_path": match.get("relative_path", ""), "archive_text_relative_path": match.get("text_relative_path", ""), "archive_error": match.get("error", ""), "archive_status": archive_status, "original_line_extract_status": "NOT_STARTED", "value_consistency_status": "NOT_VERIFIED", "evidence_status": evidence_status, "next_action": next_action, "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ) output_path = project_root / args.output fields = list(rows[0].keys()) if rows else ["company_source_archive_review_id", "case_id", "run_id", "review_status"] write_csv(output_path, fields, rows) output_sha = sha256_file(output_path) manifest_path = project_root / args.manifest write_csv( manifest_path, ["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"], [ { "case_id": "ANA-YS-INDUSTRY-001", "batch_id": "BATCH-003", "run_id": args.run_id, "artifact_type": "company_source_archive_review", "artifact_path": output_path.relative_to(project_root).as_posix(), "row_count": str(len(rows)), "sha256": output_sha, "review_status": "DRAFT_FOR_REVIEW", "created_at": created_at, } ], ) archive_counts = Counter(row["archive_status"] for row in rows) fetch_counts = Counter(row["archive_fetch_status"] for row in rows) company_counts = Counter(row["company_name"] for row in rows) summary_path = project_root / args.summary lines = [ "# 公司官方源归档复核 PASS-013 摘要", "", "状态：DRAFT_FOR_REVIEW", f"生成时间：{created_at}", "", "## 输出", "", f"- 归档复核表：`{output_path.relative_to(project_root).as_posix()}`", f"- manifest：`{manifest_path.relative_to(project_root).as_posix()}`", f"- 记录数：{len(rows)}", f"- sha256：`{output_sha}`", "", "## 归档状态", "", "| 状态 | 数量 |", "|---|---:|", ] for key, count in archive_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## 抓取状态", "", "| 状态 | 数量 |", "|---|---:|"]) for key, count in fetch_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## 公司分布", "", "| 公司 | 数量 |", "|---|---:|"]) for key, count in company_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend( [ "", "## 边界", "", "本轮执行公开 URL 归档尝试并记录状态；当前归档失败项只表示环境或网络路径缺口，不构成公司官方证据通过。正式证据升级前仍需网页/PDF 归档、hash、原文行或表格行抽取和数值一致性核验。", ] ) summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"rows={len(rows)}") print(dict(archive_counts)) print(dict(fetch_counts)) print(f"output={output_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()