import argparse import csv import hashlib from collections import Counter 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 recovery_plan(source_type: str, archive_status: str): source_type = source_type or "" if archive_status != "HELD_BY_ENV_NETWORK": return ( "ARCHIVE_RECHECK", "review_existing_archive_then_extract_original_line", "archive path/hash, source timestamp, original line or table row, value consistency result", ) if "media" in source_type: return ( "REPLACE_SECONDARY_POINTER_WITH_PRIMARY_SOURCE", "use media page only as locator; obtain exchange filing, company announcement, or annual report as primary source", "primary filing/archive path/hash, source timestamp, original announcement or annual report line", ) if "pdf" in source_type or "annual_report" in source_type or "announcement" in source_type: return ( "MANUAL_DOWNLOAD_OR_BROWSER_ARCHIVE", "download or browser-archive PDF/filing; register local file path and hash before extraction", "local PDF/html path, sha256, source timestamp, operator, original page/table row", ) return ( "BROWSER_SNAPSHOT_OR_APPROVED_NETWORK_ARCHIVE", "retry with browser or approved network path; if still blocked, replace with exchange filing or company IR source", "snapshot/html/pdf path, sha256, source timestamp, operator, original line", ) def build_rows(input_rows, run_id: str, created_at: str): rows = [] for idx, row in enumerate(input_rows, start=1): recovery_type, next_action, required_artifacts = recovery_plan( row.get("source_type", ""), row.get("archive_status", "") ) rows.append( { "archive_recovery_queue_id": f"YS-COMPANY-ARCHIVE-RECOVERY-014-{idx:04d}", "case_id": row.get("case_id", ""), "batch_id": row.get("batch_id", ""), "run_id": run_id, "company_source_archive_review_id": row.get("company_source_archive_review_id", ""), "company_official_source_id": row.get("company_official_source_id", ""), "gap_priority_id": row.get("gap_priority_id", ""), "evidence_card_id": row.get("evidence_card_id", ""), "fact_id": row.get("fact_id", ""), "doc_id": row.get("doc_id", ""), "company_name": row.get("company_name", ""), "stock_code": row.get("stock_code", ""), "source_type": row.get("source_type", ""), "source_name": row.get("source_name", ""), "source_url": row.get("source_url", ""), "source_period_or_date": row.get("source_period_or_date", ""), "prior_archive_status": row.get("archive_status", ""), "prior_fetch_status": row.get("archive_fetch_status", ""), "prior_error": row.get("archive_error", ""), "archive_recovery_type": recovery_type, "queue_status": "QUEUE_ARCHIVE_RECOVERY", "queue_priority": "HIGH", "required_artifacts": required_artifacts, "next_action": next_action, "formal_upgrade_blockers": "archive_path_or_hash_missing; original_line_or_table_row_missing; value_consistency_not_verified", "evidence_status": "ARCHIVE_RECOVERY_DRAFT_QUEUE", "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_company = Counter(r["company_name"] for r in rows) by_type = Counter(r["archive_recovery_type"] for r in rows) lines = [ "# 公司官方源归档恢复队列 PASS-014 摘要", "", "状态：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_type.items(): lines.append(f"| {key} | {count} |") lines += ["", "## 公司分布", "", "| 公司 | 数量 |", "|---|---:|"] for key, count in by_company.items(): lines.append(f"| {key} | {count} |") lines += [ "", "## 边界", "", "本轮只把 PASS-013 的公司官方源归档失败项拆成浏览器归档、人工下载或替代主源队列。队列不构成归档成功、官方数值核验通过或正式公司证据；正式升级前仍需补本地文件路径、hash、来源时间、操作者记录、原文页/表格行和数值一致性结果。", "", ] 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("--input", 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") input_path = Path(args.input) output_path = Path(args.output) manifest_path = Path(args.manifest) summary_path = Path(args.summary) rows = build_rows(read_csv(input_path), args.run_id, created_at) fieldnames = list(rows[0].keys()) if rows else [ "archive_recovery_queue_id", "case_id", "batch_id", "run_id", "review_status", ] write_csv(output_path, fieldnames, rows) digest = sha256_file(output_path) manifest_rows = [ { "case_id": rows[0]["case_id"] if rows else "", "batch_id": rows[0]["batch_id"] if rows else "", "run_id": args.run_id, "artifact_type": "company_source_archive_recovery_queue", "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(summary_path, output_path, manifest_path, rows, digest, created_at) print(f"OK rows={len(rows)} output={output_path} sha256={digest}") if __name__ == "__main__": main()