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()
|
