"""Convert nonferrous raw reports for an approved batch.
|
|
This script is intentionally narrow: it reads a raw inventory CSV, converts
|
PDF files with pypdf, extracts basic text from Office zip containers when
|
possible, and writes conversion status rows. It does not create conclusions.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import re
|
import zipfile
|
from pathlib import Path
|
|
from pypdf import PdfReader
|
|
|
def clean_text(text: str) -> str:
|
text = text.replace("\x00", " ")
|
text = re.sub(r"[ \t]+", " ", text)
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
return text.strip()
|
|
|
def convert_pdf(path: Path) -> tuple[str, int, str]:
|
reader = PdfReader(str(path))
|
parts: list[str] = []
|
for idx, page in enumerate(reader.pages, start=1):
|
page_text = page.extract_text() or ""
|
parts.append(f"\n\n[[PAGE {idx}]]\n{page_text}")
|
text = clean_text("\n".join(parts))
|
return text, len(reader.pages), ""
|
|
|
def convert_office_zip(path: Path) -> tuple[str, int, str]:
|
parts: list[str] = []
|
with zipfile.ZipFile(path) as zf:
|
names = zf.namelist()
|
for name in names:
|
lower = name.lower()
|
if not (
|
lower.endswith(".xml")
|
and (
|
lower.startswith("word/")
|
or lower.startswith("ppt/")
|
or lower.startswith("xl/sharedstrings")
|
)
|
):
|
continue
|
raw = zf.read(name).decode("utf-8", errors="ignore")
|
raw = re.sub(r"<[^>]+>", " ", raw)
|
raw = re.sub(r"\s+", " ", raw)
|
if raw.strip():
|
parts.append(f"[[{name}]]\n{raw.strip()}")
|
if not parts:
|
return "", 0, "no readable Office XML text found"
|
return clean_text("\n\n".join(parts)), 0, ""
|
|
|
def main() -> int:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--inventory", required=True)
|
parser.add_argument("--converted-dir", required=True)
|
parser.add_argument("--status-out", required=True)
|
parser.add_argument("--case-id", required=True)
|
parser.add_argument("--batch-id", required=True)
|
parser.add_argument("--run-id", required=True)
|
parser.add_argument("--only-unregistered", action="store_true")
|
args = parser.parse_args()
|
|
root = Path(args.project_root).resolve()
|
inventory = root / args.inventory
|
converted_dir = root / args.converted_dir
|
converted_dir.mkdir(parents=True, exist_ok=True)
|
status_out = root / args.status_out
|
status_out.parent.mkdir(parents=True, exist_ok=True)
|
|
with inventory.open("r", encoding="utf-8-sig", newline="") as f:
|
rows = list(csv.DictReader(f))
|
|
status_rows: list[dict[str, str | int]] = []
|
for idx, row in enumerate(rows, start=1):
|
if args.only_unregistered and row.get("already_in_source_document") == "YES":
|
continue
|
raw_path = root / row["raw_file_path"]
|
stem = raw_path.name
|
out_name = f"{stem}.txt" if not stem.lower().endswith(".txt") else stem
|
out_path = converted_dir / out_name
|
error = ""
|
page_count = 0
|
char_count = 0
|
status = "FAILED"
|
try:
|
detected = row.get("detected_type", "")
|
if detected == "PDF":
|
text, page_count, error = convert_pdf(raw_path)
|
elif detected == "ZIP_OR_OFFICE":
|
text, page_count, error = convert_office_zip(raw_path)
|
else:
|
text, page_count, error = "", 0, f"unsupported detected_type={detected}"
|
if text and not error:
|
out_path.write_text(text, encoding="utf-8")
|
char_count = len(text)
|
status = "TEXT_CONVERTED"
|
elif not error:
|
error = "empty converted text"
|
except Exception as exc: # keep batch running and record the failure
|
error = f"{type(exc).__name__}: {exc}"
|
|
status_rows.append(
|
{
|
"conversion_id": f"YS-B3-CONV-{idx:04d}",
|
"source_doc_id": row.get("inventory_id", ""),
|
"case_id": args.case_id,
|
"batch_id": args.batch_id,
|
"run_id": args.run_id,
|
"raw_file_path": row.get("raw_file_path", ""),
|
"raw_sha256": row.get("sha256", ""),
|
"detected_type": row.get("detected_type", ""),
|
"conversion_tool": "pypdf_or_office_zip_xml",
|
"converted_text_path": out_path.relative_to(root).as_posix() if status == "TEXT_CONVERTED" else "",
|
"page_count": page_count,
|
"char_count": char_count,
|
"conversion_status": status,
|
"error_message": error,
|
}
|
)
|
|
if status_rows:
|
with status_out.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=list(status_rows[0].keys()))
|
writer.writeheader()
|
writer.writerows(status_rows)
|
print(f"status_rows={len(status_rows)}")
|
print(status_out.relative_to(root))
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
