From ae986b4f3f04d31db79ec4816dff02f891ac9f0b Mon Sep 17 00:00:00 2001 From: dfuhrmann3 Date: Tue, 7 Jul 2026 13:43:13 +0200 Subject: [PATCH] refactor: extract pure invoice logic into invoice_logic.py for testability Move all dependency-free business logic (metadata schema, rounding/parsing helpers, item extraction, and the XRechnung/ZUGFeRD XML generators plus the PDF/A output-intent helpers) out of main.py into a new invoice_logic.py that imports only pydantic and pikepdf. This makes the logic unit-testable without a Streamlit runtime. Pure cut-and-paste, no behavior change; main.py now imports these names from invoice_logic instead of defining them locally. Co-Authored-By: Claude Opus 4.8 --- invoice_logic.py | 615 +++++++++++++++++++++++++++++++++++ main.py | 627 +----------------------------------- pytest.ini | 2 + requirements-dev.txt | 1 + tests/test_invoice_logic.py | 96 ++++++ 5 files changed, 731 insertions(+), 610 deletions(-) create mode 100644 invoice_logic.py create mode 100644 pytest.ini create mode 100644 requirements-dev.txt create mode 100644 tests/test_invoice_logic.py diff --git a/invoice_logic.py b/invoice_logic.py new file mode 100644 index 0000000..ed06ba4 --- /dev/null +++ b/invoice_logic.py @@ -0,0 +1,615 @@ +import os +import io +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 pikepdf +from pydantic import BaseModel, Field + + +# ========================================== +# 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") + customer_number: str = Field(default="", description="Kundennummer, wie sie beim Empfänger hinterlegt ist (z.B. der exakte Wert nach 'Kunden-Nr:'). Wird als Käuferreferenz (BT-10) verwendet. Falls im Text keine Kundennummer erkennbar ist: leer lassen.") + buyer_reference: str = Field(default="", description="Bestellnummer des KÄUFERS für diesen Auftrag (BT-13): der exakte Wert nach 'Ihr Auftrag:' o.ä. (z.B. 05.06.2026-Herr.Knospe). NICHT die Kundennummer (die gehört in customer_number) und NICHT unsere eigene Auftragsbestätigungsnummer (die gehört in order_id). Wenn im Text keine solche Käufer-Bestellnummer erkennbar ist: leer lassen.") + 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 EIGENE Auftragsbestätigungs-/Verkaufsauftragsnummer (BT-14, z.B. AB26-080) VOR Position 1. Das ist unsere eigene Nummer, NICHT die Bestellnummer des Käufers. 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.") + payment_term_days: str = Field(default="", description="Das Zahlungsziel in TAGEN als reine Zahl (z.B. '14' oder '30'). Suche im Text nach Formulierungen wie 'zahlbar innerhalb X Tagen', 'Zahlungsziel X Tage', 'X Tage netto'. Steht stattdessen ein konkretes Fälligkeitsdatum (z.B. 'zahlbar bis 31.07.2026'), berechne die Anzahl Tage zwischen Rechnungsdatum und diesem Datum und gib NUR diese Zahl zurück. Wenn im Text KEIN Zahlungsziel und KEIN Fälligkeitsdatum erkennbar ist: leer lassen (NICHT raten, NICHT 14 annehmen).") + 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 +# ========================================== + +# Zentrale Geschäftskonstanten - hier zentral pflegen statt an mehreren Stellen im Code. +VERKAEUFER_IBAN = "DE06120700240456207000" + +# Verkäufer-Stammdaten (SCITECH). Werte bleiben fest hinterlegt, sind aber an EINER Stelle +# gepflegt und werden in beiden XML-Generatoren sowie in der UI referenziert - so bleibt bei +# einer Adress-/Kontaktänderung nichts inkonsistent zurück. +VERKAEUFER = { + "name": "scitech it solutions GmbH", # BT-27 (Firmenname) + "email": "office@scitech-gmbh.de", # BT-34 (elektronische Adresse) / Kontakt + "phone": "+49 3302 4940537", # BT-42 (Kontakttelefon) + "contact_name": "Geschäftsführung", # BG-6 (Ansprechpartner) + "street": "Hermann-Schumann-Str. 4", # BT-35 + "zip": "16761", # BT-38 + "city": "Hennigsdorf", # BT-37 + "country": "DE", # BT-40 + "vat_id": "DE218436692", # BT-31 (Umsatzsteuer-ID) + "register_id": "HRB 6058 NP", # BT-30 (Handelsregisternummer, NICHT die USt-ID) +} + + +def berechne_faelligkeit(invoice_date_str, term_days): + """Fälligkeitsdatum = Rechnungsdatum + term_days. Gibt 'YYYY-MM-DD' zurück, oder leer, + wenn Rechnungsdatum unlesbar oder kein gültiges Zahlungsziel angegeben ist. KEIN fester + Fallback - fehlt das Zahlungsziel, bleibt das Feld leer und die UI blockiert den Export.""" + base = parse_to_strict_xml_date(invoice_date_str) + if not base: + return "" + try: + days = int(term_days) + except (ValueError, TypeError): + return "" + if days < 0: + return "" + try: + base_date = datetime.strptime(base, "%Y-%m-%d") + except (ValueError, TypeError): + return "" + return (base_date + timedelta(days=days)).strftime("%Y-%m-%d") + + +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): + """Normalisiert ein Datum auf 'YYYY-MM-DD'. Gibt leeren String zurück, wenn das Datum + fehlt ODER nicht geparst werden kann - NICHT still auf 'heute' fallen, damit fehlerhafte + Datumsangaben sichtbar werden statt unbemerkt durchzurutschen.""" + 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 "" + +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"]) + + _ubl_due = berechne_faelligkeit(data["invoice_date"], data.get("payment_term_days")) + if _ubl_due: + ET.SubElement(root, f"{{{CBC}}}DueDate").text = _ubl_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" + # BT-10 Käuferreferenz: NUR eintragen, wenn tatsächlich ein Wert vorliegt. Für private/ + # geschäftliche Empfänger (B2B) reicht die Kundennummer, die der Kunde bei euch hat; nur bei + # Behörden (B2G) ist eine echte Leitweg-ID vorgesehen - die kommt dann vom Empfänger selbst. + if data.get("customer_number") and str(data["customer_number"]).strip() != "": + ET.SubElement(root, f"{{{CBC}}}BuyerReference").text = str(data["customer_number"]).strip() + + 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"]) + + # BT-13 "Purchase order reference" (Bestellnummer des KÄUFERS, z.B. nach "Ihr Auftrag:") und + # BT-14 "Sales order reference" (unsere EIGENE Auftrags-/Verkaufsauftragsnummer) leben beide + # unter demselben cac:OrderReference-Element. cac:OrderReference/cbc:ID ist laut UBL-Schema + # zwingend, sobald OrderReference verwendet wird - liegt keine Käufer-Bestellnummer (BT-13) + # vor, aber eine eigene Nummer (BT-14), ist laut Peppol-Spezifikation der Platzhalter "NA" + # vorgeschrieben. + has_bt13 = data.get("buyer_reference") and str(data["buyer_reference"]).strip() != "" + has_bt14 = data.get("order_id") and str(data["order_id"]).strip() != "" + if has_bt13 or has_bt14: + order_ref = ET.SubElement(root, f"{{{CAC}}}OrderReference") + ET.SubElement(order_ref, f"{{{CBC}}}ID").text = str(data["buyer_reference"]).strip() if has_bt13 else "NA" + if has_bt14: + ET.SubElement(order_ref, f"{{{CBC}}}SalesOrderID").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 = VERKAEUFER["email"] + + addr_supp = ET.SubElement(party_supp, f"{{{CAC}}}PostalAddress") + ET.SubElement(addr_supp, f"{{{CBC}}}StreetName").text = VERKAEUFER["street"] + ET.SubElement(addr_supp, f"{{{CBC}}}CityName").text = VERKAEUFER["city"] + ET.SubElement(addr_supp, f"{{{CBC}}}PostalZone").text = VERKAEUFER["zip"] + country_supp = ET.SubElement(addr_supp, f"{{{CAC}}}Country") + ET.SubElement(country_supp, f"{{{CBC}}}IdentificationCode").text = VERKAEUFER["country"] + + tax_scheme_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyTaxScheme") + ET.SubElement(tax_scheme_supp, f"{{{CBC}}}CompanyID").text = VERKAEUFER["vat_id"] + 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 = VERKAEUFER["name"] + # BT-30 Registernummer (Handelsregister) - NICHT identisch mit der Umsatzsteuer-ID (BT-31)! + ET.SubElement(legal_supp, f"{{{CBC}}}CompanyID").text = VERKAEUFER["register_id"] + + contact_supp = ET.SubElement(party_supp, f"{{{CAC}}}Contact") + ET.SubElement(contact_supp, f"{{{CBC}}}Name").text = VERKAEUFER["contact_name"] + ET.SubElement(contact_supp, f"{{{CBC}}}Telephone").text = VERKAEUFER["phone"] + ET.SubElement(contact_supp, f"{{{CBC}}}ElectronicMail").text = VERKAEUFER["email"] + + # --- 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.get("recipient_email", "")).strip() + 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") + # BT-81 Zahlungsart: Code 58 = SEPA-Überweisung (statt 30, das für Überweisungen + # außerhalb des SEPA-Raums gedacht ist - VD-Valitool-169-Hinweis). + ET.SubElement(payment_means, f"{{{CBC}}}PaymentMeansCode").text = "58" + payee_acc = ET.SubElement(payment_means, f"{{{CAC}}}PayeeFinancialAccount") + ET.SubElement(payee_acc, f"{{{CBC}}}ID").text = str(data["iban"]).replace(" ", "") + + # --- STEUERN (Gezielt erzwungener CBC Namespace für Beträge) --- + 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") + + # FIX: cbc:Description MUSS laut XSD-Schema zwingend VOR cbc:Name stehen! + 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" + QDT = "urn:un:unece:uncefact:data:standard:QualifiedDataType:100" + + ET.register_namespace('rsm', RSM) + ET.register_namespace('ram', RAM) + ET.register_namespace('udt', UDT) + ET.register_namespace('qdt', QDT) + + root = ET.Element(f"{{{RSM}}}CrossIndustryInvoice") + # VD-Valitool-14: Der qdt-Namespace wird in dieser Instanz nicht benötigt, soll aber laut + # Spezifikation trotzdem am Root-Element deklariert werden. + root.set("xmlns:qdt", QDT) + + ex_doc_ctx = ET.SubElement(root, f"{{{RSM}}}ExchangedDocumentContext") + + # BT-23 Geschäftsprozess (PEPPOL-EN16931-R001 Pflicht). Muss laut XSD VOR + # GuidelineSpecifiedDocumentContextParameter stehen. + biz_process = ET.SubElement(ex_doc_ctx, f"{{{RAM}}}BusinessProcessSpecifiedDocumentContextParameter") + ET.SubElement(biz_process, f"{{{RAM}}}ID").text = "urn:fdc:peppol.eu:2017:poacc:billing:01:1.0" + + 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}}}NetPriceProductTradePrice") + 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") + # BT-10 Käuferreferenz: NUR eintragen, wenn tatsächlich ein Wert vorliegt (siehe UBL-Generator + # für die ausführliche Begründung). + if data.get("customer_number") and str(data["customer_number"]).strip() != "": + ET.SubElement(header_agree, f"{{{RAM}}}BuyerReference").text = str(data["customer_number"]).strip() + + supplier_party = ET.SubElement(header_agree, f"{{{RAM}}}SellerTradeParty") + ET.SubElement(supplier_party, f"{{{RAM}}}Name").text = VERKAEUFER["name"] + + supp_legal = ET.SubElement(supplier_party, f"{{{RAM}}}SpecifiedLegalOrganization") + # BT-30 Registernummer (Handelsregister) - NICHT identisch mit der Umsatzsteuer-ID (BT-31)! + # Fest eingetragen, analog zu den übrigen fixen Verkäufer-Stammdaten. + ET.SubElement(supp_legal, f"{{{RAM}}}ID").text = VERKAEUFER["register_id"] + + # BG-6 SELLER CONTACT (BR-DE-2 Pflicht für XRechnung). Muss laut XSD nach + # SpecifiedLegalOrganization und vor PostalTradeAddress stehen. + supp_contact = ET.SubElement(supplier_party, f"{{{RAM}}}DefinedTradeContact") + ET.SubElement(supp_contact, f"{{{RAM}}}PersonName").text = VERKAEUFER["contact_name"] + supp_contact_tel = ET.SubElement(supp_contact, f"{{{RAM}}}TelephoneUniversalCommunication") + ET.SubElement(supp_contact_tel, f"{{{RAM}}}CompleteNumber").text = VERKAEUFER["phone"] + supp_contact_mail = ET.SubElement(supp_contact, f"{{{RAM}}}EmailURIUniversalCommunication") + ET.SubElement(supp_contact_mail, f"{{{RAM}}}URIID").text = VERKAEUFER["email"] + + supp_addr = ET.SubElement(supplier_party, f"{{{RAM}}}PostalTradeAddress") + ET.SubElement(supp_addr, f"{{{RAM}}}PostcodeCode").text = VERKAEUFER["zip"] + ET.SubElement(supp_addr, f"{{{RAM}}}LineOne").text = VERKAEUFER["street"] + ET.SubElement(supp_addr, f"{{{RAM}}}CityName").text = VERKAEUFER["city"] + ET.SubElement(supp_addr, f"{{{RAM}}}CountryID").text = VERKAEUFER["country"] + + # BT-34 Elektronische Adresse des Verkäufers (PEPPOL-EN16931-R020 Pflicht). Muss laut + # XSD nach PostalTradeAddress und vor SpecifiedTaxRegistration stehen. + supp_uri = ET.SubElement(supplier_party, f"{{{RAM}}}URIUniversalCommunication") + ET.SubElement(supp_uri, f"{{{RAM}}}URIID", {"schemeID": "EM"}).text = VERKAEUFER["email"] + + supp_tax = ET.SubElement(supplier_party, f"{{{RAM}}}SpecifiedTaxRegistration") + ET.SubElement(supp_tax, f"{{{RAM}}}ID", {"schemeID": "VA"}).text = VERKAEUFER["vat_id"] + + 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" + + # BT-49 Elektronische Adresse des Käufers (PEPPOL-EN16931-R010 Pflicht). Muss laut + # XSD nach PostalTradeAddress stehen. KEIN erfundener Fallback - fehlt die Adresse, wird das + # (leere) Pflichtfeld sichtbar und die UI blockiert den Export mit Warnung. + cust_uri = ET.SubElement(customer_party, f"{{{RAM}}}URIUniversalCommunication") + ET.SubElement(cust_uri, f"{{{RAM}}}URIID", {"schemeID": "EM"}).text = str(data.get("recipient_email", "")).strip() + + if data.get("order_id") and str(data["order_id"]).strip() != "": + # BT-14 "Sales order reference" - unsere EIGENE Auftrags-/Verkaufsauftragsnummer. + # In CII ein eigenständiges Element. Muss laut XSD VOR BuyerOrderReferencedDocument stehen. + order_ref = ET.SubElement(header_agree, f"{{{RAM}}}SellerOrderReferencedDocument") + ET.SubElement(order_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["order_id"]).strip() + + if data.get("buyer_reference") and str(data["buyer_reference"]).strip() != "": + # BT-13 "Purchase order reference" - die Bestellnummer des KÄUFERS (z.B. nach + # "Ihr Auftrag:"), NICHT unsere eigene Nummer (die ist BT-14/order_id oben). + buyer_order_ref = ET.SubElement(header_agree, f"{{{RAM}}}BuyerOrderReferencedDocument") + ET.SubElement(buyer_order_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["buyer_reference"]).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_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"]) + + # HINWEIS: "DeliveryNoteReferencedDocument" existiert im EN16931/Factur-X-Schema NICHT + # (im offiziellen XSD nicht vorhanden). Das korrekte, schema-konforme Element für eine + # Lieferschein-/Lieferavis-Referenz ist "DespatchAdviceReferencedDocument" und muss laut + # XSD-Sequenz NACH ActualDeliverySupplyChainEvent stehen. + if data.get("delivery_note_id") and str(data["delivery_note_id"]).strip() != "": + dn_ref = ET.SubElement(header_deliv, f"{{{RAM}}}DespatchAdviceReferencedDocument") + ET.SubElement(dn_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["delivery_note_id"]).strip() + + 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}}}SpecifiedTradeSettlementPaymentMeans") + # BT-81 Zahlungsart: Code 58 = SEPA-Überweisung. Code 30 ist für Überweisungen AUSSERHALB + # des SEPA-Raums gedacht (VD-Valitool-169-Hinweis) - da hier immer eine IBAN verwendet wird, + # ist 58 der korrekte Code. + ET.SubElement(pay_means, f"{{{RAM}}}TypeCode").text = "58" + 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"]) + + # BT-9 Fälligkeitsdatum, abgeleitet aus dem tatsächlichen Zahlungsziel der Rechnung. Erfüllt + # allein BR-CO-25 (Fälligkeitsdatum ODER Zahlungsbedingungen - wir nutzen nur das Datum). + # Kein fester Fallback: ohne gültiges Zahlungsziel wird der Block weggelassen (die UI blockiert + # dann den Export). + _cii_due = berechne_faelligkeit(data["invoice_date"], data.get("payment_term_days")) + if _cii_due: + payment_terms = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradePaymentTerms") + due_date_elem = ET.SubElement(payment_terms, f"{{{RAM}}}DueDateDateTime") + ET.SubElement(due_date_elem, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = _cii_due.replace("-", "") + + 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}" + 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=" ") + + +# Pfad zum sRGB-ICC-Profil, das für ein PDF/A-3-konformes OutputIntent benötigt wird. +# (PDF/A verlangt zwingend ein eingebettetes Farbprofil - ohne das gilt die PDF nicht als PDF/A, +# selbst wenn XMP-Metadaten und eingebettetes XML korrekt sind.) +SRGB_ICC_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sRGB.icc") + + +def add_pdfa_output_intent(pdf_bytes: bytes, icc_path: str = SRGB_ICC_PATH) -> bytes: + """ + Fügt der PDF ein OutputIntent mit sRGB-ICC-Profil hinzu, falls noch keines vorhanden ist. + Das ist zwingend erforderlich, damit eine PDF als PDF/A-3(B) gilt (ZUGFeRD-Pflicht). + """ + pdf = pikepdf.open(io.BytesIO(pdf_bytes)) + try: + if "/OutputIntents" not in pdf.Root or len(pdf.Root.get("/OutputIntents", [])) == 0: + if not os.path.isfile(icc_path): + raise FileNotFoundError( + f"sRGB-ICC-Profil nicht gefunden unter '{icc_path}'. " + "Für echte PDF/A-3-Konformität muss diese Datei mit ausgeliefert werden." + ) + with open(icc_path, "rb") as f: + icc_bytes = f.read() + icc_stream = pikepdf.Stream(pdf, icc_bytes) + icc_stream["/N"] = 3 + icc_stream["/Alternate"] = pikepdf.Name("/DeviceRGB") + output_intent = pdf.make_indirect(pikepdf.Dictionary({ + "/Type": pikepdf.Name("/OutputIntent"), + "/S": pikepdf.Name("/GTS_PDFA1"), + "/OutputConditionIdentifier": pikepdf.String("sRGB IEC61966-2.1"), + "/Info": pikepdf.String("sRGB IEC61966-2.1"), + "/DestOutputProfile": icc_stream, + })) + pdf.Root["/OutputIntents"] = pikepdf.Array([output_intent]) + buf = io.BytesIO() + pdf.save(buf) + return buf.getvalue() + finally: + pdf.close() + + +def fix_xrechnung_xmp_conformance(pdf_bytes: bytes) -> bytes: + """ + Die factur-x-Bibliothek schreibt standardmäßig fx:ConformanceLevel = 'EN 16931' und + fx:Version = '1.0' in die XMP-Metadaten (generischer Factur-X-Standard). Da unsere CII-XML + aber explizit das KoSIT-XRechnung-Profil deklariert (GuidelineID .../xrechnung_3.0), erwarten + Validatoren hier stattdessen 'XRECHNUNG' und '3.0' - sonst schlagen z. B. + VD-Valitool-H-022/H-024 fehl, weil PDF-Metadaten und eingebettetes XML nicht zusammenpassen. + """ + pdf = pikepdf.open(io.BytesIO(pdf_bytes)) + try: + if "/Metadata" in pdf.Root: + xmp_bytes = pdf.Root.Metadata.read_bytes() + xmp_str = xmp_bytes.decode("utf-8") + xmp_str = xmp_str.replace( + "EN 16931", + "XRECHNUNG", + ).replace( + "1.0", + "3.0", + ) + pdf.Root.Metadata.write(xmp_str.encode("utf-8")) + buf = io.BytesIO() + pdf.save(buf) + return buf.getvalue() + finally: + pdf.close() diff --git a/main.py b/main.py index e74a049..baad7f9 100644 --- a/main.py +++ b/main.py @@ -3,15 +3,26 @@ import io 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 -import pikepdf from facturx import generate_from_binary as facturx_generate_from_binary +from invoice_logic import ( + InvoiceMetadata, + VERKAEUFER, + VERKAEUFER_IBAN, + 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, +) # ========================================== # 0. SITZUNGS-STATUS INITIALISIEREN (ZÄHLER GEGEN WIDGET-CACHING) @@ -62,610 +73,6 @@ class LLMClient: 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") - customer_number: str = Field(default="", description="Kundennummer, wie sie beim Empfänger hinterlegt ist (z.B. der exakte Wert nach 'Kunden-Nr:'). Wird als Käuferreferenz (BT-10) verwendet. Falls im Text keine Kundennummer erkennbar ist: leer lassen.") - buyer_reference: str = Field(default="", description="Bestellnummer des KÄUFERS für diesen Auftrag (BT-13): der exakte Wert nach 'Ihr Auftrag:' o.ä. (z.B. 05.06.2026-Herr.Knospe). NICHT die Kundennummer (die gehört in customer_number) und NICHT unsere eigene Auftragsbestätigungsnummer (die gehört in order_id). Wenn im Text keine solche Käufer-Bestellnummer erkennbar ist: leer lassen.") - 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 EIGENE Auftragsbestätigungs-/Verkaufsauftragsnummer (BT-14, z.B. AB26-080) VOR Position 1. Das ist unsere eigene Nummer, NICHT die Bestellnummer des Käufers. 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.") - payment_term_days: str = Field(default="", description="Das Zahlungsziel in TAGEN als reine Zahl (z.B. '14' oder '30'). Suche im Text nach Formulierungen wie 'zahlbar innerhalb X Tagen', 'Zahlungsziel X Tage', 'X Tage netto'. Steht stattdessen ein konkretes Fälligkeitsdatum (z.B. 'zahlbar bis 31.07.2026'), berechne die Anzahl Tage zwischen Rechnungsdatum und diesem Datum und gib NUR diese Zahl zurück. Wenn im Text KEIN Zahlungsziel und KEIN Fälligkeitsdatum erkennbar ist: leer lassen (NICHT raten, NICHT 14 annehmen).") - 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 -# ========================================== - -# Zentrale Geschäftskonstanten - hier zentral pflegen statt an mehreren Stellen im Code. -VERKAEUFER_IBAN = "DE06120700240456207000" - -# Verkäufer-Stammdaten (SCITECH). Werte bleiben fest hinterlegt, sind aber an EINER Stelle -# gepflegt und werden in beiden XML-Generatoren sowie in der UI referenziert - so bleibt bei -# einer Adress-/Kontaktänderung nichts inkonsistent zurück. -VERKAEUFER = { - "name": "scitech it solutions GmbH", # BT-27 (Firmenname) - "email": "office@scitech-gmbh.de", # BT-34 (elektronische Adresse) / Kontakt - "phone": "+49 3302 4940537", # BT-42 (Kontakttelefon) - "contact_name": "Geschäftsführung", # BG-6 (Ansprechpartner) - "street": "Hermann-Schumann-Str. 4", # BT-35 - "zip": "16761", # BT-38 - "city": "Hennigsdorf", # BT-37 - "country": "DE", # BT-40 - "vat_id": "DE218436692", # BT-31 (Umsatzsteuer-ID) - "register_id": "HRB 6058 NP", # BT-30 (Handelsregisternummer, NICHT die USt-ID) -} - - -def berechne_faelligkeit(invoice_date_str, term_days): - """Fälligkeitsdatum = Rechnungsdatum + term_days. Gibt 'YYYY-MM-DD' zurück, oder leer, - wenn Rechnungsdatum unlesbar oder kein gültiges Zahlungsziel angegeben ist. KEIN fester - Fallback - fehlt das Zahlungsziel, bleibt das Feld leer und die UI blockiert den Export.""" - base = parse_to_strict_xml_date(invoice_date_str) - if not base: - return "" - try: - days = int(term_days) - except (ValueError, TypeError): - return "" - if days < 0: - return "" - try: - base_date = datetime.strptime(base, "%Y-%m-%d") - except (ValueError, TypeError): - return "" - return (base_date + timedelta(days=days)).strftime("%Y-%m-%d") - - -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): - """Normalisiert ein Datum auf 'YYYY-MM-DD'. Gibt leeren String zurück, wenn das Datum - fehlt ODER nicht geparst werden kann - NICHT still auf 'heute' fallen, damit fehlerhafte - Datumsangaben sichtbar werden statt unbemerkt durchzurutschen.""" - 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 "" - -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"]) - - _ubl_due = berechne_faelligkeit(data["invoice_date"], data.get("payment_term_days")) - if _ubl_due: - ET.SubElement(root, f"{{{CBC}}}DueDate").text = _ubl_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" - # BT-10 Käuferreferenz: NUR eintragen, wenn tatsächlich ein Wert vorliegt. Für private/ - # geschäftliche Empfänger (B2B) reicht die Kundennummer, die der Kunde bei euch hat; nur bei - # Behörden (B2G) ist eine echte Leitweg-ID vorgesehen - die kommt dann vom Empfänger selbst. - if data.get("customer_number") and str(data["customer_number"]).strip() != "": - ET.SubElement(root, f"{{{CBC}}}BuyerReference").text = str(data["customer_number"]).strip() - - 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"]) - - # BT-13 "Purchase order reference" (Bestellnummer des KÄUFERS, z.B. nach "Ihr Auftrag:") und - # BT-14 "Sales order reference" (unsere EIGENE Auftrags-/Verkaufsauftragsnummer) leben beide - # unter demselben cac:OrderReference-Element. cac:OrderReference/cbc:ID ist laut UBL-Schema - # zwingend, sobald OrderReference verwendet wird - liegt keine Käufer-Bestellnummer (BT-13) - # vor, aber eine eigene Nummer (BT-14), ist laut Peppol-Spezifikation der Platzhalter "NA" - # vorgeschrieben. - has_bt13 = data.get("buyer_reference") and str(data["buyer_reference"]).strip() != "" - has_bt14 = data.get("order_id") and str(data["order_id"]).strip() != "" - if has_bt13 or has_bt14: - order_ref = ET.SubElement(root, f"{{{CAC}}}OrderReference") - ET.SubElement(order_ref, f"{{{CBC}}}ID").text = str(data["buyer_reference"]).strip() if has_bt13 else "NA" - if has_bt14: - ET.SubElement(order_ref, f"{{{CBC}}}SalesOrderID").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 = VERKAEUFER["email"] - - addr_supp = ET.SubElement(party_supp, f"{{{CAC}}}PostalAddress") - ET.SubElement(addr_supp, f"{{{CBC}}}StreetName").text = VERKAEUFER["street"] - ET.SubElement(addr_supp, f"{{{CBC}}}CityName").text = VERKAEUFER["city"] - ET.SubElement(addr_supp, f"{{{CBC}}}PostalZone").text = VERKAEUFER["zip"] - country_supp = ET.SubElement(addr_supp, f"{{{CAC}}}Country") - ET.SubElement(country_supp, f"{{{CBC}}}IdentificationCode").text = VERKAEUFER["country"] - - tax_scheme_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyTaxScheme") - ET.SubElement(tax_scheme_supp, f"{{{CBC}}}CompanyID").text = VERKAEUFER["vat_id"] - 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 = VERKAEUFER["name"] - # BT-30 Registernummer (Handelsregister) - NICHT identisch mit der Umsatzsteuer-ID (BT-31)! - ET.SubElement(legal_supp, f"{{{CBC}}}CompanyID").text = VERKAEUFER["register_id"] - - contact_supp = ET.SubElement(party_supp, f"{{{CAC}}}Contact") - ET.SubElement(contact_supp, f"{{{CBC}}}Name").text = VERKAEUFER["contact_name"] - ET.SubElement(contact_supp, f"{{{CBC}}}Telephone").text = VERKAEUFER["phone"] - ET.SubElement(contact_supp, f"{{{CBC}}}ElectronicMail").text = VERKAEUFER["email"] - - # --- 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.get("recipient_email", "")).strip() - 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") - # BT-81 Zahlungsart: Code 58 = SEPA-Überweisung (statt 30, das für Überweisungen - # außerhalb des SEPA-Raums gedacht ist - VD-Valitool-169-Hinweis). - ET.SubElement(payment_means, f"{{{CBC}}}PaymentMeansCode").text = "58" - payee_acc = ET.SubElement(payment_means, f"{{{CAC}}}PayeeFinancialAccount") - ET.SubElement(payee_acc, f"{{{CBC}}}ID").text = str(data["iban"]).replace(" ", "") - - # --- STEUERN (Gezielt erzwungener CBC Namespace für Beträge) --- - 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") - - # FIX: cbc:Description MUSS laut XSD-Schema zwingend VOR cbc:Name stehen! - 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" - QDT = "urn:un:unece:uncefact:data:standard:QualifiedDataType:100" - - ET.register_namespace('rsm', RSM) - ET.register_namespace('ram', RAM) - ET.register_namespace('udt', UDT) - ET.register_namespace('qdt', QDT) - - root = ET.Element(f"{{{RSM}}}CrossIndustryInvoice") - # VD-Valitool-14: Der qdt-Namespace wird in dieser Instanz nicht benötigt, soll aber laut - # Spezifikation trotzdem am Root-Element deklariert werden. - root.set("xmlns:qdt", QDT) - - ex_doc_ctx = ET.SubElement(root, f"{{{RSM}}}ExchangedDocumentContext") - - # BT-23 Geschäftsprozess (PEPPOL-EN16931-R001 Pflicht). Muss laut XSD VOR - # GuidelineSpecifiedDocumentContextParameter stehen. - biz_process = ET.SubElement(ex_doc_ctx, f"{{{RAM}}}BusinessProcessSpecifiedDocumentContextParameter") - ET.SubElement(biz_process, f"{{{RAM}}}ID").text = "urn:fdc:peppol.eu:2017:poacc:billing:01:1.0" - - 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}}}NetPriceProductTradePrice") - 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") - # BT-10 Käuferreferenz: NUR eintragen, wenn tatsächlich ein Wert vorliegt (siehe UBL-Generator - # für die ausführliche Begründung). - if data.get("customer_number") and str(data["customer_number"]).strip() != "": - ET.SubElement(header_agree, f"{{{RAM}}}BuyerReference").text = str(data["customer_number"]).strip() - - supplier_party = ET.SubElement(header_agree, f"{{{RAM}}}SellerTradeParty") - ET.SubElement(supplier_party, f"{{{RAM}}}Name").text = VERKAEUFER["name"] - - supp_legal = ET.SubElement(supplier_party, f"{{{RAM}}}SpecifiedLegalOrganization") - # BT-30 Registernummer (Handelsregister) - NICHT identisch mit der Umsatzsteuer-ID (BT-31)! - # Fest eingetragen, analog zu den übrigen fixen Verkäufer-Stammdaten. - ET.SubElement(supp_legal, f"{{{RAM}}}ID").text = VERKAEUFER["register_id"] - - # BG-6 SELLER CONTACT (BR-DE-2 Pflicht für XRechnung). Muss laut XSD nach - # SpecifiedLegalOrganization und vor PostalTradeAddress stehen. - supp_contact = ET.SubElement(supplier_party, f"{{{RAM}}}DefinedTradeContact") - ET.SubElement(supp_contact, f"{{{RAM}}}PersonName").text = VERKAEUFER["contact_name"] - supp_contact_tel = ET.SubElement(supp_contact, f"{{{RAM}}}TelephoneUniversalCommunication") - ET.SubElement(supp_contact_tel, f"{{{RAM}}}CompleteNumber").text = VERKAEUFER["phone"] - supp_contact_mail = ET.SubElement(supp_contact, f"{{{RAM}}}EmailURIUniversalCommunication") - ET.SubElement(supp_contact_mail, f"{{{RAM}}}URIID").text = VERKAEUFER["email"] - - supp_addr = ET.SubElement(supplier_party, f"{{{RAM}}}PostalTradeAddress") - ET.SubElement(supp_addr, f"{{{RAM}}}PostcodeCode").text = VERKAEUFER["zip"] - ET.SubElement(supp_addr, f"{{{RAM}}}LineOne").text = VERKAEUFER["street"] - ET.SubElement(supp_addr, f"{{{RAM}}}CityName").text = VERKAEUFER["city"] - ET.SubElement(supp_addr, f"{{{RAM}}}CountryID").text = VERKAEUFER["country"] - - # BT-34 Elektronische Adresse des Verkäufers (PEPPOL-EN16931-R020 Pflicht). Muss laut - # XSD nach PostalTradeAddress und vor SpecifiedTaxRegistration stehen. - supp_uri = ET.SubElement(supplier_party, f"{{{RAM}}}URIUniversalCommunication") - ET.SubElement(supp_uri, f"{{{RAM}}}URIID", {"schemeID": "EM"}).text = VERKAEUFER["email"] - - supp_tax = ET.SubElement(supplier_party, f"{{{RAM}}}SpecifiedTaxRegistration") - ET.SubElement(supp_tax, f"{{{RAM}}}ID", {"schemeID": "VA"}).text = VERKAEUFER["vat_id"] - - 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" - - # BT-49 Elektronische Adresse des Käufers (PEPPOL-EN16931-R010 Pflicht). Muss laut - # XSD nach PostalTradeAddress stehen. KEIN erfundener Fallback - fehlt die Adresse, wird das - # (leere) Pflichtfeld sichtbar und die UI blockiert den Export mit Warnung. - cust_uri = ET.SubElement(customer_party, f"{{{RAM}}}URIUniversalCommunication") - ET.SubElement(cust_uri, f"{{{RAM}}}URIID", {"schemeID": "EM"}).text = str(data.get("recipient_email", "")).strip() - - if data.get("order_id") and str(data["order_id"]).strip() != "": - # BT-14 "Sales order reference" - unsere EIGENE Auftrags-/Verkaufsauftragsnummer. - # In CII ein eigenständiges Element. Muss laut XSD VOR BuyerOrderReferencedDocument stehen. - order_ref = ET.SubElement(header_agree, f"{{{RAM}}}SellerOrderReferencedDocument") - ET.SubElement(order_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["order_id"]).strip() - - if data.get("buyer_reference") and str(data["buyer_reference"]).strip() != "": - # BT-13 "Purchase order reference" - die Bestellnummer des KÄUFERS (z.B. nach - # "Ihr Auftrag:"), NICHT unsere eigene Nummer (die ist BT-14/order_id oben). - buyer_order_ref = ET.SubElement(header_agree, f"{{{RAM}}}BuyerOrderReferencedDocument") - ET.SubElement(buyer_order_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["buyer_reference"]).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_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"]) - - # HINWEIS: "DeliveryNoteReferencedDocument" existiert im EN16931/Factur-X-Schema NICHT - # (im offiziellen XSD nicht vorhanden). Das korrekte, schema-konforme Element für eine - # Lieferschein-/Lieferavis-Referenz ist "DespatchAdviceReferencedDocument" und muss laut - # XSD-Sequenz NACH ActualDeliverySupplyChainEvent stehen. - if data.get("delivery_note_id") and str(data["delivery_note_id"]).strip() != "": - dn_ref = ET.SubElement(header_deliv, f"{{{RAM}}}DespatchAdviceReferencedDocument") - ET.SubElement(dn_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["delivery_note_id"]).strip() - - 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}}}SpecifiedTradeSettlementPaymentMeans") - # BT-81 Zahlungsart: Code 58 = SEPA-Überweisung. Code 30 ist für Überweisungen AUSSERHALB - # des SEPA-Raums gedacht (VD-Valitool-169-Hinweis) - da hier immer eine IBAN verwendet wird, - # ist 58 der korrekte Code. - ET.SubElement(pay_means, f"{{{RAM}}}TypeCode").text = "58" - 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"]) - - # BT-9 Fälligkeitsdatum, abgeleitet aus dem tatsächlichen Zahlungsziel der Rechnung. Erfüllt - # allein BR-CO-25 (Fälligkeitsdatum ODER Zahlungsbedingungen - wir nutzen nur das Datum). - # Kein fester Fallback: ohne gültiges Zahlungsziel wird der Block weggelassen (die UI blockiert - # dann den Export). - _cii_due = berechne_faelligkeit(data["invoice_date"], data.get("payment_term_days")) - if _cii_due: - payment_terms = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradePaymentTerms") - due_date_elem = ET.SubElement(payment_terms, f"{{{RAM}}}DueDateDateTime") - ET.SubElement(due_date_elem, f"{{{UDT}}}DateTimeString", {"format": "102"}).text = _cii_due.replace("-", "") - - 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}" - 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=" ") - - -# Pfad zum sRGB-ICC-Profil, das für ein PDF/A-3-konformes OutputIntent benötigt wird. -# (PDF/A verlangt zwingend ein eingebettetes Farbprofil - ohne das gilt die PDF nicht als PDF/A, -# selbst wenn XMP-Metadaten und eingebettetes XML korrekt sind.) -SRGB_ICC_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sRGB.icc") - - -def add_pdfa_output_intent(pdf_bytes: bytes, icc_path: str = SRGB_ICC_PATH) -> bytes: - """ - Fügt der PDF ein OutputIntent mit sRGB-ICC-Profil hinzu, falls noch keines vorhanden ist. - Das ist zwingend erforderlich, damit eine PDF als PDF/A-3(B) gilt (ZUGFeRD-Pflicht). - """ - pdf = pikepdf.open(io.BytesIO(pdf_bytes)) - try: - if "/OutputIntents" not in pdf.Root or len(pdf.Root.get("/OutputIntents", [])) == 0: - if not os.path.isfile(icc_path): - raise FileNotFoundError( - f"sRGB-ICC-Profil nicht gefunden unter '{icc_path}'. " - "Für echte PDF/A-3-Konformität muss diese Datei mit ausgeliefert werden." - ) - with open(icc_path, "rb") as f: - icc_bytes = f.read() - icc_stream = pikepdf.Stream(pdf, icc_bytes) - icc_stream["/N"] = 3 - icc_stream["/Alternate"] = pikepdf.Name("/DeviceRGB") - output_intent = pdf.make_indirect(pikepdf.Dictionary({ - "/Type": pikepdf.Name("/OutputIntent"), - "/S": pikepdf.Name("/GTS_PDFA1"), - "/OutputConditionIdentifier": pikepdf.String("sRGB IEC61966-2.1"), - "/Info": pikepdf.String("sRGB IEC61966-2.1"), - "/DestOutputProfile": icc_stream, - })) - pdf.Root["/OutputIntents"] = pikepdf.Array([output_intent]) - buf = io.BytesIO() - pdf.save(buf) - return buf.getvalue() - finally: - pdf.close() - - -def fix_xrechnung_xmp_conformance(pdf_bytes: bytes) -> bytes: - """ - Die factur-x-Bibliothek schreibt standardmäßig fx:ConformanceLevel = 'EN 16931' und - fx:Version = '1.0' in die XMP-Metadaten (generischer Factur-X-Standard). Da unsere CII-XML - aber explizit das KoSIT-XRechnung-Profil deklariert (GuidelineID .../xrechnung_3.0), erwarten - Validatoren hier stattdessen 'XRECHNUNG' und '3.0' - sonst schlagen z. B. - VD-Valitool-H-022/H-024 fehl, weil PDF-Metadaten und eingebettetes XML nicht zusammenpassen. - """ - pdf = pikepdf.open(io.BytesIO(pdf_bytes)) - try: - if "/Metadata" in pdf.Root: - xmp_bytes = pdf.Root.Metadata.read_bytes() - xmp_str = xmp_bytes.decode("utf-8") - xmp_str = xmp_str.replace( - "EN 16931", - "XRECHNUNG", - ).replace( - "1.0", - "3.0", - ) - pdf.Root.Metadata.write(xmp_str.encode("utf-8")) - buf = io.BytesIO() - pdf.save(buf) - return buf.getvalue() - finally: - pdf.close() - # ========================================== # 4. API CLIENT INITIALISIERUNG @@ -1027,4 +434,4 @@ st.caption( "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" -) \ No newline at end of file +) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..a635c5c --- /dev/null +++ b/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +pythonpath = . diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..d197ada --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1 @@ +pytest==8.3.4 diff --git a/tests/test_invoice_logic.py b/tests/test_invoice_logic.py new file mode 100644 index 0000000..1e95427 --- /dev/null +++ b/tests/test_invoice_logic.py @@ -0,0 +1,96 @@ +from decimal import Decimal +import xml.etree.ElementTree as ET + +from invoice_logic import ( + berechne_faelligkeit, + kaufmaennisches_runden, + clean_german_float, + extract_invoice_items_pure_python, + map_unece_unit_code, + parse_to_strict_xml_date, + generate_compliant_xrechnung, + generate_compliant_zugferd_cii, + VERKAEUFER_IBAN, +) + + +def _single_rate_invoice(): + """Eine einfache Einzelsteuersatz-Rechnung, wie main.py sie vor Phase 1 erzeugt hat.""" + items = [ + {"name": "Beratung", "description": "", "quantity": 2.0, "unit": "Std.", + "unit_price": 100.0, "total_amount": 200.0}, + ] + return { + "invoice_type_code": "380", + "invoice_number": "R26-001", + "invoice_date": "2026-01-15", + "delivery_date": "", + "buyer_reference": "", + "customer_number": "", + "recipient_name": "Test GmbH", + "recipient_email": "kunde@example.com", + "recipient_street": "Musterweg 1", + "recipient_zip": "12345", + "recipient_city": "Berlin", + "items": items, + "total_net_amount": 200.0, + "total_vat_amount": 38.0, + "vat_rate_percent": 19.0, + "total_gross_amount": 238.0, + "iban": VERKAEUFER_IBAN, + "delivery_note_id": "", + "order_id": "", + "contract_id": "", + "service_start_date": "", + "service_end_date": "", + "payment_term_days": "14", + "invoice_note": "", + } + + +def test_generate_compliant_xrechnung_single_rate_unchanged(): + xml_str = generate_compliant_xrechnung(_single_rate_invoice()) + root = ET.fromstring(xml_str.encode("utf-8")) + ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2", + "cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"} + assert root.find("cbc:ID", ns).text == "R26-001" + tax_amount = root.find(".//cac:TaxTotal/cbc:TaxAmount", ns).text + assert tax_amount == "38.00" + payable = root.find(".//cac:LegalMonetaryTotal/cbc:PayableAmount", ns).text + assert payable == "238.00" + + +def test_generate_compliant_zugferd_cii_single_rate_unchanged(): + xml_str = generate_compliant_zugferd_cii(_single_rate_invoice()) + root = ET.fromstring(xml_str.encode("utf-8")) + ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"} + grand_total = root.find(".//ram:SpecifiedTradeSettlementHeaderMonetarySummation/ram:GrandTotalAmount", ns).text + assert grand_total == "238.00" + + +def test_berechne_faelligkeit_basic(): + assert berechne_faelligkeit("2026-01-15", "14") == "2026-01-29" + + +def test_kaufmaennisches_runden_half_up(): + assert kaufmaennisches_runden(1.005) == 1.01 + + +def test_clean_german_float_thousands_and_comma(): + assert clean_german_float("1.234,56") == 1234.56 + + +def test_extract_invoice_items_pure_python_basic(): + text = "1\nBeratung\n2,00 Std.\n100,00\n200,00\n" + items = extract_invoice_items_pure_python(text) + assert items[0]["name"] == "Beratung" + assert items[0]["quantity"] == 2.0 + assert items[0]["unit_price"] == 100.0 + + +def test_map_unece_unit_code_hours(): + assert map_unece_unit_code("Std.") == "HUR" + + +def test_parse_to_strict_xml_date_german_format(): + assert parse_to_strict_xml_date("15.01.2026") == "2026-01-15"