import argparse
|
import csv
|
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):
|
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
writer = csv.DictWriter(handle, fieldnames=fieldnames)
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def norm_path(value):
|
return (value or "").replace("\\", "/").strip()
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--source-document", required=True)
|
parser.add_argument("--conversion-status", required=True)
|
parser.add_argument("--updated-manifest", required=True)
|
args = parser.parse_args()
|
|
source_path = Path(args.source_document)
|
status_path = Path(args.conversion_status)
|
updated_manifest_path = Path(args.updated_manifest)
|
|
source_rows = read_csv(source_path)
|
status_rows = read_csv(status_path)
|
status_by_path = {norm_path(row.get("raw_file_path")): row for row in status_rows}
|
|
update_rows = []
|
for row in source_rows:
|
raw_path = norm_path(row.get("raw_file_path"))
|
status = status_by_path.get(raw_path)
|
if not status:
|
continue
|
before_processing = row.get("processing_status", "")
|
before_text = row.get("converted_text_path", "")
|
conversion_status = status.get("conversion_status", "")
|
if conversion_status == "TEXT_CONVERTED":
|
row["converted_text_path"] = status.get("converted_text_path", "")
|
row["processing_status"] = "TEXT_CONVERTED"
|
else:
|
row["processing_status"] = "CONVERSION_GAP"
|
row["run_id"] = status.get("run_id", row.get("run_id", ""))
|
update_rows.append(
|
{
|
"doc_id": row.get("doc_id", ""),
|
"raw_file_path": row.get("raw_file_path", ""),
|
"file_sha256": row.get("file_sha256", ""),
|
"conversion_status": conversion_status,
|
"processing_status_before": before_processing,
|
"processing_status_after": row.get("processing_status", ""),
|
"converted_text_path_before": before_text,
|
"converted_text_path_after": row.get("converted_text_path", ""),
|
"run_id": row.get("run_id", ""),
|
}
|
)
|
|
fieldnames = list(source_rows[0].keys()) if source_rows else []
|
write_csv(source_path, fieldnames, source_rows)
|
|
update_fields = [
|
"doc_id",
|
"raw_file_path",
|
"file_sha256",
|
"conversion_status",
|
"processing_status_before",
|
"processing_status_after",
|
"converted_text_path_before",
|
"converted_text_path_after",
|
"run_id",
|
]
|
updated_manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
write_csv(updated_manifest_path, update_fields, update_rows)
|
print(f"updated={len(update_rows)}")
|
print(f"manifest={updated_manifest_path.as_posix()}")
|
|
|
if __name__ == "__main__":
|
main()
|
