import os import streamlit as st import json import re import xml.etree.ElementTree as ET from xml.dom import minidom from datetime import datetime, timedelta from decimal import Decimal, ROUND_HALF_UP import fitz import requests from pydantic import BaseModel, Field # ========================================== # 1. API-CLIENT FÜR EXTERNE BACKENDS (vLLM / Ollama) # ========================================== class LLMClient: """Client für Ollama und vLLM (OpenAI-kompatibel) mit JSON-Schema-Unterstützung.""" def __init__(self, backend_type="vllm", model="gemma431b", base_url=None): self.backend_type = backend_type.lower() self.model = model if not base_url: base_url = "http://10.192.0.140:8002" if self.backend_type == "vllm" else "http://localhost:11434" # KUGELSICHERER REINIGER: Filtert versehentliche Markdown-Links [...] vorab heraus 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} ] if self.backend_type == "ollama": return self._ollama_request(messages, temperature, json_schema) else: return self._openai_compatible_request(messages, temperature, json_schema) def _ollama_request(self, messages, temperature, json_schema): url = f"{self.base_url}/api/chat" payload = { "model": self.model, "messages": messages, "stream": False, "options": { "temperature": temperature } } if json_schema: payload["format"] = json_schema try: response = requests.post(url, json=payload, timeout=120) response.raise_for_status() return response.json().get("message", {}).get("content", "") except Exception as e: return f"Ollama Error: {e}" 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}" # ========================================== # 2. METADATEN SCHREIBSCHEMA # ========================================== class InvoiceMetadata(BaseModel): invoice_number: str = Field(description="Die Rechnungsnummer nach 'Nummer:' (z.B. R26-219)") invoice_date: str = Field(description="Das Rechnungsdatum nach 'Datum:' im Format YYYY-MM-DD") delivery_date: str = Field(description="Das Lieferdatum nach 'Lieferdatum:' im Format YYYY-MM-DD") buyer_reference: str = Field(description="Der exakte Wert nach 'Ihr Auftrag:' (z.B. 05.06.2026-Herr.Knospe)") recipient_name: str = Field(description="Vollständige Firma des Kunden/Empfängers (z.B. Stadtwerke Hennigsdorf GmbH)") recipient_street: str = Field(description="Straße und Hausnummer des Kunden") recipient_zip: str = Field(description="Postleitzahl (PLZ) des Kunden") recipient_city: str = Field(description="Stadt des Kunden") recipient_email: str = Field(description="E-Mail-Adresse des Kunden (LEER lassen, wenn nicht explizit im Text!)") vat_rate_percent: float = Field(description="Der primäre Umsatzsteuersatz in Prozent (z.B. 19.0)") # ========================================== # 3. MATHEMATISCHE FUNKTIONEN # ========================================== def kaufmaennisches_runden(betrag, nachkommastellen=2): if not betrag: return 0.0 quantizer = Decimal('1.' + '0' * nachkommastellen) return float(Decimal(f"{betrag:.6f}").quantize(quantizer, rounding=ROUND_HALF_UP)) def clean_german_float(val_str): if not val_str: return 0.0 s = str(val_str).strip().replace("€", "").replace(" ", "").replace("'", "").replace("\n", "").replace("\r", "") if not s: return 0.0 if "." in s and "," in s: s = s.replace(".", "").replace(",", ".") elif "," in s: s = s.replace(",", ".") try: return float(s) except ValueError: return 0.0 def extract_invoice_items_pure_python(text): items = [] lines = [line.strip() for line in text.split("\n") if line.strip()] unit_pattern = re.compile(r"^(\d+(?:[\.,]\d+)?)\s*(Tag\(e\)|Std\.|Stk\.|Stunden|Stück|stk|stck|stück)$", re.IGNORECASE) price_stop_pattern = re.compile(r"^[\d\.]+,[\d]{2}$") for i, line in enumerate(lines): match = unit_pattern.match(line) if match: raw_qty = match.group(1) raw_unit = match.group(2) raw_price = "0,00" for k in range(1, 3): if i + k < len(lines): candidate = lines[i+k].replace("€", "").replace(" ", "").strip() if re.match(r"^[\d.,]+$", candidate): raw_price = candidate break desc_lines = [] for j in range(i - 1, -1, -1): prev_line = lines[j] if unit_pattern.match(prev_line) or price_stop_pattern.match(prev_line): break prev_lower = prev_line.lower() if prev_lower in ["pos.", "pos", "bezeichnung", "beschreibung", "menge", "einheit", "einzelpreis", "gesamtpreis", "betrag €", "betrag", "artikel"]: break if any(prev_lower.startswith(w) for w in ["zwischensumme", "endbetrag", "gesamtbetrag", "umsatzsteuer"]): break if prev_line.isdigit() and len(prev_line) <= 2: continue desc_lines.insert(0, prev_line) description = " ".join(desc_lines).strip() if description: items.append({ "description": description, "quantity": clean_german_float(raw_qty), "unit": raw_unit, "unit_price": clean_german_float(raw_price) }) return items def map_unece_unit_code(unit_str): u = str(unit_str).lower() if "tag" in u or "day" in u: return "DAY" elif "std" in u or "stund" in u or "hour" in u or "hur" in u: return "HUR" return "C62" def parse_to_strict_xml_date(date_str): clean_str = str(date_str).strip() for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d-%m-%Y"): try: return datetime.strptime(clean_str, fmt).strftime("%Y-%m-%d") except ValueError: continue return datetime.today().strftime("%Y-%m-%d") def generate_compliant_xrechnung(data): INVOICE_NS = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2" CBC = "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2" CAC = "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2" # Namespaces explizit registrieren ET.register_namespace('ubl', INVOICE_NS) ET.register_namespace('cbc', CBC) ET.register_namespace('cac', CAC) root = ET.Element(f"{{{INVOICE_NS}}}Invoice") # Offizielle Profil-IDs für XRechnung 3.0 nach Kosit-Vorgabe ET.SubElement(root, f"{{{CBC}}}CustomizationID").text = "urn:cen.eu:en16931:2017#compliant#urn:xeinkauf.de:kosit:xrechnung_3.0" ET.SubElement(root, f"{{{CBC}}}ProfileID").text = "urn:fdc:peppol.eu:2017:poacc:billing:01:1.0" ET.SubElement(root, f"{{{CBC}}}ID").text = str(data["invoice_number"]) ET.SubElement(root, f"{{{CBC}}}IssueDate").text = parse_to_strict_xml_date(data["invoice_date"]) try: base_date = datetime.strptime(parse_to_strict_xml_date(data["invoice_date"]), "%Y-%m-%d") computed_due = (base_date + timedelta(days=14)).strftime("%Y-%m-%d") except Exception: computed_due = (datetime.today() + timedelta(days=14)).strftime("%Y-%m-%d") ET.SubElement(root, f"{{{CBC}}}DueDate").text = computed_due ET.SubElement(root, f"{{{CBC}}}InvoiceTypeCode").text = "380" ET.SubElement(root, f"{{{CBC}}}DocumentCurrencyCode").text = "EUR" ET.SubElement(root, f"{{{CBC}}}BuyerReference").text = str(data["buyer_reference"]) vat_rate = float(data['vat_rate_percent']) vat_category = "S" if vat_rate > 0.0 else "E" # --- SENDER (SCITECH) --- supplier = ET.SubElement(root, f"{{{CAC}}}AccountingSupplierParty") party_supp = ET.SubElement(supplier, f"{{{CAC}}}Party") ET.SubElement(party_supp, f"{{{CBC}}}EndpointID", {"schemeID": "EM"}).text = "office@scitech-gmbh.de" party_name_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyName") ET.SubElement(party_name_supp, f"{{{CBC}}}Name").text = "scitech it solutions GmbH" addr_supp = ET.SubElement(party_supp, f"{{{CAC}}}PostalAddress") ET.SubElement(addr_supp, f"{{{CBC}}}StreetName").text = "Hermann-Schumann-Str. 4" ET.SubElement(addr_supp, f"{{{CBC}}}CityName").text = "Hennigsdorf" ET.SubElement(addr_supp, f"{{{CBC}}}PostalZone").text = "16761" country_supp = ET.SubElement(addr_supp, f"{{{CAC}}}Country") ET.SubElement(country_supp, f"{{{CBC}}}IdentificationCode").text = "DE" tax_scheme_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyTaxScheme") ET.SubElement(tax_scheme_supp, f"{{{CBC}}}CompanyID").text = "DE218436692" scheme_supp = ET.SubElement(tax_scheme_supp, f"{{{CAC}}}TaxScheme") ET.SubElement(scheme_supp, f"{{{CBC}}}ID").text = "VAT" legal_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyLegalEntity") ET.SubElement(legal_supp, f"{{{CBC}}}RegistrationName").text = "scitech it solutions GmbH" # KORREKTUR: Saubere, valide Erzeugung des Contact-Blocks mit cbc-Namespaces contact_supp = ET.SubElement(party_supp, f"{{{CAC}}}Contact") ET.SubElement(contact_supp, f"{{{CBC}}}Name").text = "Geschäftsführung" ET.SubElement(contact_supp, f"{{{CBC}}}Telephone").text = "+49 3302 4940537" ET.SubElement(contact_supp, f"{{{CBC}}}ElectronicMail").text = "office@scitech-gmbh.de" # --- RECIPIENT --- customer = ET.SubElement(root, f"{{{CAC}}}AccountingCustomerParty") party_cust = ET.SubElement(customer, f"{{{CAC}}}Party") ET.SubElement(party_cust, f"{{{CBC}}}EndpointID", {"schemeID": "EM"}).text = str(data["recipient_email"]) if data.get("recipient_email") else "rechnung@kunde.de" party_name_cust = ET.SubElement(party_cust, f"{{{CAC}}}PartyName") ET.SubElement(party_name_cust, f"{{{CBC}}}Name").text = str(data["recipient_name"]) addr_cust = ET.SubElement(party_cust, f"{{{CAC}}}PostalAddress") ET.SubElement(addr_cust, f"{{{CBC}}}StreetName").text = str(data["recipient_street"]) ET.SubElement(addr_cust, f"{{{CBC}}}CityName").text = str(data["recipient_city"]) ET.SubElement(addr_cust, f"{{{CBC}}}PostalZone").text = str(data["recipient_zip"]) country_cust = ET.SubElement(addr_cust, f"{{{CAC}}}Country") ET.SubElement(country_cust, f"{{{CBC}}}IdentificationCode").text = "DE" legal_cust = ET.SubElement(party_cust, f"{{{CAC}}}PartyLegalEntity") ET.SubElement(legal_cust, f"{{{CBC}}}RegistrationName").text = str(data["recipient_name"]) if data.get("delivery_date"): delivery = ET.SubElement(root, f"{{{CAC}}}Delivery") ET.SubElement(delivery, f"{{{CBC}}}ActualDeliveryDate").text = parse_to_strict_xml_date(data["delivery_date"]) payment_means = ET.SubElement(root, f"{{{CAC}}}PaymentMeans") ET.SubElement(payment_means, f"{{{CBC}}}PaymentMeansCode").text = "30" payee_acc = ET.SubElement(payment_means, f"{{{CAC}}}PayeeFinancialAccount") ET.SubElement(payee_acc, f"{{{CBC}}}ID").text = str(data["iban"]).replace(" ", "") tax_total = ET.SubElement(root, f"{{{CAC}}}TaxTotal") ET.SubElement(tax_total, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}" tax_subtotal = ET.SubElement(tax_total, f"{{{CAC}}}TaxSubtotal") ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxableAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}" ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}" tax_cat = ET.SubElement(tax_subtotal, f"{{{CAC}}}TaxCategory") ET.SubElement(tax_cat, f"{{{CBC}}}ID").text = vat_category ET.SubElement(tax_cat, f"{{{CBC}}}Percent").text = f"{vat_rate:.2f}" tax_scheme = ET.SubElement(tax_cat, f"{{{CAC}}}TaxScheme") ET.SubElement(tax_scheme, f"{{{CBC}}}ID").text = "VAT" monetary_total = ET.SubElement(root, f"{{{CAC}}}LegalMonetaryTotal") ET.SubElement(monetary_total, f"{{{CBC}}}LineExtensionAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}" ET.SubElement(monetary_total, f"{{{CBC}}}TaxExclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}" ET.SubElement(monetary_total, f"{{{CBC}}}TaxInclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}" ET.SubElement(monetary_total, f"{{{CBC}}}PayableAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}" for index, item in enumerate(data["items"], start=1): line = ET.SubElement(root, f"{{{CAC}}}InvoiceLine") ET.SubElement(line, f"{{{CBC}}}ID").text = str(index) u_code = map_unece_unit_code(item.get("unit", "C62")) q_val = float(item.get("quantity") or 0.0) qty_formatted = f"{q_val:.4f}".rstrip('0').rstrip('.') if not q_val.is_integer() else f"{q_val:.2f}" if qty_formatted == "" or qty_formatted == "0": qty_formatted = "0.00" ET.SubElement(line, f"{{{CBC}}}InvoicedQuantity", {"unitCode": u_code}).text = qty_formatted ET.SubElement(line, f"{{{CBC}}}LineExtensionAmount", {"currencyID": "EUR"}).text = f"{item['total_amount']:.2f}" item_elem = ET.SubElement(line, f"{{{CAC}}}Item") ET.SubElement(item_elem, f"{{{CBC}}}Name").text = str(item["description"]) # Saubere, UBL-konforme Schachtelung der Steuerkategorie innerhalb des Items item_tax = ET.SubElement(item_elem, f"{{{CAC}}}ClassifiedTaxCategory") ET.SubElement(item_tax, f"{{{CBC}}}ID").text = vat_category ET.SubElement(item_tax, f"{{{CBC}}}Percent").text = f"{vat_rate:.2f}" item_scheme = ET.SubElement(item_tax, f"{{{CAC}}}TaxScheme") ET.SubElement(item_scheme, f"{{{CBC}}}ID").text = "VAT" price_elem = ET.SubElement(line, f"{{{CAC}}}Price") ET.SubElement(price_elem, f"{{{CBC}}}PriceAmount", {"currencyID": "EUR"}).text = f"{item['unit_price']:.2f}" xml_str = ET.tostring(root, encoding="utf-8") return minidom.parseString(xml_str).toprettyxml(indent=" ") # ========================================== # 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') llm_backend = os.environ.get('LLM_BACKEND', 'vllm') st.session_state.llm_instance = LLMClient(backend_type=llm_backend, 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!") 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..."): # STUFE 1: Stammdaten via API mit Schema-Zwang 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" "- 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\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") == "office@scitech-gmbh.de": final_data["recipient_email"] = "" # STUFE 2: Zeilenbasiertes Python-Parsing für die Tabelle final_data["items"] = extract_invoice_items_pure_python(pdf_text) st.session_state.extracted_data = final_data if st.session_state.extracted_data: data = st.session_state.extracted_data inv_num = st.text_input("Rechnungsnummer", value=data.get("invoice_number", "")) inv_date = st.text_input("Rechnungsdatum (YYYY-MM-DD)", value=data.get("invoice_date", "")) inv_delivery = st.text_input("Lieferdatum (YYYY-MM-DD)", value=data.get("delivery_date", "")) buyer_ref = st.text_input("Ihr Auftrag / Referenz", value=data.get("buyer_reference", "")) c_s, c_r = st.columns(2) with c_s: st.markdown("**Absender (Verkäufer - Fest hinterlegt):**") st.text_input("Firma (Sender)", value="scitech it solutions GmbH", disabled=True) st.text_input("E-Mail (Sender)", value="office@scitech-gmbh.de", disabled=True) st.text_input("Telefon (Sender)", value="+49 3302 4940537", disabled=True) st.text_input("Straße (Sender)", value="Hermann-Schumann-Str. 4", disabled=True) st.text_input("PLZ (Sender)", value="16761", disabled=True) st.text_input("Stadt (Sender)", value="Hennigsdorf", disabled=True) with c_r: st.markdown("**Empfänger (Kunde):**") r_name = st.text_input("Firma (Kunde)", value=data.get("recipient_name", "")) r_email = st.text_input("E-Mail (Kunde)", value=data.get("recipient_email", "")) r_street = st.text_input("Straße (Kunde)", value=data.get("recipient_street", "")) r_zip = st.text_input("PLZ (Kunde)", value=data.get("recipient_zip", "")) r_city = st.text_input("Stadt (Kunde)", value=data.get("recipient_city", "")) st.markdown("**3. Positionen:**") edited_items = st.data_editor(data.get("items", []), num_rows="dynamic", width="stretch") vat_rate = st.number_input("Steuersatz (%)", value=float(data.get("vat_rate_percent") or 19.0)) for item in edited_items: raw_total = float(item.get("quantity") or 0.0) * float(item.get("unit_price") or 0.0) item["total_amount"] = kaufmaennisches_runden(raw_total) computed_net_amount = sum(item["total_amount"] for item in edited_items) raw_vat_amount = computed_net_amount * (vat_rate / 100.0) computed_vat_amount = kaufmaennisches_runden(raw_vat_amount) computed_gross_amount = kaufmaennisches_runden(computed_net_amount + computed_vat_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}") st.markdown("---") if st.button("🔒 XRechnung exportieren", type="primary", width="stretch"): final_invoice_dict = { "invoice_number": inv_num, "invoice_date": inv_date, "delivery_date": inv_delivery, "buyer_reference": buyer_ref, "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": "DE06120700240456207000" } st.download_button(label="📥 Download XML", data=generate_compliant_xrechnung(final_invoice_dict), file_name=f"XRechnung_{inv_num}.xml", mime="text/xml")