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