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