import argparse import csv import hashlib import re from collections import Counter from datetime import datetime, timedelta, timezone from pathlib import Path PAGE_RE = re.compile(r"\[\[PAGE\s+(\d+)\]\]") 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 read_lines(path): try: return path.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: return [] def file_head(path, size=16): try: return path.read_bytes()[:size].hex() except OSError: return "" def file_size(path): try: return str(path.stat().st_size) except OSError: return "0" def normalize(text): return re.sub(r"\s+", "", text or "") def find_line(lines, evidence): needle = normalize(evidence) if not needle: return None for i, line in enumerate(lines, start=1): if needle in normalize(line): return i prefix = needle[:80] if not prefix: return None for i in range(len(lines)): window = "".join(normalize(x) for x in lines[max(0, i - 2) : min(len(lines), i + 3)]) if prefix in window: return i + 1 return None def page_before(lines, line_no): page = "" page_line = "" for i, line in enumerate(lines, start=1): match = PAGE_RE.search(line) if match and i <= line_no: page = match.group(1) page_line = str(i) if i > line_no: break return page, page_line def nearest_page_from_text(lines): for i, line in enumerate(lines, start=1): match = PAGE_RE.search(line) if match: return match.group(1), str(i) return "", "" def classify_non_pdf(raw_path, converted_path): head = file_head(raw_path) suffix = raw_path.suffix.lower() if head.startswith("504b0304"): return "ZIP_OR_OFFICE_HEADER", "route to Office/ZIP conversion or source anomaly review" if suffix in [".pptx", ".docx", ".xlsx", ".zip"]: return "OFFICE_OR_ZIP_SUFFIX", "route to Office/ZIP conversion or source anomaly review" if converted_path.suffix.lower() == ".txt" and converted_path.exists(): return "CONVERTED_TEXT_EXISTS_BUT_RAW_NOT_PDF", "verify original file type and conversion provenance" return "UNKNOWN_NON_PDF_HEADER", "manual source file inspection required" def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-root", default=".") parser.add_argument("--queue-input", required=True) parser.add_argument("--location-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", default="RUN-ANA-YS-PAGE-GAP-005") args = parser.parse_args() project_root = Path(args.project_root).resolve() queue_rows = read_csv(project_root / args.queue_input) location_rows = {row.get("evidence_card_id"): row for row in read_csv(project_root / args.location_input)} targets = [ row for row in queue_rows if row.get("pdf_page_render_status") in {"HELD_BY_PAGE_UNKNOWN", "HELD_BY_NOT_PDF_HEADER"} ] created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds") rows = [] for row in targets: loc = location_rows.get(row.get("evidence_card_id"), {}) raw_path = project_root / row.get("raw_file_path", "") converted_path = project_root / row.get("converted_text_path", "") original_status = row.get("pdf_page_render_status", "") lines = read_lines(converted_path) evidence = row.get("evidence_text", "") line_no = find_line(lines, evidence) inferred_page = "" page_marker_line = "" repair_status = "" repair_action = "" anomaly_type = "" anomaly_action = "" if original_status == "HELD_BY_PAGE_UNKNOWN": if line_no: inferred_page, page_marker_line = page_before(lines, line_no) if not inferred_page: inferred_page, page_marker_line = nearest_page_from_text(lines) if inferred_page: repair_status = "PAGE_INFERRED_FROM_CONVERTED_MARKER" repair_action = "use inferred page as candidate only; verify against PDF original after renderer restored" else: repair_status = "NEEDS_MANUAL_PAGE_RECONSTRUCTION" repair_action = "open source document manually and reconstruct page number from context" else: anomaly_type, anomaly_action = classify_non_pdf(raw_path, converted_path) repair_status = "NON_PDF_HEADER_CLASSIFIED" repair_action = anomaly_action rows.append( { "page_gap_repair_id": f"YS-PAGE-GAP-005-{len(rows) + 1:04d}", "case_id": row.get("case_id", ""), "batch_id": row.get("batch_id", ""), "run_id": args.run_id, "evidence_card_id": row.get("evidence_card_id", ""), "fact_id": row.get("fact_id", ""), "doc_id": row.get("doc_id", ""), "original_pdf_page_render_status": original_status, "raw_file_path": row.get("raw_file_path", ""), "raw_exists": row.get("raw_exists", ""), "raw_size_bytes": file_size(raw_path), "raw_header_hex": file_head(raw_path), "converted_text_path": row.get("converted_text_path", ""), "precheck_location": row.get("precheck_location", ""), "located_line_no": str(line_no or ""), "candidate_page_no": inferred_page, "candidate_page_marker_line": page_marker_line, "non_pdf_anomaly_type": anomaly_type, "repair_status": repair_status, "next_action": repair_action, "evidence_text": evidence, "location_context": row.get("location_context", "")[:1200], "review_status": "DRAFT_FOR_REVIEW" if repair_status == "PAGE_INFERRED_FROM_CONVERTED_MARKER" else "HELD_FOR_MANUAL_REVIEW", "created_at": created_at, } ) output_path = project_root / args.output fields = list(rows[0].keys()) if rows else ["page_gap_repair_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-001+BATCH-003", "run_id": args.run_id, "artifact_type": "key_fact_page_gap_repair", "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, } ], ) status_counts = Counter(row["repair_status"] for row in rows) review_counts = Counter(row["review_status"] for row in rows) summary_path = project_root / args.summary lines = [ "# \u5173\u952e\u4e8b\u5b9e\u9875\u7801\u7f3a\u53e3\u548c\u975e PDF \u5f02\u5e38\u8865\u67e5 PASS-005 \u6458\u8981", "", "\u72b6\u6001\uff1aDRAFT_FOR_REVIEW", f"\u751f\u6210\u65f6\u95f4\uff1a{created_at}", "", "## \u8f93\u51fa", "", f"- \u8865\u67e5\u8868\uff1a`{output_path.relative_to(project_root).as_posix()}`", f"- manifest\uff1a`{manifest_path.relative_to(project_root).as_posix()}`", f"- \u8bb0\u5f55\u6570\uff1a{len(rows)}", f"- sha256\uff1a`{output_sha}`", "", "## \u8865\u67e5\u72b6\u6001", "", "| \u72b6\u6001 | \u6570\u91cf |", "|---|---:|", ] for key, count in status_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend(["", "## review_status", "", "| \u72b6\u6001 | \u6570\u91cf |", "|---|---:|"]) for key, count in review_counts.most_common(): lines.append(f"| {key} | {count} |") lines.extend( [ "", "## \u8fb9\u754c", "", "\u672c\u8f6e\u53ea\u662f\u9875\u7801\u7f3a\u53e3\u548c\u975e PDF \u6587\u4ef6\u5934\u5f02\u5e38\u8865\u67e5\u3002\u4ece converted text \u56de\u63a8\u7684\u9875\u7801\u53ea\u80fd\u4f5c\u4e3a\u5019\u9009\uff0c\u4e0d\u7b49\u4e8e PDF \u539f\u9875\u8868\u683c\u590d\u6838\u901a\u8fc7\uff1b\u975e PDF \u5f02\u5e38\u4e0d\u5f97\u4f2a\u88c5\u6210 PDF \u6210\u529f\u5904\u7406\u3002", ] ) 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(status_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()