E-Rechnung/main.py

494 lines
24 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import os
import io
import streamlit as st
import json
import re
import fitz
import requests
from facturx import generate_from_binary as facturx_generate_from_binary
from invoice_logic import (
InvoiceMetadata,
VERKAEUFER,
VERKAEUFER_IBAN,
PAYMENT_MEANS_CODES,
berechne_faelligkeit,
kaufmaennisches_runden,
clean_german_float,
extract_invoice_items_pure_python,
map_unece_unit_code,
parse_to_strict_xml_date,
format_cii_date,
generate_compliant_xrechnung,
generate_compliant_zugferd_cii,
add_pdfa_output_intent,
fix_xrechnung_xmp_conformance,
compute_invoice_totals,
extract_service_period,
detect_contract_type,
extract_hardware_serial_numbers,
)
# ==========================================
# 0. SITZUNGS-STATUS INITIALISIEREN (ZÄHLER GEGEN WIDGET-CACHING)
# ==========================================
if "invoice_counter" not in st.session_state:
st.session_state.invoice_counter = 0
# ==========================================
# 1. API-CLIENT FÜR DAS vLLM-BACKEND
# ==========================================
class LLMClient:
def __init__(self, model="gemma431b", base_url="http://10.192.0.140:8002"):
self.model = model
base_url = base_url.strip()
if base_url.startswith("[") and ")" in base_url:
match = re.search(r"\((https?://.*?)\)", base_url)
if match:
base_url = match.group(1)
self.base_url = base_url.rstrip("/")
def chat(self, prompt, system_prompt="You are a helpful assistant.", temperature=0.0, json_schema=None):
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
]
return self._openai_compatible_request(messages, temperature, json_schema)
def _openai_compatible_request(self, messages, temperature, json_schema):
url = f"{self.base_url}/v1/chat/completions"
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature
}
if json_schema:
payload["guided_json"] = json.dumps(json_schema) if isinstance(json_schema, dict) else json_schema
try:
response = requests.post(url, json=payload, timeout=120)
response.raise_for_status()
data = response.json()
return data.get("choices", [{}])[0].get("message", {}).get("content", "")
except Exception as e:
return f"vLLM/OpenAI Error: {e}"
# ==========================================
# 4. API CLIENT INITIALISIERUNG
# ==========================================
def get_cached_llm():
if "llm_instance" not in st.session_state:
llm_url = os.environ.get('VLLM_URL', 'http://10.192.0.140:8002')
llm_model = os.environ.get('LLM_MODEL', 'gemma431b')
st.session_state.llm_instance = LLMClient(model=llm_model, base_url=llm_url)
return st.session_state.llm_instance
# ==========================================
# 5. STREAMLIT FRONTEND
# ==========================================
st.set_page_config(page_title="PDF zu E-Rechnung", layout="wide")
st.title("📄 PDF zu E-Rechnung Konverter")
llm = get_cached_llm()
col1, col2 = st.columns(2)
with col1:
st.subheader("1. Dokument hochladen")
uploaded_file = st.file_uploader("Wähle eine Rechnungs-PDF", type=["pdf"])
pdf_text = ""
if uploaded_file is not None:
doc = fitz.open(stream=uploaded_file.read(), filetype="pdf")
for page in doc:
pdf_text += page.get_text("text") + "\n"
st.success("PDF eingelesen!")
uploaded_file.seek(0)
st.pdf(uploaded_file)
with st.expander("Gefundener Text"): st.text(pdf_text)
with col2:
st.subheader("2. Strukturierte Daten validieren")
if "extracted_data" not in st.session_state: st.session_state.extracted_data = None
if uploaded_file is not None and pdf_text.strip() != "":
if st.button("Daten extrahieren", width="stretch"):
with st.spinner("Extrahiere Daten über vLLM-API..."):
schema_dict = InvoiceMetadata.model_json_schema()
sys_prompt = (
"Du bist ein präziser OCR-Parser für deutsche B2B-Rechnungen. "
"Extrahiere die Metadaten fehlerfrei als valides JSON-Objekt. "
"Das JSON MUSS zwingend exakt folgende Schlüssel (Keys) enthalten:\n"
"- invoice_number\n"
"- invoice_date (im Format YYYY-MM-DD)\n"
"- delivery_date (im Format YYYY-MM-DD)\n"
"- buyer_reference\n"
"- customer_number\n"
"- recipient_name\n"
"- recipient_street\n"
"- recipient_zip\n"
"- recipient_city\n"
"- recipient_email\n"
"- vat_rate_percent (als Zahl, z.B. 19.0)\n"
"- delivery_note_id\n"
"- order_id\n"
"- contract_id\n"
"- service_start_date\n"
"- service_end_date\n"
"- payment_term_days\n"
"- invoice_note\n\n"
"WICHTIGE REGEL: Analysiere den Text direkt VOR der ersten Tabellenposition '1'. "
"Sortiere alle dort vorlaufenden Zeilen vollständig in die dafür vorgesehenen ID-, Datums- und Freitext-Felder ein. "
"Schreibe diese vorlaufenden Metadaten NIEMALS in die Beschreibung der ersten Position hinein!\n\n"
"Antworte ausschließlich mit dem reinen JSON-Objekt. Schreibe keinen anderen Text."
)
prompt_meta = f"Rechnungstext:\n{pdf_text}"
res_meta_text = llm.chat(
prompt=prompt_meta,
system_prompt=sys_prompt,
temperature=0.0,
json_schema=schema_dict
)
clean_json_text = res_meta_text.strip()
backtick_trigger = chr(96) * 3
if clean_json_text.startswith(backtick_trigger):
lines = clean_json_text.splitlines()
if lines[0].startswith(backtick_trigger):
lines = lines[1:]
if lines and lines[-1].startswith(backtick_trigger):
lines = lines[:-1]
clean_json_text = "\n".join(lines).strip()
try:
final_data = json.loads(clean_json_text)
except Exception as e:
st.error(f"Fehler beim Parsen der API-Antwort: {e}")
st.code(res_meta_text)
st.stop()
if final_data.get("recipient_email") == VERKAEUFER["email"]:
final_data["recipient_email"] = ""
final_data["items"] = extract_invoice_items_pure_python(pdf_text)
# IT-Systemhaus-Regex-Fallbacks: nur ergaenzen, wenn das LLM die entsprechenden
# Felder nicht bereits geliefert hat.
if not final_data.get("service_start_date") and not final_data.get("service_end_date"):
fallback_start, fallback_end = extract_service_period(pdf_text)
if fallback_start and fallback_end:
final_data["service_start_date"] = fallback_start
final_data["service_end_date"] = fallback_end
final_data["contract_type"] = detect_contract_type(pdf_text)
final_data["hardware_serials"] = extract_hardware_serial_numbers(pdf_text)
st.session_state.invoice_counter += 1
st.session_state.extracted_data = final_data
if st.session_state.extracted_data:
data = st.session_state.extracted_data
st.caption("Mit :red[*] markierte Felder sind EN16931-Pflichtfelder und müssen für einen gültigen Export befüllt sein.")
if data.get("contract_type"):
st.caption(f"🏷️ Erkannter Vertragstyp: **{data['contract_type']}**")
if data.get("hardware_serials"):
st.caption(f"🔧 Erkannte Seriennummern: {', '.join(data['hardware_serials'])}")
# --- BT-3 RECHNUNGSTYP DROPDOWN-MENÜ ---
inv_type_ui = st.selectbox(
"Rechnungstyp (BT-3)",
options=["380 (Handelsrechnung)", "381 (Gutschrift)", "384 (Korrekturrechnung)", "326 (Teilrechnung)"],
index=0,
help="Bestimmt den steuerrechtlichen Belegtyp im XML-Header."
)
selected_type_code = inv_type_ui.split(" ")[0]
inv_num = st.text_input("Rechnungsnummer (BT-1) :red[*]", value=data.get("invoice_number", ""))
inv_date = st.text_input("Rechnungsdatum (BT-2, YYYY-MM-DD) :red[*]", value=data.get("invoice_date", ""))
inv_delivery = st.text_input("Lieferdatum (BT-72, YYYY-MM-DD)", value=data.get("delivery_date", ""))
cust_num = st.text_input(
"Kundennummer / Leitweg-ID (BT-10)",
value=data.get("customer_number", ""),
help="Eure Kundennummer beim Empfänger (Standardfall für B2B) oder eine Leitweg-ID, falls "
"der Empfänger eine Behörde ist."
)
buyer_ref = st.text_input(
"Ihr Auftrag / Bestellnummer (BT-13)",
value=data.get("buyer_reference", "")
)
# UI-Eingabefelder für die Daten vor Position 1 (BT-22 etc.)
st.markdown("---")
st.markdown("**Globale Referenzen & Vorlauftexte:**")
c_ref1, c_ref2, c_ref3 = st.columns(3)
with c_ref1:
dn_id = st.text_input("Lieferscheinnummer (BT-16)", value=data.get("delivery_note_id", ""))
with c_ref2:
o_id = st.text_input("Auftragsbestätigung / Verkaufsauftrag (BT-14)", value=data.get("order_id", ""))
with c_ref3:
c_id = st.text_input("Vertrag / Vergabeangebot (BT-12)", value=data.get("contract_id", ""))
c_p1, c_p2 = st.columns(2)
with c_p1:
s_start = st.text_input("Leistungszeitraum Start (BT-73, YYYY-MM-DD)", value=data.get("service_start_date", ""))
with c_p2:
s_end = st.text_input("Leistungszeitraum Ende (BT-74, YYYY-MM-DD)", value=data.get("service_end_date", ""))
pay_term = st.text_input(
"Zahlungsziel in Tagen (BT-9) :red[*]",
value=str(data.get("payment_term_days", "") or "")
)
i_note = st.text_area("Globale Rechnungserläuterung / Freitext (BT-22)", value=data.get("invoice_note", ""), height=150)
st.markdown("---")
# Adressdaten
c_s, c_r = st.columns(2)
with c_s:
st.markdown("**Absender (Verkäufer):**")
st.text_input("Firma (BT-27)", value=VERKAEUFER["name"], disabled=True)
st.text_input("E-Mail (BT-34)", value=VERKAEUFER["email"], disabled=True)
st.text_input("Telefon (BT-42)", value=VERKAEUFER["phone"], disabled=True)
st.text_input("Straße (BT-35)", value=VERKAEUFER["street"], disabled=True)
st.text_input("PLZ (BT-38)", value=VERKAEUFER["zip"], disabled=True)
st.text_input("Stadt (BT-37)", value=VERKAEUFER["city"], disabled=True)
st.text_input("Umsatzsteuer-ID (BT-31)", value=VERKAEUFER["vat_id"], disabled=True)
st.text_input("Registernummer (BT-30)", value=VERKAEUFER["register_id"], disabled=True)
st.text_input("IBAN (BT-84)", value=VERKAEUFER_IBAN, disabled=True)
with c_r:
st.markdown("**Empfänger (Kunde):**")
r_name = st.text_input("Firma (BT-44) :red[*]", value=data.get("recipient_name", ""))
r_email = st.text_input("E-Mail (BT-49) :red[*]", value=data.get("recipient_email", ""))
r_street = st.text_input("Straße (BT-50)", value=data.get("recipient_street", ""))
r_zip = st.text_input("PLZ (BT-53)", value=data.get("recipient_zip", ""))
r_city = st.text_input("Stadt (BT-52) :red[*]", value=data.get("recipient_city", ""))
st.markdown("---")
payment_means_codes_list = list(PAYMENT_MEANS_CODES.keys())
payment_means_labels = [f"{code} - {PAYMENT_MEANS_CODES[code]}" for code in payment_means_codes_list]
selected_payment_means_label = st.selectbox(
"Zahlungsmittel (BT-81)",
options=payment_means_labels,
index=payment_means_codes_list.index("58"),
)
selected_payment_means_code = selected_payment_means_label.split(" - ")[0]
st.markdown("### 3. Positionen")
if st.button("➕ Neue Position hinzufügen"):
data["items"].append({
"name": "Neue Dienstleistung", "description": "", "quantity": 1.0, "unit": "Stk.",
"unit_price": 0.0, "vat_rate_percent": float(data.get("vat_rate_percent") or 19.0),
})
st.rerun()
edited_items = []
inv_id = st.session_state.invoice_counter
for idx, item in enumerate(data.get("items", [])):
with st.expander(f"📋 Position {idx+1}: {item.get('name', 'Dienstleistung')}", expanded=True):
c_del, _ = st.columns([1, 4])
with c_del:
if st.button("🗑️ Position löschen", key=f"del_btn_{inv_id}_{idx}"):
data["items"].pop(idx)
st.rerun()
c_name, c_desc = st.columns([1, 1])
with c_name:
item_name = st.text_input("Artikelname (BT-153) :red[*]", value=item.get("name", ""), key=f"item_name_{inv_id}_{idx}")
with c_desc:
desc_text = item.get("description", "")
estimated_lines = desc_text.count("\n") + (len(desc_text) // 45) + 1
calc_height = max(68, min(400, estimated_lines * 24))
item_description = st.text_area("Zusatzbeschreibung (BT-154)", value=desc_text, key=f"item_desc_{inv_id}_{idx}", height=calc_height)
c_qty, c_unit, c_price, c_vat, c_total = st.columns(5)
with c_qty:
item_quantity = st.number_input("Menge (BT-129) :red[*]", value=float(item.get("quantity") or 0.0), format="%.3f", step=0.1, key=f"item_qty_{inv_id}_{idx}")
with c_unit:
item_unit = st.text_input("Einheit (BT-130) :red[*]", value=item.get("unit", "Stk."), key=f"item_unit_{inv_id}_{idx}")
with c_price:
item_price = st.number_input("Einzelpreis (BT-146, €) :red[*]", value=float(item.get("unit_price") or 0.0), format="%.2f", key=f"item_price_{inv_id}_{idx}")
with c_vat:
item_vat_rate = st.number_input(
"Steuersatz (BT-152, %) :red[*]",
value=float(item.get("vat_rate_percent") or data.get("vat_rate_percent") or 19.0),
key=f"item_vat_{inv_id}_{idx}",
)
with c_total:
raw_item_total = item_quantity * item_price
item_total_amount = kaufmaennisches_runden(raw_item_total)
st.markdown("**Gesamtbetrag (€):**")
st.markdown(f"**{item_total_amount:.2f} €**")
edited_items.append({
"name": item_name,
"description": item_description,
"quantity": item_quantity,
"unit": item_unit,
"unit_price": item_price,
"total_amount": item_total_amount,
"vat_rate_percent": item_vat_rate,
})
st.markdown("---")
# Der frühere globale Steuersatz dient jetzt nur noch als Vorbelegung für neu
# hinzugefügte Positionen (siehe "Neue Position hinzufügen" oben) - abgerechnet wird
# pro Position mit dem jeweils eingetragenen Steuersatz (Multi-Tax-Support).
vat_rate = st.number_input("Standard-Steuersatz für neue Positionen (%)", value=float(data.get("vat_rate_percent") or 19.0))
c_disc1, c_disc2, c_disc3, c_disc4 = st.columns(4)
with c_disc1:
discount_percent = st.number_input("Rabatt (BT-21, %)", value=0.0, min_value=0.0, max_value=100.0, step=0.5)
with c_disc2:
discount_reason = st.text_input("Rabattgrund", value="")
with c_disc3:
skonto_percent = st.number_input("Skonto (BT-20, %)", value=0.0, min_value=0.0, max_value=100.0, step=0.5)
with c_disc4:
skonto_days = st.text_input("Skonto-Frist (Tage)", value="")
totals = compute_invoice_totals(edited_items, discount_percent=discount_percent)
computed_net_amount = totals["tax_basis_total"]
computed_vat_amount = totals["total_vat_amount"]
computed_gross_amount = totals["total_gross_amount"]
st.markdown("**Berechnete Summen:**")
c_net_ui, c_vat_ui, c_gross_ui = st.columns(3)
with c_net_ui: st.metric("Netto-Gesamt (€)", f"{computed_net_amount:.2f}")
with c_vat_ui: st.metric("MwSt-Gesamt (€)", f"{computed_vat_amount:.2f}")
with c_gross_ui: st.metric("Brutto-Zahlbetrag (€)", f"{computed_gross_amount:.2f}")
# --- EXPORT-SEKTION DROPDOWN-AUSWAHL ---
st.markdown("---")
export_format = st.selectbox(
"💾 Export-Format wählen",
options=["XRechnung v3.0.2 (Reines UBL XML)", "ZUGFeRD v2.4 (Hybrid-PDF mit CII XML)"],
index=1,
help="XRechnung generiert ein reines, valides UBL-XML. ZUGFeRD bettet ein schema-konformes CII-XML (factur-x.xml) direkt in deine PDF ein."
)
final_invoice_dict = {
"invoice_type_code": selected_type_code,
"invoice_number": inv_num, "invoice_date": inv_date, "delivery_date": inv_delivery,
"buyer_reference": buyer_ref, "customer_number": cust_num, "recipient_name": r_name, "recipient_email": r_email,
"recipient_street": r_street, "recipient_zip": r_zip, "recipient_city": r_city,
"items": edited_items, "total_net_amount": computed_net_amount, "total_vat_amount": computed_vat_amount,
"vat_rate_percent": vat_rate, "total_gross_amount": computed_gross_amount, "iban": VERKAEUFER_IBAN,
"payment_means_code": selected_payment_means_code,
"tax_buckets": totals["tax_buckets"],
"line_net_total": totals["line_net_total"],
"allowance_amount": totals["allowance_amount"],
"allowance_reason": discount_reason,
"skonto_percent": skonto_percent,
"skonto_days": skonto_days,
"delivery_note_id": dn_id,
"order_id": o_id,
"contract_id": c_id,
"service_start_date": s_start,
"service_end_date": s_end,
"payment_term_days": pay_term,
"invoice_note": i_note
}
# --- PFLICHTFELD-VALIDIERUNG vor dem Export ---
# Sammelt blockierende Probleme (Export nicht möglich), damit fehlerhafte/fehlende
# Pflichtangaben sichtbar werden, statt still mit Platzhaltern oder falschen Werten
# in die Rechnung zu wandern.
blocker = []
if not str(inv_num).strip():
blocker.append("Rechnungsnummer (BT-1) fehlt.")
if parse_to_strict_xml_date(inv_date) == "":
blocker.append("Rechnungsdatum (BT-2) fehlt oder ist unlesbar (erwartet: YYYY-MM-DD). "
"Bitte korrigieren - es wird KEIN Datum automatisch eingesetzt.")
if not str(r_email).strip():
blocker.append("E-Mail des Kunden (BT-49) fehlt. Diese elektronische Adresse ist für "
"XRechnung/ZUGFeRD ein Pflichtfeld (PEPPOL-EN16931-R010) und wird NICHT "
"mit einem Platzhalter befüllt. Bitte eintragen.")
if inv_delivery and parse_to_strict_xml_date(inv_delivery) == "":
blocker.append("Lieferdatum (BT-72) ist unlesbar (erwartet: YYYY-MM-DD). Bitte korrigieren oder leeren.")
if not edited_items:
blocker.append("Die Rechnung enthält keine Position (mindestens eine ist erforderlich).")
_pt = str(pay_term).strip()
if _pt == "":
blocker.append("Zahlungsziel (BT-9/BT-20) fehlt. Es wird KEIN Standardwert (z.B. 14 Tage) "
"angenommen - bitte das tatsächliche Zahlungsziel der Rechnung in Tagen eintragen.")
elif not _pt.isdigit():
blocker.append("Zahlungsziel (BT-9/BT-20) muss eine Zahl in Tagen sein (z.B. 14 oder 30).")
if blocker:
st.warning("⚠️ Export blockiert - folgende Pflichtangaben fehlen oder sind fehlerhaft:\n\n"
+ "\n".join(f"- {b}" for b in blocker))
export_disabled = len(blocker) > 0
if "UBL XML" in export_format:
if st.button("🔒 XRechnung exportieren", type="primary", width="stretch", disabled=export_disabled):
xml_data = generate_compliant_xrechnung(final_invoice_dict)
st.download_button(label="📥 Download XRechnung XML", data=xml_data, file_name=f"XRechnung_{inv_num}.xml", mime="text/xml")
else:
if st.button("🔒 ZUGFeRD PDF exportieren", type="primary", width="stretch", disabled=export_disabled):
try:
cii_xml_string = generate_compliant_zugferd_cii(final_invoice_dict)
uploaded_file.seek(0)
original_pdf_bytes = uploaded_file.read()
# 1) factur-x übernimmt: XSD-Validierung der CII-XML, korrektes Einbetten
# (Filespec mit AFRelationship=/Data, /F + /UF, Subtype text/xml) und die
# komplette PDF/A-3-konforme XMP-Metadatenstruktur (pdfaid:part/conformance,
# fx:-Schema-Erweiterung). Bricht die XML-Validierung fehl, wird hier ein
# klarer Fehler geworfen statt einer stillen, ungültigen PDF.
zugferd_pdf_bytes = facturx_generate_from_binary(
original_pdf_bytes,
cii_xml_string.encode("utf-8"),
flavor="factur-x",
level="en16931",
check_xsd=True,
check_schematron=False, # benötigt optional 'saxonche' für Schematron-Regeln
afrelationship="data",
)
# 2) Zusätzliches OutputIntent (sRGB-ICC-Profil) ergänzen. Das ist für echte
# PDF/A-3-Konformität zwingend, wird von factur-x aber nicht selbst gesetzt.
zugferd_pdf_bytes = add_pdfa_output_intent(zugferd_pdf_bytes)
zugferd_pdf_bytes = fix_xrechnung_xmp_conformance(zugferd_pdf_bytes)
st.download_button(
label="📥 Download ZUGFeRD Hybrid-PDF",
data=zugferd_pdf_bytes,
file_name=f"ZUGFeRD_{inv_num}.pdf",
mime="application/pdf"
)
st.info(
"ℹ️ Hinweis: Die eingebettete CII-XML wurde gegen das offizielle EN16931-XSD-Schema "
"validiert. Für eine vollständige PDF/A-3-Zertifizierung (z. B. Schriftarten-Einbettung "
"der Ursprungs-PDF) empfiehlt sich zusätzlich eine Prüfung mit veraPDF."
)
except Exception as e:
st.error(f"Fehler bei der ZUGFeRD-Generierung: {e}")
# ==========================================
# FOOTER
# ==========================================
st.markdown("---")
st.caption(
"📝 **Hinweise:**\n\n"
"**Human-in-the-Loop:** "
"Das Tool dient als Automations-Assistent. Vor dem finalen Export ist eine visuelle Endkontrolle der extrahierten Daten "
"(insbesondere Rechnungsnummer und Empfängeradresse) durch den Sachbearbeiter zwingend erforderlich, um KI-Fehlinterpretationen auszuschließen.\n\n"
"**GoBD-Konformität:** "
"Das Tool generiert ausschließlich das strukturierte XML- bzw. hybride PDF-Format. Für die Einhaltung der GoBD-Richtlinien "
"(u. a. die revisionssichere Archivierung der Original-PDF sowie der generierten E-Rechnung über den gesetzlichen Zeitraum von 10 Jahren) "
"ist der Betreiber der Software selbst verantwortlich. \n\n"
)