"""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)