Reisacher-Forderung/app/docuware_export.py

"""Export a bounded selection into the offline DocuWare manifest format."""

import argparse
import hashlib
import json
import logging
from pathlib import Path

from .docuware_search import LiveAdapter
from .logging_setup import run_cli

logger = logging.getLogger(__name__)


def export(adapter, tenant, searches, output):
    output = Path(output)
    output.mkdir(parents=True, exist_ok=True)
    if (output / "manifest.json").exists():
        raise ValueError("Ziel enthält bereits einen Export; bitte neuen Ausgabeordner verwenden")
    documents, errors = {}, []
    for search in searches:
        try:
            for doc in adapter.search(tenant, **search):
                documents[(doc["archive"], doc["id"])] = {**doc, "client_db": tenant}
        except Exception as exc:
            errors.append(dict(stage="search", error=type(exc).__name__))
    for (archive, id_), doc in documents.items():
        # Hash both identifiers so archive labels cannot become paths.
        stem = hashlib.sha256(f"{archive}\0{id_}".encode()).hexdigest()
        try:
            data, media, name = adapter.download(tenant, doc)
            if media.split(";", 1)[0].lower() != "application/pdf":
                raise ValueError("PDF erwartet")
            filename = stem + ".pdf"
            (output / filename).write_bytes(data)
            doc.update(file=filename, name=name)
        except Exception as exc:
            errors.append(dict(stage="download", archive=archive, id=id_, error=type(exc).__name__))
    manifest = dict(version=1, documents=list(documents.values()), errors=errors)
    (output / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    return manifest


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--client-db", required=True, choices=["1", "2", "3", "4"])
    parser.add_argument("--document-no", action="append", default=[])
    parser.add_argument("--customer-no")
    parser.add_argument("--vehicle-no")
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    searches = [dict(document_no=n) for n in args.document_no]
    if args.customer_no:
        searches.append(dict(customer_no=args.customer_no))
    if args.vehicle_no:
        searches.append(dict(vehicle_no=args.vehicle_no))
    if not searches:
        parser.error("Mindestens ein Beleg-, Kunden- oder Fahrzeugfilter ist erforderlich")
    result = export(LiveAdapter(), args.client_db, searches, args.output)
    logger.info(
        "Export abgeschlossen: %s Dokumente, %s Fehler. Bericht: manifest.json",
        len(result["documents"]),
        len(result["errors"]),
    )
    return 1 if result["errors"] else 0


if __name__ == "__main__":

    def run():
        raise SystemExit(main())

    run_cli(run)