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 # ========================================== # 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}" # ========================================== # 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)") delivery_note_id: str = Field(default="", description="Gefundene Lieferscheinnummer (z.B. LS26-035) VOR Position 1. Sonst leer lassen.") order_id: str = Field(default="", description="Gefundene Nummer der Auftragsbestätigung oder Bestellung (z.B. AB26-080) VOR Position 1. Sonst leer lassen.") contract_id: str = Field(default="", description="Gefundene Nummer von Vergabeangebot, Vertrag oder Ausschreibung (z.B. AN26-130) VOR Position 1. Sonst leer lassen.") service_start_date: str = Field(default="", description="Startdatum eines explizit genannten Einsatz- oder Leistungszeitraums VOR Position 1 im Format YYYY-MM-DD. Sonst leer.") service_end_date: str = Field(default="", description="Enddatum eines explizit genannten Einsatz- oder Leistungszeitraums VOR Position 1 im Format YYYY-MM-DD. Sonst leer.") invoice_note: str = Field(default="", description="Sämtlicher unstrukturierter Fließtext, Beschreibungen, Abnahmekriterien (Acceptance Criteria) oder Projektklauseln, die sich VOR Position 1 befinden. Vollständig mit Zeilenumbrüchen sichern.") # ========================================== # 3. MATHEMATISCHE & PARSING-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: break desc_lines.insert(0, prev_line) if desc_lines: item_name = desc_lines[0].strip() item_description = " ".join(desc_lines[1:]).strip() if len(desc_lines) > 1 else "" items.append({ "name": item_name, "description": item_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): if not date_str or str(date_str).strip() == "": return "" 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 format_cii_date(date_str): clean = parse_to_strict_xml_date(date_str) return clean.replace("-", "") # ========================================== # GENERATOR A: XRECHNUNG (UBL FORMAT) # ========================================== 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" ET.register_namespace('ubl', INVOICE_NS) ET.register_namespace('cbc', CBC) ET.register_namespace('cac', CAC) root = ET.Element(f"{{{INVOICE_NS}}}Invoice") 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 = str(data.get("invoice_type_code", "380")) if data.get("invoice_note") and str(data["invoice_note"]).strip() != "": ET.SubElement(root, f"{{{CBC}}}Note").text = str(data["invoice_note"]).strip() ET.SubElement(root, f"{{{CBC}}}DocumentCurrencyCode").text = "EUR" ET.SubElement(root, f"{{{CBC}}}BuyerReference").text = str(data["buyer_reference"]) if data.get("service_start_date") and data.get("service_end_date"): period_elem = ET.SubElement(root, f"{{{CAC}}}InvoicePeriod") ET.SubElement(period_elem, f"{{{CBC}}}StartDate").text = parse_to_strict_xml_date(data["service_start_date"]) ET.SubElement(period_elem, f"{{{CBC}}}EndDate").text = parse_to_strict_xml_date(data["service_end_date"]) if data.get("order_id") and str(data["order_id"]).strip() != "": order_ref = ET.SubElement(root, f"{{{CAC}}}OrderReference") ET.SubElement(order_ref, f"{{{CBC}}}ID").text = str(data["order_id"]).strip() if data.get("delivery_note_id") and str(data["delivery_note_id"]).strip() != "": despatch_ref = ET.SubElement(root, f"{{{CAC}}}DespatchDocumentReference") ET.SubElement(despatch_ref, f"{{{CBC}}}ID").text = str(data["delivery_note_id"]).strip() if data.get("contract_id") and str(data["contract_id"]).strip() != "": contract_ref = ET.SubElement(root, f"{{{CAC}}}ContractDocumentReference") ET.SubElement(contract_ref, f"{{{CBC}}}ID").text = str(data["contract_id"]).strip() 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" 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" # --- Empfänger --- 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") if item.get("description") and str(item["description"]).strip() != "": ET.SubElement(item_elem, f"{{{CBC}}}Description").text = str(item["description"]).strip() ET.SubElement(item_elem, f"{{{CBC}}}Name").text = str(item.get("name", "Dienstleistung")) 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=" ") # ========================================== # GENERATOR B: ZUGFeRD 2.4 (CII FORMAT UN/CEFACT) # ========================================== def generate_compliant_zugferd_cii(data): RSM = "urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100" RAM = "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100" UDT = "urn:un:unece:uncefact:data:standard:UnqualifiedDataType:100" ET.register_namespace('rsm', RSM) ET.register_namespace('ram', RAM) ET.register_namespace('udt', UDT) root = ET.Element(f"{{{RSM}}}CrossIndustryInvoice") ex_doc_ctx = ET.SubElement(root, f"{{{RAM}}}ExchangedDocumentContext") guideline = ET.SubElement(ex_doc_ctx, f"{{{RAM}}}GuidelineSpecifiedDocumentContextParameter") ET.SubElement(guideline, f"{{{RAM}}}ID").text = "urn:cen.eu:en16931:2017#compliant#urn:xeinkauf.de:kosit:xrechnung_3.0" ex_doc = ET.SubElement(root, f"{{{RSM}}}ExchangedDocument") ET.SubElement(ex_doc, f"{{{RAM}}}ID").text = str(data["invoice_number"]) ET.SubElement(ex_doc, f"{{{RAM}}}TypeCode").text = str(data.get("invoice_type_code", "380")) issue_date = ET.SubElement(ex_doc, f"{{{RAM}}}IssueDateTime") ET.SubElement(issue_date, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["invoice_date"]) if data.get("invoice_note") and str(data["invoice_note"]).strip() != "": note = ET.SubElement(ex_doc, f"{{{RAM}}}IncludedNote") ET.SubElement(note, f"{{{RAM}}}Content").text = str(data["invoice_note"]).strip() trade_trans = ET.SubElement(root, f"{{{RSM}}}SupplyChainTradeTransaction") vat_rate = float(data['vat_rate_percent']) vat_category = "S" if vat_rate > 0.0 else "E" for index, item in enumerate(data["items"], start=1): line_item = ET.SubElement(trade_trans, f"{{{RAM}}}IncludedSupplyChainTradeLineItem") doc_line = ET.SubElement(line_item, f"{{{RAM}}}AssociatedDocumentLineDocument") ET.SubElement(doc_line, f"{{{RAM}}}LineID").text = str(index) trade_prod = ET.SubElement(line_item, f"{{{RAM}}}SpecifiedTradeProduct") ET.SubElement(trade_prod, f"{{{RAM}}}Name").text = str(item.get("name", "Dienstleistung")) if item.get("description") and str(item["description"]).strip() != "": ET.SubElement(trade_prod, f"{{{RAM}}}Description").text = str(item["description"]).strip() line_agree = ET.SubElement(line_item, f"{{{RAM}}}SpecifiedLineTradeAgreement") net_price = ET.SubElement(line_agree, f"{{{RAM}}}GrossPriceProductTradePrice") ET.SubElement(net_price, f"{{{RAM}}}ChargeAmount").text = f"{item['unit_price']:.2f}" line_deliv = ET.SubElement(line_item, f"{{{RAM}}}SpecifiedLineTradeDelivery") u_code = map_unece_unit_code(item.get("unit", "Stk.")) ET.SubElement(line_deliv, f"{{{RAM}}}BilledQuantity", {"unitCode": u_code}).text = f"{item['quantity']:.4f}" line_settle = ET.SubElement(line_item, f"{{{RAM}}}SpecifiedLineTradeSettlement") trade_tax = ET.SubElement(line_settle, f"{{{RAM}}}ApplicableTradeTax") ET.SubElement(trade_tax, f"{{{RAM}}}TypeCode").text = "VAT" ET.SubElement(trade_tax, f"{{{RAM}}}CategoryCode").text = vat_category ET.SubElement(trade_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{vat_rate:.2f}" mon_sum = ET.SubElement(line_settle, f"{{{RAM}}}SpecifiedTradeSettlementLineMonetarySummation") ET.SubElement(mon_sum, f"{{{RAM}}}LineTotalAmount").text = f"{item['total_amount']:.2f}" header_agree = ET.SubElement(trade_trans, f"{{{RAM}}}ApplicableHeaderTradeAgreement") ET.SubElement(header_agree, f"{{{RAM}}}BuyerReference").text = str(data["buyer_reference"]) supplier_party = ET.SubElement(header_agree, f"{{{RAM}}}SellerTradeParty") ET.SubElement(supplier_party, f"{{{RAM}}}Name").text = "scitech it solutions GmbH" supp_legal = ET.SubElement(supplier_party, f"{{{RAM}}}SpecifiedLegalOrganization") ET.SubElement(supp_legal, f"{{{RAM}}}ID").text = "DE218436692" supp_addr = ET.SubElement(supplier_party, f"{{{RAM}}}PostalTradeAddress") ET.SubElement(supp_addr, f"{{{RAM}}}PostcodeCode").text = "16761" ET.SubElement(supp_addr, f"{{{RAM}}}LineOne").text = "Hermann-Schumann-Str. 4" ET.SubElement(supp_addr, f"{{{RAM}}}CityName").text = "Hennigsdorf" ET.SubElement(supp_addr, f"{{{RAM}}}CountryID").text = "DE" supp_tax = ET.SubElement(supplier_party, f"{{{RAM}}}SpecifiedTaxRegistration") ET.SubElement(supp_tax, f"{{{RAM}}}ID", {"schemeID": "VA"}).text = "DE218436692" customer_party = ET.SubElement(header_agree, f"{{{RAM}}}BuyerTradeParty") ET.SubElement(customer_party, f"{{{RAM}}}Name").text = str(data["recipient_name"]) cust_addr = ET.SubElement(customer_party, f"{{{RAM}}}PostalTradeAddress") ET.SubElement(cust_addr, f"{{{RAM}}}PostcodeCode").text = str(data["recipient_zip"]) ET.SubElement(cust_addr, f"{{{RAM}}}LineOne").text = str(data["recipient_street"]) ET.SubElement(cust_addr, f"{{{RAM}}}CityName").text = str(data["recipient_city"]) ET.SubElement(cust_addr, f"{{{RAM}}}CountryID").text = "DE" if data.get("order_id") and str(data["order_id"]).strip() != "": order_ref = ET.SubElement(header_agree, f"{{{RAM}}}BuyerOrderReferencedDocument") ET.SubElement(order_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["order_id"]).strip() if data.get("contract_id") and str(data["contract_id"]).strip() != "": contract_ref = ET.SubElement(header_agree, f"{{{RAM}}}ContractReferencedDocument") ET.SubElement(contract_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["contract_id"]).strip() header_deliv = ET.SubElement(trade_trans, f"{{{RAM}}}ApplicableHeaderTradeDelivery") if data.get("delivery_note_id") and str(data["delivery_note_id"]).strip() != "": dn_ref = ET.SubElement(header_deliv, f"{{{RAM}}}DeliveryNoteReferencedDocument") ET.SubElement(dn_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["delivery_note_id"]).strip() if data.get("delivery_date"): act_deliv = ET.SubElement(header_deliv, f"{{{RAM}}}ActualDeliverySupplyChainEvent") occ_date = ET.SubElement(act_deliv, f"{{{RAM}}}OccurrenceDateTime") ET.SubElement(occ_date, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["delivery_date"]) header_settle = ET.SubElement(trade_trans, f"{{{RAM}}}ApplicableHeaderTradeSettlement") ET.SubElement(header_settle, f"{{{RAM}}}InvoiceCurrencyCode").text = "EUR" pay_means = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradePaymentMeans") ET.SubElement(pay_means, f"{{{RAM}}}TypeCode").text = "30" payee_acc = ET.SubElement(pay_means, f"{{{RAM}}}PayeePartyCreditorFinancialAccount") ET.SubElement(payee_acc, f"{{{RAM}}}IBANID").text = str(data["iban"]).replace(" ", "") header_tax = ET.SubElement(header_settle, f"{{{RAM}}}ApplicableTradeTax") ET.SubElement(header_tax, f"{{{RAM}}}CalculatedAmount").text = f"{data['total_vat_amount']:.2f}" ET.SubElement(header_tax, f"{{{RAM}}}TypeCode").text = "VAT" ET.SubElement(header_tax, f"{{{RAM}}}BasisAmount").text = f"{data['total_net_amount']:.2f}" ET.SubElement(header_tax, f"{{{RAM}}}CategoryCode").text = vat_category ET.SubElement(header_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{vat_rate:.2f}" if data.get("service_start_date") and data.get("service_end_date"): period = ET.SubElement(header_settle, f"{{{RAM}}}BillingSpecifiedPeriod") start_pt = ET.SubElement(period, f"{{{RAM}}}StartDateTime") ET.SubElement(start_pt, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["service_start_date"]) end_pt = ET.SubElement(period, f"{{{RAM}}}EndDateTime") ET.SubElement(end_pt, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = format_cii_date(data["service_end_date"]) header_sum = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradeSettlementHeaderMonetarySummation") ET.SubElement(header_sum, f"{{{RAM}}}LineTotalAmount").text = f"{data['total_net_amount']:.2f}" ET.SubElement(header_sum, f"{{{RAM}}}TaxBasisTotalAmount").text = f"{data['total_net_amount']:.2f}" ET.SubElement(header_sum, f"{{{RAM}}}TaxTotalAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}" header_sum.find(f"{{{RAM}}}TaxTotalAmount").set("currencyID", "EUR") ET.SubElement(header_sum, f"{{{RAM}}}GrandTotalAmount").text = f"{data['total_gross_amount']:.2f}" ET.SubElement(header_sum, f"{{{RAM}}}DuePayableAmount").text = f"{data['total_gross_amount']:.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') 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!") 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" "- 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" "- 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") == "office@scitech-gmbh.de": final_data["recipient_email"] = "" final_data["items"] = extract_invoice_items_pure_python(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 # --- BT-3 RECHNUNGSTYP DROPDOWN-MENÜ --- inv_type_ui = st.selectbox( "Rechnungstyp (BT-3)", options=["380 (Handelsrechnung)", "381 (Gutschrift)", "384 (Korrekturrechnung)"], index=0, help="Bestimmt den steuerrechtlichen Belegtyp im XML-Header." ) selected_type_code = inv_type_ui.split(" ")[0] 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", "")) # 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 / Bestellung (BT-13)", 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 (YYYY-MM-DD)", value=data.get("service_start_date", "")) with c_p2: s_end = st.text_input("Leistungszeitraum Ende (YYYY-MM-DD)", value=data.get("service_end_date", "")) 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 (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") if st.button("➕ Neue Position hinzufügen"): data["items"].append({"name": "Neue Dienstleistung", "description": "", "quantity": 1.0, "unit": "Stk.", "unit_price": 0.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)", 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_total = st.columns(4) with c_qty: item_quantity = st.number_input("Menge", 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", value=item.get("unit", "Stk."), key=f"item_unit_{inv_id}_{idx}") with c_price: item_price = st.number_input("Einzelpreis (€)", value=float(item.get("unit_price") or 0.0), format="%.2f", key=f"item_price_{inv_id}_{idx}") with c_total: raw_item_total = item_quantity * item_price item_total_amount = kaufmaennisches_runden(raw_item_total) st.markdown("**Gesamt betrag (€):**") 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 }) st.markdown("---") vat_rate = st.number_input("Steuersatz (%)", value=float(data.get("vat_rate_percent") or 19.0)) 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}") # --- 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)"], 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, "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", "delivery_note_id": dn_id, "order_id": o_id, "contract_id": c_id, "service_start_date": s_start, "service_end_date": s_end, "invoice_note": i_note } if "UBL XML" in export_format: if st.button("🔒 XRechnung exportieren", type="primary", width="stretch"): 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"): try: cii_xml_string = generate_compliant_zugferd_cii(final_invoice_dict) uploaded_file.seek(0) original_pdf_bytes = uploaded_file.read() doc = fitz.open(stream=original_pdf_bytes, filetype="pdf") # 1. XML-Datei einbetten nach ZUGFeRD Standard doc.embfile_add( "factur-x.xml", cii_xml_string.encode("utf-8"), filename="factur-x.xml", desc="Factur-X/ZUGFeRD E-Invoice" ) # 2. XMP-Metadaten inkl. PDF/A-3-Identifikation & ZUGFeRD-Schema aufbauen xmp_xml = """ 3 B Factur-X PDFA Extension Schema fx urn:factur-x:pdfa:CrossIndustryDocument:invoice:1p0# DocumentType Text external The type of the document, usually INVOICE DocumentFileName Text external The name of the embedded file Version Text external The version of the document type schema ConformanceLevel Text external The conformance level of the document INVOICE factur-x.xml 1.0 XRECHNUNG """ doc.set_metadata(doc.metadata, xmp=xmp_xml) # 3. Low-Level XREF-Injektion für strikte PDF/A-3 & ZUGFeRD Konformität filespec_xref = None for xref in range(1, doc.xref_length()): obj_str = doc.xref_object(xref) if "/Type /Filespec" in obj_str and "factur-x.xml" in obj_str: filespec_xref = xref break if filespec_xref: # 3.1 Erzwinge das /AFRelationship Attribut am eingebetteten FileSpec doc.xref_set_key(filespec_xref, "AFRelationship", "/Alternative") # 3.2 Ermittle den Catalog-Knoten und verknüpfe das XML im globalen /AF Array cat_xref = doc.pdf_catalog() doc.xref_set_key(cat_xref, "AF", f"[{filespec_xref} 0 R]") # 3.3 Standard-sRGB OutputIntent erstellen & injizieren (Farbprofil-Marker für PDF/A) oi_xref = doc.xref_create() oi_dict = ( "<< /Type /OutputIntent " "/S /GTS_PDFA1 " "/OutputConditionIdentifier (sRGB) " "/RegistryName (http://www.color.org) " "/Info (sRGB IEC61966-2.1) >>" ) doc.xref_update(oi_xref, oi_dict) doc.xref_set_key(cat_xref, "OutputIntents", f"[{oi_xref} 0 R]") # PDF finalisieren und umwandeln zugferd_pdf_bytes = doc.tobytes() doc.close() st.download_button( label="📥 Download ZUGFeRD Hybrid-PDF", data=zugferd_pdf_bytes, file_name=f"ZUGFeRD_{inv_num}.pdf", mime="application/pdf" ) 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 XML-Export ist eine visuelle Endkontrolle der extrahierten Daten " "(insbesondere Rechnungsnummer und Empfängeradresse) durch den Sachbearbeiter zwingend erforderlich.\n\n" "**GoBD-Konformität:** " "Das Tool generiert ausschließlich das strukturierte XML-Format. Für die Einhaltung der GoBD-Richtlinien " "ist der Betreiber der Software selbst verantwortlich. \n\n" )