import argparse import csv import hashlib import importlib.util 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 file_header(path, size=8): try: return path.read_bytes()[:size].hex() except OSError: return "" def is_pdf_header(path): try: return path.read_bytes()[:5] == b"%PDF-" except OSError: return False def rel(path, project_root): try: return path.resolve().relative_to(project_root.resolve()).as_posix() except ValueError: return path.as_posix() def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--location-input", required=True) parser.add_argument("--card-input", required=True) parser.add_argument("--output", required=True) parser.add_argument("--manifest", required=True) parser.add_argument("--summary", required=True) parser.add_argument("--image-root", required=True) parser.add_argument("--run-id", default="RUN-ANA-YS-PDF-TABLE-004") args = parser.parse_args() project_root = Path(args.project_root).resolve() locations = [ row for row in read_csv(project_root / args.location_input) if row.get("location_type_candidate") == "TABLE_CANDIDATE" ] cards = {row.get("evidence_card_id"): row for row in read_csv(project_root / args.card_input)} fitz_available = importlib.util.find_spec("fitz") is not None created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") image_root = project_root / args.image_root image_root.mkdir(parents=True, exist_ok=True) rows = [] for item in locations: card = cards.get(item.get("evidence_card_id"), {}) raw_rel = card.get("raw_file_path", "") raw_path = project_root / raw_rel raw_exists = raw_path.exists() page_no = item.get("precheck_page_no", "") page_known = page_no.isdigit() pdf_header_ok = is_pdf_header(raw_path) if raw_exists else False if not raw_exists: render_status = "HELD_BY_RAW_MISSING" failure_reason = "raw file path does not exist" elif not pdf_header_ok: render_status = "HELD_BY_NOT_PDF_HEADER" failure_reason = "raw file header is not %PDF-" elif not page_known: render_status = "HELD_BY_PAGE_UNKNOWN" failure_reason = "precheck page number is unknown" elif not fitz_available: render_status = "HELD_BY_ENV_PDF_RENDERER_MISSING" failure_reason = "PyMuPDF/fitz is not installed and local PDF CLI renderer is unavailable" else: render_status = "READY_FOR_RENDER" failure_reason = "" rows.append( { "pdf_table_review_id": f"YS-PDF-TABLE-004-{len(rows) + 1:04d}", "case_id": item.get("case_id", ""), "batch_id": item.get("batch_id", ""), "run_id": args.run_id, "evidence_card_id": item.get("evidence_card_id", ""), "fact_id": item.get("fact_id", ""), "doc_id": item.get("doc_id", ""), "raw_file_path": raw_rel, "raw_exists": "YES" if raw_exists else "NO", "raw_header_hex": file_header(raw_path) if raw_exists else "", "pdf_header_ok": "YES" if pdf_header_ok else "NO", "precheck_page_no": page_no, "page_known": "YES" if page_known else "NO", "converted_text_path": item.get("converted_text_path", ""), "precheck_location": item.get("precheck_location", ""), "image_path": "", "pdf_page_render_status": render_status, "failure_reason": failure_reason, "table_header_review_status": "NEEDS_MANUAL_REVIEW", "unit_review_status": "NEEDS_MANUAL_REVIEW", "date_review_status": "NEEDS_MANUAL_REVIEW", "source_review_status": "NEEDS_MANUAL_REVIEW", "evidence_text": item.get("evidence_text", ""), "location_context": item.get("location_context", ""), "review_status": "DRAFT_FOR_REVIEW" if render_status == "READY_FOR_RENDER" else "HELD_BY_ENV", "created_at": created_at, } ) output_path = project_root / args.output fields = list(rows[0].keys()) if rows else [ "pdf_table_review_id", "case_id", "batch_id", "run_id", "review_status", ] write_csv(output_path, fields, rows) output_sha = sha256_file(output_path) manifest_path = project_root / args.manifest status_counts = Counter(row["pdf_page_render_status"] for row in rows) review_status_counts = Counter(row["review_status"] for row in rows) manifest_rows = [ { "case_id": "ANA-YS-INDUSTRY-001", "batch_id": "BATCH-001+BATCH-003", "run_id": args.run_id, "artifact_type": "key_fact_pdf_table_review_queue", "artifact_path": rel(output_path, project_root), "row_count": str(len(rows)), "sha256": output_sha, "review_status": "HELD_BY_ENV" if any(k.startswith("HELD") for k in status_counts) else "DRAFT_FOR_REVIEW", "created_at": created_at, } ] write_csv( manifest_path, ["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"], manifest_rows, ) summary_path = project_root / args.summary known_pages = sum(1 for row in rows if row["page_known"] == "YES") pdf_headers = sum(1 for row in rows if row["pdf_header_ok"] == "YES") summary_lines = [ "# \u5173\u952e\u4e8b\u5b9e PDF \u539f\u9875\u8868\u683c\u590d\u6838\u961f\u5217 PASS-004 \u6458\u8981", "", "\u72b6\u6001\uff1aHELD_BY_ENV", f"\u751f\u6210\u65f6\u95f4\uff1a{created_at}", "", "## \u8f93\u51fa", "", f"- \u590d\u6838\u961f\u5217\uff1a`{rel(output_path, project_root)}`", f"- manifest\uff1a`{rel(manifest_path, project_root)}`", f"- \u9875\u56fe\u76ee\u5f55\uff1a`{rel(image_root, project_root)}`", f"- \u8bb0\u5f55\u6570\uff1a{len(rows)}", f"- sha256\uff1a`{output_sha}`", "", "## \u81ea\u68c0\u7ed3\u679c", "", f"- \u8868\u683c\u5019\u9009\u8f93\u5165\uff1a{len(rows)}", f"- raw PDF \u6587\u4ef6\u5934\u901a\u8fc7\uff1a{pdf_headers}", f"- \u9875\u7801\u5df2\u77e5\uff1a{known_pages}", f"- PyMuPDF/fitz \u53ef\u7528\uff1a{'YES' if fitz_available else 'NO'}", "", "## PDF \u9875\u9762\u72b6\u6001", "", "| \u72b6\u6001 | \u6570\u91cf |", "|---|---:|", ] for key, count in status_counts.most_common(): summary_lines.append(f"| {key} | {count} |") summary_lines.extend(["", "## review_status", "", "| \u72b6\u6001 | \u6570\u91cf |", "|---|---:|"]) for key, count in review_status_counts.most_common(): summary_lines.append(f"| {key} | {count} |") summary_lines.extend( [ "", "## \u8fb9\u754c", "", "\u672c\u8f6e\u56e0 PDF \u89e3\u6790/\u6e32\u67d3\u4f9d\u8d56\u7f3a\u5931\uff0c\u672a\u80fd\u5bfc\u51fa PDF \u539f\u9875\u56fe\u7247\uff0c\u53ea\u5f62\u6210\u8868\u683c\u590d\u6838\u961f\u5217\u548c\u73af\u5883\u7f3a\u53e3\u8bb0\u5f55\u3002\u8be5\u4ea7\u7269\u4e0d\u662f PDF \u539f\u9875\u4eba\u5de5\u590d\u6838\u901a\u8fc7\uff0c\u4e0d\u80fd\u5347\u7ea7\u4e3a\u6b63\u5f0f\u4e8b\u5b9e\u6216\u6b63\u5f0f\u6307\u6807\u3002", ] ) summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text("\n".join(summary_lines) + "\n", encoding="utf-8") print(f"rows={len(rows)}") print(f"pdf_headers={pdf_headers}") print(f"known_pages={known_pages}") print(f"fitz_available={'YES' if fitz_available else 'NO'}") print(f"output={rel(output_path, project_root)}") print(f"manifest={rel(manifest_path, project_root)}") print(f"summary={rel(summary_path, project_root)}") if __name__ == "__main__": main()