240 lines
9.9 KiB
Python
240 lines
9.9 KiB
Python
"""Stapelbetrieb: mehrere Rechnungen in einem Durchgang statt einzeln.
|
|
|
|
Zwei Quellen:
|
|
* Dateien hochladen - funktioniert immer, egal ob die PDFs lokal oder auf einer
|
|
Netzwerkfreigabe liegen; der Browser kommt an beides. Ergebnis kommt als ZIP zurueck.
|
|
* Ordner auf dem Server - nur wenn ein Austauschverzeichnis gemountet ist. Dann bleiben die
|
|
Dateien serverseitig und die Ergebnisse landen direkt im Ausgabeordner.
|
|
|
|
Die Logik liegt in batch_folder (ohne Streamlit, daher testbar); hier steht nur die
|
|
Darstellung. build_zugferd_pdf wird hereingereicht, weil sie in main.py gecacht definiert ist -
|
|
ein Import waere zirkulaer.
|
|
"""
|
|
import json
|
|
|
|
import streamlit as st
|
|
|
|
import batch_folder
|
|
from invoice_logic import MAX_ATTACHMENT_PDFS, append_pdf_pages, build_export_basename
|
|
|
|
_STATE_KEY = "batch_invoices"
|
|
|
|
|
|
def _reset_batch():
|
|
st.session_state.pop(_STATE_KEY, None)
|
|
|
|
|
|
def _make_entry(name, pdf_bytes):
|
|
"""Analysiert eine PDF und verpackt sie mit ihrer Bewertung. Ein Lesefehler stoppt den
|
|
Stapel nicht - die Datei wird als zu pruefen mit Klartext gemeldet."""
|
|
try:
|
|
data = batch_folder.analyse_pdf(pdf_bytes)
|
|
problems = batch_folder.evaluate_invoice(data, data["computed_net_amount"])
|
|
except Exception as e:
|
|
data, problems = None, [f"PDF konnte nicht gelesen werden: {e}"]
|
|
return {"name": name, "bytes": pdf_bytes, "data": data,
|
|
"problems": problems, "attachments": [], "source": None}
|
|
|
|
|
|
def _render_overview(entries, build_zugferd_pdf, write_to_folder=None):
|
|
"""Zuordnung, Ampel und Sammel-Export. write_to_folder ist der Ausgabeordner im
|
|
Server-Betrieb; ist er None, wird ein ZIP zum Download erzeugt."""
|
|
all_names = [e["name"] for e in entries]
|
|
|
|
if len(entries) > 1:
|
|
with st.expander("Leistungsnachweise zuordnen (optional)"):
|
|
st.caption(
|
|
f"Gehoert eine der hochgeladenen PDFs als Nachweis zu einer Rechnung, ordnen Sie "
|
|
f"sie hier zu (max. {MAX_ATTACHMENT_PDFS} je Rechnung). Zugeordnete Dateien "
|
|
"zaehlen dann nicht mehr als eigene Rechnung, sondern werden hinten angehaengt."
|
|
)
|
|
for entry in entries:
|
|
others = [n for n in all_names if n != entry["name"]]
|
|
chosen = st.multiselect(
|
|
f"Nachweise für {entry['name']}", options=others,
|
|
key=f"batch_att_{entry['name']}",
|
|
)
|
|
entry["attachments"] = chosen[:MAX_ATTACHMENT_PDFS]
|
|
if len(chosen) > MAX_ATTACHMENT_PDFS:
|
|
st.error(f"Maximal {MAX_ATTACHMENT_PDFS} Nachweise je Rechnung - "
|
|
"die überzähligen werden ignoriert.")
|
|
|
|
assigned = {n for e in entries for n in e["attachments"]}
|
|
invoices = [e for e in entries if e["name"] not in assigned]
|
|
green = [e for e in invoices if not e["problems"]]
|
|
red = [e for e in invoices if e["problems"]]
|
|
|
|
st.markdown("---")
|
|
c1, c2, c3 = st.columns(3)
|
|
c1.metric("Rechnungen", len(invoices))
|
|
c2.metric("Freigabefähig", len(green))
|
|
c3.metric("Zu prüfen", len(red))
|
|
|
|
for entry in invoices:
|
|
data = entry["data"] or {}
|
|
nr = data.get("invoice_number") or "?"
|
|
empf = data.get("recipient_name") or "?"
|
|
betrag = f"{data.get('computed_net_amount', 0.0):.2f} €" if entry["data"] else "-"
|
|
anh = f" · +{len(entry['attachments'])} Nachweis(e)" if entry["attachments"] else ""
|
|
titel = f"{nr} — {empf} — {betrag}{anh}"
|
|
|
|
if entry["problems"]:
|
|
with st.expander(f"⚠️ {titel}"):
|
|
for p in entry["problems"]:
|
|
st.warning(p)
|
|
st.caption(
|
|
"Diese Rechnung wird beim Sammel-Export übersprungen. Korrigieren Sie sie in "
|
|
"der Betriebsart *Einzelrechnung* - dort stehen alle Felder bereit."
|
|
)
|
|
else:
|
|
st.markdown(f"✅ **{nr}** — {empf} — {betrag}{anh}")
|
|
|
|
st.markdown("---")
|
|
if not green:
|
|
st.info("Keine freigabefähige Rechnung - bitte zuerst die markierten Fälle klären.")
|
|
return
|
|
|
|
if not st.button(f"🔒 {len(green)} Rechnung(en) konvertieren", type="primary", width="stretch"):
|
|
return
|
|
|
|
by_name = {e["name"]: e for e in entries}
|
|
progress = st.progress(0.0)
|
|
results, failed = [], []
|
|
|
|
for idx, entry in enumerate(green, start=1):
|
|
try:
|
|
pdf_bytes = entry["bytes"]
|
|
if entry["attachments"]:
|
|
extras = [by_name[n]["bytes"] for n in entry["attachments"]]
|
|
pdf_bytes = append_pdf_pages(pdf_bytes, *extras)
|
|
|
|
zugferd = build_zugferd_pdf(
|
|
json.dumps(entry["data"], default=str, sort_keys=True), pdf_bytes
|
|
)
|
|
basename = build_export_basename(
|
|
entry["data"].get("invoice_number", ""),
|
|
entry["data"].get("recipient_name", ""),
|
|
entry["data"].get("bearbeiter", ""),
|
|
)
|
|
results.append((f"{basename}.pdf", zugferd, entry))
|
|
except Exception as e:
|
|
failed.append((entry["name"], str(e)))
|
|
progress.progress(idx / len(green))
|
|
|
|
if results and write_to_folder is not None:
|
|
# Server-Betrieb: direkt in den Ausgabeordner schreiben, Quellen wegraeumen.
|
|
written = []
|
|
for name, data, entry in results:
|
|
try:
|
|
out = batch_folder.write_output(write_to_folder, name, data)
|
|
batch_folder.move_to_processed(entry["source"])
|
|
for n in entry["attachments"]:
|
|
batch_folder.move_to_processed(by_name[n]["source"])
|
|
written.append(out.name)
|
|
except Exception as e:
|
|
failed.append((entry["name"], str(e)))
|
|
if written:
|
|
st.success(
|
|
f"✅ {len(written)} Rechnung(en) nach `{write_to_folder}` geschrieben. Die "
|
|
f"Quelldateien liegen jetzt im Unterordner `{batch_folder.PROCESSED_SUBDIR}`."
|
|
)
|
|
st.caption(" · ".join(written))
|
|
_reset_batch()
|
|
elif results:
|
|
# Upload-Betrieb: alles in ein ZIP, das der Nutzer herunterlaedt.
|
|
zip_bytes = batch_folder.build_zip([(name, data) for name, data, _ in results])
|
|
st.success(f"✅ {len(results)} Rechnung(en) konvertiert.")
|
|
st.download_button(
|
|
f"⬇️ {len(results)} E-Rechnung(en) als ZIP herunterladen",
|
|
data=zip_bytes,
|
|
file_name="e-rechnungen.zip",
|
|
mime="application/zip",
|
|
type="primary",
|
|
width="stretch",
|
|
)
|
|
with st.expander("Enthaltene Dateien"):
|
|
for name, data, _ in results:
|
|
st.markdown(f"- `{name}` ({len(data) / 1024:.0f} KB)")
|
|
|
|
if failed:
|
|
# Ein Fehlschlag stoppt den Stapel nicht - die uebrigen laufen durch.
|
|
st.error(
|
|
f"❌ {len(failed)} Rechnung(en) konnten nicht konvertiert werden:\n\n"
|
|
+ "\n".join(f"- **{n}**: {msg}" for n, msg in failed)
|
|
)
|
|
if red:
|
|
st.info(f"{len(red)} Rechnung(en) wurden übersprungen, weil sie zu prüfen sind.")
|
|
|
|
|
|
def render_batch_mode(build_zugferd_pdf):
|
|
st.subheader("Stapelverarbeitung")
|
|
st.caption(
|
|
"Mehrere Rechnungen in einem Durchgang: einlesen, per Ampel prüfen, gesammelt "
|
|
"konvertieren. Geprüft wird, ob alle Pflichtfelder belegt sind und ob die Summe der "
|
|
"erkannten Positionen zur ausgewiesenen Bemessungsgrundlage passt."
|
|
)
|
|
|
|
subfolders = batch_folder.list_subfolders()
|
|
quelle = "Dateien hochladen"
|
|
if subfolders:
|
|
quelle = st.radio(
|
|
"Quelle", ["Dateien hochladen", "Ordner auf dem Server"], horizontal=True,
|
|
on_change=_reset_batch,
|
|
)
|
|
|
|
if quelle == "Dateien hochladen":
|
|
uploads = st.file_uploader(
|
|
"Rechnungs-PDFs auswählen (mehrere möglich)", type=["pdf"],
|
|
accept_multiple_files=True, key="batch_uploads",
|
|
)
|
|
st.caption(
|
|
"Markieren Sie im Explorer alle gewünschten PDFs und ziehen Sie sie hierher - "
|
|
"lokale Ordner und Netzlaufwerke funktionieren gleichermaßen."
|
|
)
|
|
if not uploads:
|
|
return
|
|
|
|
signature = [(f.name, f.size) for f in uploads]
|
|
if st.session_state.get("_batch_signature") != signature:
|
|
st.session_state._batch_signature = signature
|
|
with st.spinner(f"Analysiere {len(uploads)} Rechnung(en)..."):
|
|
entries = []
|
|
for f in uploads:
|
|
f.seek(0)
|
|
entries.append(_make_entry(f.name, f.read()))
|
|
st.session_state[_STATE_KEY] = entries
|
|
|
|
entries = st.session_state.get(_STATE_KEY) or []
|
|
if entries:
|
|
_render_overview(entries, build_zugferd_pdf)
|
|
return
|
|
|
|
# --- Server-Ordner ---------------------------------------------------------------
|
|
c_in, c_out = st.columns(2)
|
|
with c_in:
|
|
input_dir = st.selectbox("Eingangsordner", options=subfolders,
|
|
key="batch_input_dir", on_change=_reset_batch)
|
|
with c_out:
|
|
output_dir = st.selectbox("Ausgabeordner", options=subfolders, key="batch_output_dir")
|
|
|
|
if input_dir == output_dir:
|
|
st.warning("⚠️ Ein- und Ausgabeordner sind identisch. Bitte unterschiedliche Ordner wählen.")
|
|
return
|
|
|
|
if st.button("📂 Ordner einlesen", width="stretch"):
|
|
with st.spinner("Lese Rechnungen ein..."):
|
|
entries = []
|
|
for path in batch_folder.list_input_pdfs(input_dir):
|
|
entry = _make_entry(path.name, path.read_bytes())
|
|
entry["source"] = path
|
|
entries.append(entry)
|
|
st.session_state[_STATE_KEY] = entries
|
|
|
|
entries = st.session_state.get(_STATE_KEY)
|
|
if entries is None:
|
|
return
|
|
if not entries:
|
|
st.info(f"Im Ordner `{input_dir}` liegen keine PDFs.")
|
|
return
|
|
_render_overview(entries, build_zugferd_pdf, write_to_folder=output_dir)
|