"""Incrementally convert nonferrous reports with per-file timeout."""
|
|
from __future__ import annotations
|
|
import argparse
|
import csv
|
import multiprocessing as mp
|
import re
|
import zipfile
|
from pathlib import Path
|
|
|
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 worker(raw_path_s: str, detected_type: str, queue: mp.Queue) -> None:
|
raw_path = Path(raw_path_s)
|
try:
|
if detected_type == "PDF":
|
from pypdf import PdfReader
|
|
reader = PdfReader(str(raw_path))
|
parts: list[str] = []
|
for idx, page in enumerate(reader.pages, start=1):
|
parts.append(f"\n\n[[PAGE {idx}]]\n{page.extract_text() or ''}")
|
queue.put(("OK", clean_text("\n".join(parts)), len(reader.pages), ""))
|
return
|
if detected_type == "ZIP_OR_OFFICE":
|
parts: list[str] = []
|
with zipfile.ZipFile(raw_path) as zf:
|
for name in zf.namelist():
|
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).strip()
|
if raw:
|
parts.append(f"[[{name}]]\n{raw}")
|
text = clean_text("\n\n".join(parts))
|
if text:
|
queue.put(("OK", text, 0, ""))
|
else:
|
queue.put(("FAIL", "", 0, "no readable Office XML text found"))
|
return
|
queue.put(("FAIL", "", 0, f"unsupported detected_type={detected_type}"))
|
except Exception as exc:
|
queue.put(("FAIL", "", 0, f"{type(exc).__name__}: {exc}"))
|
|
|
def convert_one(root: Path, row: dict[str, str], converted_dir: Path, timeout: int) -> dict[str, str | int]:
|
raw_path = root / row["raw_file_path"]
|
out_path = converted_dir / f"{raw_path.name}.txt"
|
base = {
|
"conversion_id": row["inventory_id"].replace("YS-RAW-INV", "YS-B3-CONV"),
|
"source_doc_id": row["inventory_id"],
|
"case_id": row["case_id"],
|
"batch_id": row["batch_id"],
|
"sub_batch_id": row.get("sub_batch_id", ""),
|
"run_id": row["run_id"],
|
"raw_file_path": row["raw_file_path"],
|
"raw_sha256": row["sha256"],
|
"detected_type": row["detected_type"],
|
"conversion_tool": "pypdf_or_office_zip_xml_timeout",
|
"converted_text_path": "",
|
"page_count": 0,
|
"char_count": 0,
|
"conversion_status": "FAILED",
|
"error_message": "",
|
"review_status": "DRAFT_FOR_REVIEW",
|
}
|
if out_path.exists() and out_path.stat().st_size > 0:
|
text_len = out_path.stat().st_size
|
base.update(
|
{
|
"converted_text_path": out_path.relative_to(root).as_posix(),
|
"char_count": text_len,
|
"conversion_status": "TEXT_CONVERTED",
|
"error_message": "preexisting converted text reused",
|
}
|
)
|
return base
|
|
queue: mp.Queue = mp.Queue()
|
proc = mp.Process(target=worker, args=(str(raw_path), row["detected_type"], queue))
|
proc.start()
|
proc.join(timeout)
|
if proc.is_alive():
|
proc.terminate()
|
proc.join(5)
|
base["error_message"] = f"timeout after {timeout}s"
|
return base
|
if queue.empty():
|
base["error_message"] = "worker returned no result"
|
return base
|
status, text, pages, error = queue.get()
|
if status == "OK" and text:
|
out_path.write_text(text, encoding="utf-8")
|
base.update(
|
{
|
"converted_text_path": out_path.relative_to(root).as_posix(),
|
"page_count": pages,
|
"char_count": len(text),
|
"conversion_status": "TEXT_CONVERTED",
|
}
|
)
|
else:
|
base["error_message"] = error or "empty converted text"
|
return base
|
|
|
def main() -> int:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--project-root", default=".")
|
parser.add_argument("--inventory", required=True)
|
parser.add_argument("--status-out", required=True)
|
parser.add_argument("--converted-dir", required=True)
|
parser.add_argument("--timeout-seconds", type=int, default=60)
|
parser.add_argument("--start-index", type=int, default=1)
|
parser.add_argument("--limit", type=int, default=0)
|
parser.add_argument("--sub-batch-size", type=int, default=30)
|
args = parser.parse_args()
|
|
root = Path(args.project_root).resolve()
|
converted_dir = root / args.converted_dir
|
converted_dir.mkdir(parents=True, exist_ok=True)
|
with (root / args.inventory).open("r", encoding="utf-8-sig", newline="") as f:
|
rows = [r for r in csv.DictReader(f) if r.get("already_in_source_document") == "NO"]
|
for idx, row in enumerate(rows, start=1):
|
row["sub_batch_id"] = f"SB{((idx - 1) // args.sub_batch_size) + 1:03d}"
|
row["run_id"] = f"RUN-ANA-YS-INDUSTRY-001-BATCH-003-CONVERT-001-{row['sub_batch_id']}"
|
rows = rows[args.start_index - 1 :]
|
if args.limit:
|
rows = rows[: args.limit]
|
|
status_path = root / args.status_out
|
status_path.parent.mkdir(parents=True, exist_ok=True)
|
fields = [
|
"conversion_id",
|
"source_doc_id",
|
"case_id",
|
"batch_id",
|
"sub_batch_id",
|
"run_id",
|
"raw_file_path",
|
"raw_sha256",
|
"detected_type",
|
"conversion_tool",
|
"converted_text_path",
|
"page_count",
|
"char_count",
|
"conversion_status",
|
"error_message",
|
"review_status",
|
]
|
write_header = not status_path.exists()
|
with status_path.open("a", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=fields)
|
if write_header:
|
writer.writeheader()
|
for i, row in enumerate(rows, start=args.start_index):
|
result = convert_one(root, row, converted_dir, args.timeout_seconds)
|
writer.writerow(result)
|
f.flush()
|
print(i, result["conversion_status"], row["raw_file_path"])
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
