263 lines
10 KiB
Python
263 lines
10 KiB
Python
"""Ordner-Anbindung fuer den Stapelbetrieb.
|
|
|
|
Bewusst OHNE Streamlit-Abhaengigkeit: die Pfad-, Auswahl- und Bewertungslogik ist damit
|
|
vollstaendig testbar, auch dort wo die App selbst nicht startbar ist.
|
|
|
|
Die App laeuft im Container und kann Windows-UNC-Pfade (\\\\FS01\\...) nicht verwenden. Die
|
|
Netzwerkfreigabe wird stattdessen EINMALIG als Volume unter BATCH_ROOT_DIR gemountet; alles
|
|
darunter ist zur Laufzeit frei waehlbar. Jede Pfadoperation prueft, dass ihr Ziel innerhalb
|
|
dieser Wurzel liegt - ein im UI gewaehlter Ordnername darf nicht aus dem Mount ausbrechen.
|
|
"""
|
|
import io
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import fitz
|
|
|
|
from invoice_logic import (
|
|
VERKAEUFER,
|
|
VERKAEUFER_IBAN,
|
|
collect_blockers,
|
|
compute_invoice_totals,
|
|
detect_invoice_type_code,
|
|
extract_document_references,
|
|
extract_header_metadata_pure_python,
|
|
extract_invoice_items,
|
|
extract_labeled_header_value,
|
|
extract_pre_position_freetext,
|
|
extract_referenznummer,
|
|
extract_service_period,
|
|
extract_stated_tax_basis,
|
|
kaufmaennisches_runden,
|
|
resolve_buyer_endpoint,
|
|
sums_match,
|
|
)
|
|
|
|
# Wurzel des gemounteten Austauschverzeichnisses. Alles darunter ist im UI waehlbar.
|
|
BATCH_ROOT_DIR = os.environ.get("BATCH_ROOT_DIR", "/data")
|
|
|
|
# Unterordner, in den verarbeitete Quell-PDFs wandern. Sie werden NICHT geloescht - die
|
|
# Originale sind GoBD-relevant und muessen aufbewahrt werden.
|
|
PROCESSED_SUBDIR = "verarbeitet"
|
|
|
|
|
|
def _root(root=None) -> Path:
|
|
return Path(root if root is not None else BATCH_ROOT_DIR)
|
|
|
|
|
|
def resolve_within_root(relative_path, root=None) -> Path:
|
|
"""Loest einen (relativen) Pfad gegen die Wurzel auf und stellt sicher, dass das Ergebnis
|
|
INNERHALB der Wurzel liegt. Wirft ValueError bei einem Ausbruchsversuch ('../' oder ein
|
|
absoluter Pfad ausserhalb) - der Ordnername kommt aus dem UI und ist damit Eingabe, der
|
|
nicht blind vertraut werden darf."""
|
|
base = _root(root).resolve()
|
|
target = (base / Path(relative_path)).resolve()
|
|
if target != base and base not in target.parents:
|
|
raise ValueError(
|
|
f"Pfad '{relative_path}' liegt ausserhalb des freigegebenen Verzeichnisses '{base}'."
|
|
)
|
|
return target
|
|
|
|
|
|
def list_subfolders(root=None) -> list:
|
|
"""Namen der direkten Unterordner der Wurzel, alphabetisch - die Auswahlliste im UI.
|
|
Existiert die Wurzel nicht (Mount fehlt), kommt eine leere Liste zurueck."""
|
|
base = _root(root)
|
|
if not base.is_dir():
|
|
return []
|
|
return sorted(p.name for p in base.iterdir() if p.is_dir())
|
|
|
|
|
|
def list_input_pdfs(relative_dir, root=None) -> list:
|
|
"""Alle PDFs eines Eingangsordners, alphabetisch. Der Unterordner mit den bereits
|
|
verarbeiteten Dateien wird ausgelassen, damit ein Stapel nicht doppelt laeuft."""
|
|
directory = resolve_within_root(relative_dir, root)
|
|
if not directory.is_dir():
|
|
return []
|
|
return sorted(
|
|
(p for p in directory.iterdir() if p.is_file() and p.suffix.lower() == ".pdf"),
|
|
key=lambda p: p.name,
|
|
)
|
|
|
|
|
|
def _free_path(directory: Path, filename: str) -> Path:
|
|
"""Ein noch freier Dateiname in directory: haengt _2, _3 ... an, statt eine vorhandene
|
|
Datei zu ueberschreiben. Ein frueherer Export darf nicht still verlorengehen."""
|
|
candidate = directory / filename
|
|
if not candidate.exists():
|
|
return candidate
|
|
stem, suffix = Path(filename).stem, Path(filename).suffix
|
|
counter = 2
|
|
while (directory / f"{stem}_{counter}{suffix}").exists():
|
|
counter += 1
|
|
return directory / f"{stem}_{counter}{suffix}"
|
|
|
|
|
|
def write_output(relative_dir, filename, data: bytes, root=None) -> Path:
|
|
"""Schreibt eine fertige Export-Datei in den Ausgabeordner und gibt den tatsaechlich
|
|
verwendeten Pfad zurueck (kann bei Namenskollision abweichen). Der Ordner wird bei Bedarf
|
|
angelegt."""
|
|
directory = resolve_within_root(relative_dir, root)
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
target = _free_path(directory, filename)
|
|
target.write_bytes(data)
|
|
return target
|
|
|
|
|
|
def move_to_processed(pdf_path, root=None) -> Path:
|
|
"""Verschiebt eine verarbeitete Quell-PDF in den Unterordner 'verarbeitet' und gibt das
|
|
Ziel zurueck. Bei Namenskollision wird angehaengt statt ueberschrieben."""
|
|
source = Path(pdf_path).resolve()
|
|
base = _root(root).resolve()
|
|
if base not in source.parents:
|
|
raise ValueError(
|
|
f"Datei '{source}' liegt ausserhalb des freigegebenen Verzeichnisses '{base}'."
|
|
)
|
|
processed_dir = source.parent / PROCESSED_SUBDIR
|
|
processed_dir.mkdir(parents=True, exist_ok=True)
|
|
target = _free_path(processed_dir, source.name)
|
|
shutil.move(str(source), str(target))
|
|
return target
|
|
|
|
|
|
def build_zip(files) -> bytes:
|
|
"""Packt fertige Exporte in ein ZIP - der Rueckweg fuer den Upload-Stapel, bei dem es
|
|
keinen Ausgabeordner auf dem Server gibt.
|
|
|
|
Nimmt ein Dict {name: bytes} oder eine Liste [(name, bytes), ...]. Doppelte Namen werden
|
|
mit _2, _3 ... entzerrt, statt sich gegenseitig zu ueberschreiben - zwei Rechnungen koennen
|
|
denselben Exportnamen ergeben."""
|
|
import zipfile
|
|
|
|
items = list(files.items()) if isinstance(files, dict) else list(files)
|
|
buf = io.BytesIO()
|
|
used = set()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as archive:
|
|
for name, data in items:
|
|
final = name
|
|
if final in used:
|
|
stem, suffix = Path(name).stem, Path(name).suffix
|
|
counter = 2
|
|
while f"{stem}_{counter}{suffix}" in used:
|
|
counter += 1
|
|
final = f"{stem}_{counter}{suffix}"
|
|
used.add(final)
|
|
archive.writestr(final, data)
|
|
return buf.getvalue()
|
|
|
|
|
|
def pdf_text_of(pdf_bytes: bytes) -> str:
|
|
"""Extrahiert den Text einer PDF - identisch zum Einzelmodus (main.py)."""
|
|
text = ""
|
|
doc = fitz.open(stream=pdf_bytes, filetype="pdf")
|
|
try:
|
|
for page in doc:
|
|
text += page.get_text("text") + "\n"
|
|
finally:
|
|
doc.close()
|
|
return text
|
|
|
|
|
|
def analyse_pdf(pdf_bytes: bytes) -> dict:
|
|
"""Fuehrt fuer eine Rechnungs-PDF dieselbe Extraktion aus wie der Einzelmodus und liefert
|
|
ein exportfertiges Rechnungs-Dict samt berechneter Summen.
|
|
|
|
Spiegelt bewusst die Reihenfolge aus main.py: Kopfdaten, Positionen ueber den DISPATCHER
|
|
(nicht die Pure-Python-Variante - sonst faellt der Buendel-Parser aus), Bemessungs-
|
|
grundlage, Leistungszeitraum-Fallback, Freitext, Belegreferenzen."""
|
|
text = pdf_text_of(pdf_bytes)
|
|
data = extract_header_metadata_pure_python(text)
|
|
|
|
if data.get("recipient_email") == VERKAEUFER["email"]:
|
|
data["recipient_email"] = ""
|
|
|
|
items = extract_invoice_items(text)
|
|
for item in items:
|
|
item["total_amount"] = kaufmaennisches_runden(
|
|
float(item.get("quantity") or 0.0) * float(item.get("unit_price") or 0.0)
|
|
)
|
|
item["vat_rate_percent"] = float(
|
|
item.get("vat_rate_percent") or data.get("vat_rate_percent") or 19.0
|
|
)
|
|
data["items"] = items
|
|
data["stated_tax_basis"] = extract_stated_tax_basis(text)
|
|
|
|
if not data.get("service_start_date") and not data.get("service_end_date"):
|
|
start, end = extract_service_period(text)
|
|
if start and end:
|
|
data["service_start_date"], data["service_end_date"] = start, end
|
|
|
|
data["invoice_type_code"] = detect_invoice_type_code(text)
|
|
|
|
freetext = extract_pre_position_freetext(text)
|
|
if freetext is not None:
|
|
data["invoice_note"] = freetext
|
|
|
|
for field, value in extract_document_references(text).items():
|
|
data[field] = value
|
|
|
|
ihr_auftrag = extract_labeled_header_value(text, "Ihr Auftrag:")
|
|
if ihr_auftrag:
|
|
data["buyer_reference"] = ihr_auftrag
|
|
|
|
data["bearbeiter"] = extract_labeled_header_value(text, "Bearbeiter:")
|
|
|
|
refnr = extract_referenznummer(text)
|
|
if refnr and refnr not in str(data.get("invoice_note", "")):
|
|
existing = str(data.get("invoice_note", "")).strip()
|
|
line = f"Referenznummer: {refnr}"
|
|
data["invoice_note"] = f"{existing}\n{line}".strip() if existing else line
|
|
|
|
# Summen und die uebrigen Exportfelder ergaenzen, damit das Dict ohne UI exportfertig ist.
|
|
totals = compute_invoice_totals(items, discount_percent=0.0)
|
|
endpoint, scheme = resolve_buyer_endpoint(
|
|
data.get("recipient_email", ""), "EM", data.get("leitweg_id", "")
|
|
)
|
|
data.update({
|
|
"total_net_amount": totals["tax_basis_total"],
|
|
"total_vat_amount": totals["total_vat_amount"],
|
|
"total_gross_amount": totals["total_gross_amount"],
|
|
"tax_buckets": totals["tax_buckets"],
|
|
"line_net_total": totals["line_net_total"],
|
|
"allowance_amount": totals["allowance_amount"],
|
|
"allowance_reason": "",
|
|
"iban": VERKAEUFER_IBAN,
|
|
"payment_means_code": "58",
|
|
"skonto_percent": 0.0,
|
|
"skonto_days": 0,
|
|
"vorkasse": False,
|
|
"seller_contact_name": data.get("bearbeiter", ""),
|
|
"seller_endpoint": VERKAEUFER["email"],
|
|
"seller_contact_email": VERKAEUFER["email"],
|
|
"recipient_email": endpoint,
|
|
"recipient_endpoint_scheme": scheme,
|
|
"payment_reference": "",
|
|
})
|
|
data["computed_net_amount"] = totals["tax_basis_total"]
|
|
return data
|
|
|
|
|
|
def evaluate_invoice(data, computed_net_amount, is_xrechnung=False) -> list:
|
|
"""Ampel-Bewertung einer Rechnung: leere Liste = gruen (freigabefaehig), sonst die
|
|
Klartext-Probleme.
|
|
|
|
Enthaelt dieselbe Pflichtfeld-Pruefung wie der Einzelmodus (collect_blockers) plus den
|
|
Summen-Abgleich gegen die auf der Rechnung ausgewiesene Bemessungsgrundlage. Letzterer ist
|
|
der zuverlaessigste Hinweis auf eine unvollstaendige Positionserkennung und traegt im
|
|
Stapelbetrieb die Freigabe ohne visuelle Einzelkontrolle."""
|
|
problems = collect_blockers(data, is_xrechnung)
|
|
|
|
stated = data.get("stated_tax_basis")
|
|
if stated is None:
|
|
problems.append(
|
|
"Keine ausgewiesene Bemessungsgrundlage gefunden - die Positionssumme ist nicht "
|
|
"gegenpruefbar. Bitte einzeln kontrollieren."
|
|
)
|
|
elif not sums_match(computed_net_amount, stated):
|
|
problems.append(
|
|
f"Summen-Abgleich fehlgeschlagen: erkannte Positionen {computed_net_amount:.2f} € "
|
|
f"weichen von der ausgewiesenen Bemessungsgrundlage {float(stated):.2f} € ab."
|
|
)
|
|
return problems
|