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