Compare commits

..

No commits in common. "656040b64b6dbb95d6f2c3e16393320f23f2a836" and "994ad03387e729de2d9b68ded7fd97ab289cfcc7" have entirely different histories.

11 changed files with 628 additions and 3024 deletions

1
.gitignore vendored
View File

@ -1,5 +1,4 @@
venv
models
.superpowers/

Binary file not shown.

File diff suppressed because it is too large Load Diff

View File

@ -1,868 +0,0 @@
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)
}
# UN/EDIFACT 4461 Zahlungsmittelcodes (BT-81), Auswahl der fuer B2B/B2G relevanten Codes.
# "58" bleibt der Standardwert (SEPA-Ueberweisung), da alle bisherigen Rechnungen diesen nutzen.
PAYMENT_MEANS_CODES = {
"58": "SEPA-Überweisung",
"30": "Überweisung (allgemein)",
"31": "SEPA-Lastschrift",
"42": "Einzahlung auf Bankkonto",
"48": "Bankkarte (Kredit-/Debitkarte)",
}
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 format_skonto_text(skonto_percent, skonto_days, invoice_date_str):
"""Baut den Freitext fuer eine Skonto-Vereinbarung. EN16931 kennt kein eigenes strukturiertes
Skonto-Feld - der Text wird daher als cbc:Note (UBL) bzw. ram:Description (CII) neben dem
regulaeren Faelligkeitsdatum in den Zahlungsbedingungen (BT-20) abgelegt. Gibt "" zurueck,
wenn kein gueltiges Skonto vereinbart ist (KEIN Rateraten bei fehlenden Angaben)."""
try:
percent = float(skonto_percent)
except (TypeError, ValueError):
return ""
if percent <= 0:
return ""
try:
days = int(skonto_days)
except (TypeError, ValueError):
return ""
if days <= 0:
return ""
percent_str = f"{percent:.2f}".rstrip("0").rstrip(".")
skonto_date = berechne_faelligkeit(invoice_date_str, days)
if skonto_date:
return f"Bei Zahlung bis {skonto_date} gewähren wir {percent_str}% Skonto."
return f"Bei Zahlung innerhalb von {days} Tagen gewähren wir {percent_str}% Skonto."
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()]
# IT-Systemhaus-Erweiterung: zusaetzlich Monats-/Lizenz-/Personentag-Abrechnungseinheiten,
# wie sie bei Managed-Services-, Cloud-Abo- und Projekthonorar-Rechnungen ueblich sind.
unit_pattern = re.compile(
r"^(\d+(?:[\.,]\d+)?)\s*(Tag\(e\)|Std\.|Stk\.|Stunden|Stück|stk|stck|stück|"
r"Monat\(e\)|Monate|Personentage|Personentag|PT|Lizenz\(en\)|Lizenzen|Lizenz|Wartungspauschale)$",
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 "monat" in u: return "MON"
if "personentag" in u or u == "pt": return "DAY"
if "tag" in u or "day" in u: return "DAY"
if "std" in u or "stund" in u or "hour" in u or "hur" in u: return "HUR"
return "C62"
SERVICE_PERIOD_PATTERN = re.compile(
r"(?:Leistungszeitraum|Zeitraum|Einsatzzeitraum)\s*:?\s*"
r"(?:vom\s+)?(\d{1,2}[.\-]\d{1,2}[.\-]\d{4})\s*(?:bis|-|–)\s*(\d{1,2}[.\-]\d{1,2}[.\-]\d{4})",
re.IGNORECASE,
)
def extract_service_period(text):
"""Regex-Fallback fuer den Leistungszeitraum (z.B. 'Leistungszeitraum: 01.03.2026 bis
31.03.2026'), falls das LLM diesen nicht extrahiert hat. Gibt (start, end) als YYYY-MM-DD
zurueck, oder ("", "") wenn nichts gefunden wurde."""
match = SERVICE_PERIOD_PATTERN.search(text)
if not match:
return "", ""
start = parse_to_strict_xml_date(match.group(1).replace("-", "."))
end = parse_to_strict_xml_date(match.group(2).replace("-", "."))
return start, end
CONTRACT_TYPE_KEYWORDS = {
"Managed Services": re.compile(r"managed\s+services?", re.IGNORECASE),
"Cloud-Abonnement": re.compile(r"cloud[\s\-]?(abo(nnement)?|subscription)", re.IGNORECASE),
"Projekthonorar": re.compile(r"projekt(honorar|vergütung|pauschale)", re.IGNORECASE),
}
def detect_contract_type(text):
"""Erkennt den IT-Systemhaus-Vertragstyp anhand von Schluesselwoertern im Rechnungstext.
Gibt das erste gefundene Label zurueck oder "", wenn keines passt (keine Vermutung)."""
for label, pattern in CONTRACT_TYPE_KEYWORDS.items():
if pattern.search(text):
return label
return ""
SERIAL_NUMBER_PATTERN = re.compile(
r"(?:S/?N|Seriennummer|Serial(?:\s*No\.?)?)\s*:?\s*([A-Z0-9][A-Z0-9\-]{4,})",
re.IGNORECASE,
)
def extract_hardware_serial_numbers(text):
"""Findet Hardware-Seriennummern im Rechnungstext (Muster 'S/N: ...', 'Seriennummer: ...').
Gibt eine Liste der gefundenen Seriennummern in Text-Reihenfolge zurueck, ohne Duplikate."""
found = []
for match in SERIAL_NUMBER_PATTERN.finditer(text):
serial = match.group(1).strip()
if serial not in found:
found.append(serial)
return found
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("-", "")
def aggregate_tax_subtotals(items):
"""Gruppiert Rechnungspositionen nach Steuersatz (BG-23) und liefert je Gruppe Netto- und
Steuerbetrag - Grundlage fuer mehrere cac:TaxSubtotal/ram:ApplicableTradeTax-Bloecke bei
gemischten Steuersaetzen auf einer Rechnung."""
buckets = {}
for item in items:
rate = Decimal(str(item.get("vat_rate_percent", 0.0)))
net = Decimal(str(item["total_amount"]))
buckets[rate] = buckets.get(rate, Decimal("0")) + net
result = []
for rate in sorted(buckets.keys()):
taxable = buckets[rate].quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
tax_amount = (taxable * rate / Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
category = "S" if rate > 0 else "E"
result.append({
"rate": float(rate),
"category": category,
"taxable_amount": float(taxable),
"tax_amount": float(tax_amount),
})
return result
def compute_invoice_totals(items, discount_percent=0.0, discount_amount=0.0):
"""Berechnet Netto-/Steuer-/Bruttosummen inkl. optionalem dokumentenweiten Rabatt (BT-21).
Bei gemischten Steuersaetzen wird der Rabatt anteilig ueber die Steuersatz-Buckets verteilt
(Standardverfahren, da EN16931 keinen rabattspezifischen Steuersatz kennt). Ohne discount_*
ist das Ergebnis identisch zur reinen Summenbildung von vor Phase 1."""
line_net_total = sum((Decimal(str(item["total_amount"])) for item in items), Decimal("0"))
line_net_total = line_net_total.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
if discount_amount:
allowance = Decimal(str(discount_amount)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
elif discount_percent:
allowance = (line_net_total * Decimal(str(discount_percent)) / Decimal("100")).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP)
else:
allowance = Decimal("0.00")
raw_buckets = aggregate_tax_subtotals(items)
if allowance > 0 and line_net_total > 0:
buckets = []
allocated = Decimal("0")
for i, bucket in enumerate(raw_buckets):
taxable = Decimal(str(bucket["taxable_amount"]))
rate = Decimal(str(bucket["rate"]))
if i == len(raw_buckets) - 1:
share = allowance - allocated
else:
share = (allowance * taxable / line_net_total).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
allocated += share
new_taxable = (taxable - share).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
tax_amount = (new_taxable * rate / Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
buckets.append({
"rate": bucket["rate"],
"category": bucket["category"],
"taxable_amount": float(new_taxable),
"tax_amount": float(tax_amount),
"allowance_amount": float(share), # anteiliger Rabatt dieser Steuergruppe (BT-92 je Satz)
})
else:
buckets = raw_buckets
tax_basis_total = sum((Decimal(str(b["taxable_amount"])) for b in buckets), Decimal("0")).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP)
total_vat_amount = sum((Decimal(str(b["tax_amount"])) for b in buckets), Decimal("0")).quantize(
Decimal("0.01"), rounding=ROUND_HALF_UP)
total_gross_amount = (tax_basis_total + total_vat_amount).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
return {
"line_net_total": float(line_net_total),
"allowance_amount": float(allowance),
"tax_basis_total": float(tax_basis_total),
"total_vat_amount": float(total_vat_amount),
"total_gross_amount": float(total_gross_amount),
"tax_buckets": buckets,
}
# ==========================================
# 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: Standard bleibt 58 (SEPA-Ueberweisung), aber ueber
# PAYMENT_MEANS_CODES pro Rechnung waehlbar (siehe UN/EDIFACT 4461).
ET.SubElement(payment_means, f"{{{CBC}}}PaymentMeansCode").text = str(data.get("payment_means_code", "58"))
payee_acc = ET.SubElement(payment_means, f"{{{CAC}}}PayeeFinancialAccount")
ET.SubElement(payee_acc, f"{{{CBC}}}ID").text = str(data["iban"]).replace(" ", "")
# BT-20 Skonto-Vereinbarung als Freitext in cac:PaymentTerms/cbc:Note - EN16931 kennt kein
# eigenes strukturiertes Skonto-Feld. Muss laut UBL-Schema NACH PaymentMeans und VOR
# AllowanceCharge/TaxTotal stehen.
_skonto_text = format_skonto_text(data.get("skonto_percent"), data.get("skonto_days"), data["invoice_date"])
if _skonto_text:
payment_terms = ET.SubElement(root, f"{{{CAC}}}PaymentTerms")
ET.SubElement(payment_terms, f"{{{CBC}}}Note").text = _skonto_text
# BT-21 Dokumentenweiter Rabatt (BG-20). Muss laut UBL-Schema NACH PaymentTerms und VOR
# TaxTotal stehen. Voll EN16931-konform: bei gemischten Steuersaetzen wird JE Steuersatz ein
# eigener AllowanceCharge-Block mit dem anteilig zugeordneten Rabattbetrag erzeugt, damit jede
# Steuergruppe (BG-23) ihren eigenen Rabattanteil fuehrt (BR-CO-10/BR-DEC).
if data.get("allowance_amount", 0.0) > 0:
for bucket in data.get("tax_buckets") or []:
bucket_allowance = bucket.get("allowance_amount", 0.0)
if bucket_allowance <= 0:
continue
allowance_charge = ET.SubElement(root, f"{{{CAC}}}AllowanceCharge")
ET.SubElement(allowance_charge, f"{{{CBC}}}ChargeIndicator").text = "false"
ET.SubElement(allowance_charge, f"{{{CBC}}}AllowanceChargeReason").text = str(data.get("allowance_reason") or "Rabatt")
ET.SubElement(allowance_charge, f"{{{CBC}}}Amount", {"currencyID": "EUR"}).text = f"{bucket_allowance:.2f}"
ac_tax_cat = ET.SubElement(allowance_charge, f"{{{CAC}}}TaxCategory")
ET.SubElement(ac_tax_cat, f"{{{CBC}}}ID").text = bucket["category"]
ET.SubElement(ac_tax_cat, f"{{{CBC}}}Percent").text = f"{bucket['rate']:.2f}"
ac_tax_scheme = ET.SubElement(ac_tax_cat, f"{{{CAC}}}TaxScheme")
ET.SubElement(ac_tax_scheme, f"{{{CBC}}}ID").text = "VAT"
# --- 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}"
# BG-23: pro Steuersatz-Bucket ein eigenes cac:TaxSubtotal (Mehrfach-Steuersaetze).
# Fallback fuer Aufrufer ohne tax_buckets (Einzelsteuersatz, byte-identisch zu vor Phase 1):
# synthetischer Ein-Bucket aus dem globalen Steuersatz und den Gesamtsummen.
_tax_buckets = data.get("tax_buckets") or [{
"rate": vat_rate, "category": vat_category,
"taxable_amount": data["total_net_amount"], "tax_amount": data["total_vat_amount"],
}]
for bucket in _tax_buckets:
tax_subtotal = ET.SubElement(tax_total, f"{{{CAC}}}TaxSubtotal")
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxableAmount", {"currencyID": "EUR"}).text = f"{bucket['taxable_amount']:.2f}"
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{bucket['tax_amount']:.2f}"
tax_cat = ET.SubElement(tax_subtotal, f"{{{CAC}}}TaxCategory")
ET.SubElement(tax_cat, f"{{{CBC}}}ID").text = bucket["category"]
ET.SubElement(tax_cat, f"{{{CBC}}}Percent").text = f"{bucket['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")
# BT-106 LineExtensionAmount ist die UNRABATTIERTE Positionssumme; BT-109 TaxExclusiveAmount
# ("total_net_amount") ist bereits die um einen etwaigen dokumentenweiten Rabatt (BT-107)
# reduzierte Steuerbasis - siehe compute_invoice_totals(). Ohne Rabatt sind beide identisch.
ET.SubElement(monetary_total, f"{{{CBC}}}LineExtensionAmount", {"currencyID": "EUR"}).text = f"{data.get('line_net_total', 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}"
if data.get("allowance_amount", 0.0) > 0:
ET.SubElement(monetary_total, f"{{{CBC}}}AllowanceTotalAmount", {"currencyID": "EUR"}).text = f"{data['allowance_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"))
# BG-30 Positionssteuer: nutzt den eigenen Steuersatz der Position, falls vorhanden
# (Multi-Tax-Support) - fällt sonst auf den globalen Rechnungssteuersatz zurück.
line_vat_rate = float(item.get("vat_rate_percent", data['vat_rate_percent']))
line_vat_category = "S" if line_vat_rate > 0.0 else "E"
item_tax = ET.SubElement(item_elem, f"{{{CAC}}}ClassifiedTaxCategory")
ET.SubElement(item_tax, f"{{{CBC}}}ID").text = line_vat_category
ET.SubElement(item_tax, f"{{{CBC}}}Percent").text = f"{line_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_vat_rate = float(item.get("vat_rate_percent", data['vat_rate_percent']))
line_vat_category = "S" if line_vat_rate > 0.0 else "E"
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 = line_vat_category
ET.SubElement(trade_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{line_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: Standard bleibt 58 (SEPA-Ueberweisung), aber ueber
# PAYMENT_MEANS_CODES pro Rechnung waehlbar (siehe UN/EDIFACT 4461).
ET.SubElement(pay_means, f"{{{RAM}}}TypeCode").text = str(data.get("payment_means_code", "58"))
payee_acc = ET.SubElement(pay_means, f"{{{RAM}}}PayeePartyCreditorFinancialAccount")
ET.SubElement(payee_acc, f"{{{RAM}}}IBANID").text = str(data["iban"]).replace(" ", "")
# BG-23: pro Steuersatz-Bucket ein eigenes ram:ApplicableTradeTax (Mehrfach-Steuersaetze).
# Fallback fuer Aufrufer ohne tax_buckets (Einzelsteuersatz, byte-identisch zu vor Phase 1):
# synthetischer Ein-Bucket aus dem globalen Steuersatz und den Gesamtsummen.
_tax_buckets = data.get("tax_buckets") or [{
"rate": vat_rate, "category": vat_category,
"taxable_amount": data["total_net_amount"], "tax_amount": data["total_vat_amount"],
}]
for bucket in _tax_buckets:
header_tax = ET.SubElement(header_settle, f"{{{RAM}}}ApplicableTradeTax")
ET.SubElement(header_tax, f"{{{RAM}}}CalculatedAmount").text = f"{bucket['tax_amount']:.2f}"
ET.SubElement(header_tax, f"{{{RAM}}}TypeCode").text = "VAT"
ET.SubElement(header_tax, f"{{{RAM}}}BasisAmount").text = f"{bucket['taxable_amount']:.2f}"
ET.SubElement(header_tax, f"{{{RAM}}}CategoryCode").text = bucket["category"]
ET.SubElement(header_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{bucket['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-21 Dokumentenweiter Rabatt (BG-20). Muss laut CII-Schema NACH BillingSpecifiedPeriod und
# VOR SpecifiedTradePaymentTerms stehen. Voll EN16931-konform: bei gemischten Steuersaetzen wird
# JE Steuersatz ein eigener SpecifiedTradeAllowanceCharge-Block mit dem anteilig zugeordneten
# Rabattbetrag erzeugt (jede Steuergruppe fuehrt ihren eigenen Rabattanteil).
if data.get("allowance_amount", 0.0) > 0:
for bucket in data.get("tax_buckets") or []:
bucket_allowance = bucket.get("allowance_amount", 0.0)
if bucket_allowance <= 0:
continue
allowance_charge = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradeAllowanceCharge")
indicator = ET.SubElement(allowance_charge, f"{{{RAM}}}ChargeIndicator")
ET.SubElement(indicator, f"{{{UDT}}}Indicator").text = "false"
ET.SubElement(allowance_charge, f"{{{RAM}}}ActualAmount").text = f"{bucket_allowance:.2f}"
ET.SubElement(allowance_charge, f"{{{RAM}}}Reason").text = str(data.get("allowance_reason") or "Rabatt")
ac_tax = ET.SubElement(allowance_charge, f"{{{RAM}}}CategoryTradeTax")
ET.SubElement(ac_tax, f"{{{RAM}}}TypeCode").text = "VAT"
ET.SubElement(ac_tax, f"{{{RAM}}}CategoryCode").text = bucket["category"]
ET.SubElement(ac_tax, f"{{{RAM}}}RateApplicablePercent").text = f"{bucket['rate']:.2f}"
# BT-9 Fälligkeitsdatum, abgeleitet aus dem tatsächlichen Zahlungsziel der Rechnung, sowie
# optional die BT-20 Skonto-Freitextbeschreibung. ram:Description muss laut Schema VOR
# ram:DueDateDateTime stehen. Kein fester Fallback: ohne gültiges Zahlungsziel UND ohne
# Skonto wird der Block weggelassen (die UI blockiert dann den Export bzgl. Zahlungsziel).
_cii_due = berechne_faelligkeit(data["invoice_date"], data.get("payment_term_days"))
_skonto_text = format_skonto_text(data.get("skonto_percent"), data.get("skonto_days"), data["invoice_date"])
if _cii_due or _skonto_text:
payment_terms = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradePaymentTerms")
if _skonto_text:
ET.SubElement(payment_terms, f"{{{RAM}}}Description").text = _skonto_text
if _cii_due:
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.get('line_net_total', data['total_net_amount']):.2f}"
if data.get("allowance_amount", 0.0) > 0:
ET.SubElement(header_sum, f"{{{RAM}}}AllowanceTotalAmount").text = f"{data['allowance_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(
"<fx:ConformanceLevel>EN 16931</fx:ConformanceLevel>",
"<fx:ConformanceLevel>XRECHNUNG</fx:ConformanceLevel>",
).replace(
"<fx:Version>1.0</fx:Version>",
"<fx:Version>3.0</fx:Version>",
)
pdf.Root.Metadata.write(xmp_str.encode("utf-8"))
buf = io.BytesIO()
pdf.save(buf)
return buf.getvalue()
finally:
pdf.close()

708
main.py
View File

@ -3,30 +3,15 @@ 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,
PAYMENT_MEANS_CODES,
berechne_faelligkeit,
kaufmaennisches_runden,
clean_german_float,
extract_invoice_items_pure_python,
map_unece_unit_code,
parse_to_strict_xml_date,
format_cii_date,
generate_compliant_xrechnung,
generate_compliant_zugferd_cii,
add_pdfa_output_intent,
fix_xrechnung_xmp_conformance,
compute_invoice_totals,
extract_service_period,
detect_contract_type,
extract_hardware_serial_numbers,
)
# ==========================================
# 0. SITZUNGS-STATUS INITIALISIEREN (ZÄHLER GEGEN WIDGET-CACHING)
@ -77,6 +62,610 @@ 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(
"<fx:ConformanceLevel>EN 16931</fx:ConformanceLevel>",
"<fx:ConformanceLevel>XRECHNUNG</fx:ConformanceLevel>",
).replace(
"<fx:Version>1.0</fx:Version>",
"<fx:Version>3.0</fx:Version>",
)
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
@ -179,17 +768,6 @@ with col2:
final_data["items"] = extract_invoice_items_pure_python(pdf_text)
# IT-Systemhaus-Regex-Fallbacks: nur ergaenzen, wenn das LLM die entsprechenden
# Felder nicht bereits geliefert hat.
if not final_data.get("service_start_date") and not final_data.get("service_end_date"):
fallback_start, fallback_end = extract_service_period(pdf_text)
if fallback_start and fallback_end:
final_data["service_start_date"] = fallback_start
final_data["service_end_date"] = fallback_end
final_data["contract_type"] = detect_contract_type(pdf_text)
final_data["hardware_serials"] = extract_hardware_serial_numbers(pdf_text)
st.session_state.invoice_counter += 1
st.session_state.extracted_data = final_data
@ -197,10 +775,6 @@ with col2:
data = st.session_state.extracted_data
st.caption("Mit :red[*] markierte Felder sind EN16931-Pflichtfelder und müssen für einen gültigen Export befüllt sein.")
if data.get("contract_type"):
st.caption(f"🏷️ Erkannter Vertragstyp: **{data['contract_type']}**")
if data.get("hardware_serials"):
st.caption(f"🔧 Erkannte Seriennummern: {', '.join(data['hardware_serials'])}")
# --- BT-3 RECHNUNGSTYP DROPDOWN-MENÜ ---
inv_type_ui = st.selectbox(
@ -271,24 +845,10 @@ with col2:
r_zip = st.text_input("PLZ (BT-53)", value=data.get("recipient_zip", ""))
r_city = st.text_input("Stadt (BT-52) :red[*]", value=data.get("recipient_city", ""))
st.markdown("---")
payment_means_codes_list = list(PAYMENT_MEANS_CODES.keys())
payment_means_labels = [f"{code} - {PAYMENT_MEANS_CODES[code]}" for code in payment_means_codes_list]
selected_payment_means_label = st.selectbox(
"Zahlungsmittel (BT-81)",
options=payment_means_labels,
index=payment_means_codes_list.index("58"),
)
selected_payment_means_code = selected_payment_means_label.split(" - ")[0]
st.markdown("### 3. Positionen")
if st.button("➕ Neue Position hinzufügen"):
data["items"].append({
"name": "Neue Dienstleistung", "description": "", "quantity": 1.0, "unit": "Stk.",
"unit_price": 0.0, "vat_rate_percent": float(data.get("vat_rate_percent") or 19.0),
})
data["items"].append({"name": "Neue Dienstleistung", "description": "", "quantity": 1.0, "unit": "Stk.", "unit_price": 0.0})
st.rerun()
edited_items = []
@ -312,19 +872,13 @@ with col2:
calc_height = max(68, min(400, estimated_lines * 24))
item_description = st.text_area("Zusatzbeschreibung (BT-154)", value=desc_text, key=f"item_desc_{inv_id}_{idx}", height=calc_height)
c_qty, c_unit, c_price, c_vat, c_total = st.columns(5)
c_qty, c_unit, c_price, c_total = st.columns(4)
with c_qty:
item_quantity = st.number_input("Menge (BT-129) :red[*]", value=float(item.get("quantity") or 0.0), format="%.3f", step=0.1, key=f"item_qty_{inv_id}_{idx}")
with c_unit:
item_unit = st.text_input("Einheit (BT-130) :red[*]", value=item.get("unit", "Stk."), key=f"item_unit_{inv_id}_{idx}")
with c_price:
item_price = st.number_input("Einzelpreis (BT-146, €) :red[*]", value=float(item.get("unit_price") or 0.0), format="%.2f", key=f"item_price_{inv_id}_{idx}")
with c_vat:
item_vat_rate = st.number_input(
"Steuersatz (BT-152, %) :red[*]",
value=float(item.get("vat_rate_percent") or data.get("vat_rate_percent") or 19.0),
key=f"item_vat_{inv_id}_{idx}",
)
with c_total:
raw_item_total = item_quantity * item_price
item_total_amount = kaufmaennisches_runden(raw_item_total)
@ -337,30 +891,21 @@ with col2:
"quantity": item_quantity,
"unit": item_unit,
"unit_price": item_price,
"total_amount": item_total_amount,
"vat_rate_percent": item_vat_rate,
"total_amount": item_total_amount
})
st.markdown("---")
# Der frühere globale Steuersatz dient jetzt nur noch als Vorbelegung für neu
# hinzugefügte Positionen (siehe "Neue Position hinzufügen" oben) - abgerechnet wird
# pro Position mit dem jeweils eingetragenen Steuersatz (Multi-Tax-Support).
vat_rate = st.number_input("Standard-Steuersatz für neue Positionen (%)", value=float(data.get("vat_rate_percent") or 19.0))
vat_rate = st.number_input("Steuersatz (BT-152, %) :red[*]", value=float(data.get("vat_rate_percent") or 19.0))
c_disc1, c_disc2, c_disc3, c_disc4 = st.columns(4)
with c_disc1:
discount_percent = st.number_input("Rabatt (BT-21, %)", value=0.0, min_value=0.0, max_value=100.0, step=0.5)
with c_disc2:
discount_reason = st.text_input("Rabattgrund", value="")
with c_disc3:
skonto_percent = st.number_input("Skonto (BT-20, %)", value=0.0, min_value=0.0, max_value=100.0, step=0.5)
with c_disc4:
skonto_days = st.text_input("Skonto-Frist (Tage)", value="")
totals = compute_invoice_totals(edited_items, discount_percent=discount_percent)
computed_net_amount = totals["tax_basis_total"]
computed_vat_amount = totals["total_vat_amount"]
computed_gross_amount = totals["total_gross_amount"]
# Summen mit Decimal bilden, damit die Netto-Summe exakt der Summe der (bereits gerundeten)
# Positionsbeträge entspricht. Reine float-Summation kann minimal driften und dann zu
# BR-CO-10-Validierungsfehlern führen (Summe der Zeilen != TaxExclusiveAmount).
net_dec = sum((Decimal(str(item["total_amount"])) for item in edited_items), Decimal("0"))
vat_dec = (net_dec * Decimal(str(vat_rate)) / Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
gross_dec = (net_dec + vat_dec).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
computed_net_amount = float(net_dec.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
computed_vat_amount = float(vat_dec)
computed_gross_amount = float(gross_dec)
st.markdown("**Berechnete Summen:**")
c_net_ui, c_vat_ui, c_gross_ui = st.columns(3)
@ -383,13 +928,6 @@ with col2:
"recipient_street": r_street, "recipient_zip": r_zip, "recipient_city": r_city,
"items": edited_items, "total_net_amount": computed_net_amount, "total_vat_amount": computed_vat_amount,
"vat_rate_percent": vat_rate, "total_gross_amount": computed_gross_amount, "iban": VERKAEUFER_IBAN,
"payment_means_code": selected_payment_means_code,
"tax_buckets": totals["tax_buckets"],
"line_net_total": totals["line_net_total"],
"allowance_amount": totals["allowance_amount"],
"allowance_reason": discount_reason,
"skonto_percent": skonto_percent,
"skonto_days": skonto_days,
"delivery_note_id": dn_id,
"order_id": o_id,
"contract_id": c_id,

View File

@ -1,2 +0,0 @@
[pytest]
pythonpath = .

View File

@ -1 +0,0 @@
pytest==8.3.4

View File

@ -1,407 +0,0 @@
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,
PAYMENT_MEANS_CODES,
aggregate_tax_subtotals,
compute_invoice_totals,
format_skonto_text,
extract_service_period,
detect_contract_type,
extract_hardware_serial_numbers,
)
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_extract_invoice_items_pure_python_recognizes_person_days():
text = "1\nProjektarbeit\n5,00 Personentage\n800,00\n4000,00\n"
items = extract_invoice_items_pure_python(text)
assert items[0]["quantity"] == 5.0
assert items[0]["unit"] == "Personentage"
def test_extract_invoice_items_pure_python_recognizes_license_months():
text = "1\nCloud-Lizenz\n3,00 Monat(e)\n49,00\n147,00\n"
items = extract_invoice_items_pure_python(text)
assert items[0]["unit"] == "Monat(e)"
def test_map_unece_unit_code_extended_units():
assert map_unece_unit_code("Monat(e)") == "MON"
assert map_unece_unit_code("Personentage") == "DAY"
assert map_unece_unit_code("Lizenz(en)") == "C62"
def test_parse_to_strict_xml_date_german_format():
assert parse_to_strict_xml_date("15.01.2026") == "2026-01-15"
def test_payment_means_codes_contains_expected_entries():
assert PAYMENT_MEANS_CODES["58"] == "SEPA-Überweisung"
assert set(["30", "31", "42", "48", "58"]).issubset(PAYMENT_MEANS_CODES.keys())
def test_generate_compliant_xrechnung_uses_dynamic_payment_means_code():
data = _single_rate_invoice()
data["payment_means_code"] = "31"
xml_str = generate_compliant_xrechnung(data)
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"}
code = root.find(".//cac:PaymentMeans/cbc:PaymentMeansCode", ns).text
assert code == "31"
def test_generate_compliant_xrechnung_defaults_payment_means_code_to_58():
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"}
code = root.find(".//cac:PaymentMeans/cbc:PaymentMeansCode", ns).text
assert code == "58"
def test_generate_compliant_zugferd_cii_uses_dynamic_payment_means_code():
data = _single_rate_invoice()
data["payment_means_code"] = "48"
xml_str = generate_compliant_zugferd_cii(data)
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
code = root.find(".//ram:SpecifiedTradeSettlementPaymentMeans/ram:TypeCode", ns).text
assert code == "48"
def test_aggregate_tax_subtotals_single_rate():
items = [{"total_amount": 200.0, "vat_rate_percent": 19.0}]
buckets = aggregate_tax_subtotals(items)
assert buckets == [{"rate": 19.0, "category": "S", "taxable_amount": 200.0, "tax_amount": 38.0}]
def test_aggregate_tax_subtotals_mixed_rates():
items = [
{"total_amount": 100.0, "vat_rate_percent": 19.0},
{"total_amount": 50.0, "vat_rate_percent": 7.0},
{"total_amount": 30.0, "vat_rate_percent": 0.0},
]
buckets = aggregate_tax_subtotals(items)
assert buckets == [
{"rate": 0.0, "category": "E", "taxable_amount": 30.0, "tax_amount": 0.0},
{"rate": 7.0, "category": "S", "taxable_amount": 50.0, "tax_amount": 3.5},
{"rate": 19.0, "category": "S", "taxable_amount": 100.0, "tax_amount": 19.0},
]
def test_compute_invoice_totals_mixed_rates_no_discount():
items = [
{"total_amount": 100.0, "vat_rate_percent": 19.0},
{"total_amount": 50.0, "vat_rate_percent": 7.0},
]
totals = compute_invoice_totals(items)
assert totals["line_net_total"] == 150.0
assert totals["allowance_amount"] == 0.0
assert totals["tax_basis_total"] == 150.0
assert totals["total_vat_amount"] == 22.5
assert totals["total_gross_amount"] == 172.5
def test_compute_invoice_totals_single_rate_matches_old_behavior():
items = [{"total_amount": 200.0, "vat_rate_percent": 19.0}]
totals = compute_invoice_totals(items)
assert totals["tax_basis_total"] == 200.0
assert totals["total_vat_amount"] == 38.0
assert totals["total_gross_amount"] == 238.0
def _mixed_rate_invoice():
data = _single_rate_invoice()
items = [
{"name": "Beratung", "description": "", "quantity": 2.0, "unit": "Std.",
"unit_price": 100.0, "total_amount": 200.0, "vat_rate_percent": 19.0},
{"name": "Buch", "description": "", "quantity": 1.0, "unit": "Stk.",
"unit_price": 50.0, "total_amount": 50.0, "vat_rate_percent": 7.0},
]
totals = compute_invoice_totals(items)
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
return data
def test_generate_compliant_xrechnung_mixed_rates_two_tax_subtotals():
xml_str = generate_compliant_xrechnung(_mixed_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"}
subtotals = root.findall(".//cac:TaxTotal/cac:TaxSubtotal", ns)
assert len(subtotals) == 2
percents = sorted(st.find("cac:TaxCategory/cbc:Percent", ns).text for st in subtotals)
assert percents == ["19.00", "7.00"] or percents == ["7.00", "19.00"]
lines = root.findall(".//cac:InvoiceLine", ns)
line_percents = [ln.find(".//cbc:Percent", ns).text for ln in lines]
assert set(line_percents) == {"19.00", "7.00"}
def test_generate_compliant_zugferd_cii_mixed_rates_two_header_taxes():
xml_str = generate_compliant_zugferd_cii(_mixed_rate_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
header_taxes = root.findall(
".//ram:ApplicableHeaderTradeSettlement/ram:ApplicableTradeTax", ns)
assert len(header_taxes) == 2
def test_format_skonto_text_with_valid_percent_and_days():
text = format_skonto_text(2.0, 10, "2026-01-15")
assert "2026-01-25" in text
assert "2%" in text
def test_format_skonto_text_returns_empty_when_no_percent():
assert format_skonto_text(0, 10, "2026-01-15") == ""
assert format_skonto_text(None, 10, "2026-01-15") == ""
def test_format_skonto_text_returns_empty_when_no_days():
assert format_skonto_text(2.0, "", "2026-01-15") == ""
assert format_skonto_text(2.0, None, "2026-01-15") == ""
def _discounted_invoice():
data = _single_rate_invoice()
items = [{"total_amount": 200.0, "vat_rate_percent": 19.0, "name": "Beratung",
"description": "", "quantity": 2.0, "unit": "Std.", "unit_price": 100.0}]
totals = compute_invoice_totals(items, discount_percent=10.0)
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["allowance_reason"] = "Treuerabatt"
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
data["skonto_percent"] = 2.0
data["skonto_days"] = "10"
return data
def test_generate_compliant_xrechnung_includes_allowance_charge_and_payment_terms():
xml_str = generate_compliant_xrechnung(_discounted_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"}
allowance = root.find(".//cac:AllowanceCharge", ns)
assert allowance.find("cbc:AllowanceChargeReason", ns).text == "Treuerabatt"
assert allowance.find("cbc:Amount", ns).text == "20.00"
payment_terms_note = root.find(".//cac:PaymentTerms/cbc:Note", ns).text
assert "Skonto" in payment_terms_note
def test_generate_compliant_zugferd_cii_includes_allowance_charge_and_payment_terms_description():
xml_str = generate_compliant_zugferd_cii(_discounted_invoice())
root = ET.fromstring(xml_str.encode("utf-8"))
ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
allowance = root.find(".//ram:SpecifiedTradeAllowanceCharge", ns)
assert allowance.find("ram:ActualAmount", ns).text == "20.00"
assert allowance.find("ram:Reason", ns).text == "Treuerabatt"
description = root.find(".//ram:SpecifiedTradePaymentTerms/ram:Description", ns).text
assert "Skonto" in description
def test_generate_compliant_xrechnung_mixed_rate_discount_emits_allowance_per_rate():
"""Voll EN16931-konform: ein Rabatt auf eine Rechnung mit 19%- UND 7%-Positionen erzeugt
JE Steuersatz einen eigenen AllowanceCharge-Block mit anteiligem Rabattbetrag."""
items = [
{"total_amount": 100.0, "vat_rate_percent": 19.0, "name": "A", "description": "",
"quantity": 1.0, "unit": "Stk.", "unit_price": 100.0},
{"total_amount": 100.0, "vat_rate_percent": 7.0, "name": "B", "description": "",
"quantity": 1.0, "unit": "Stk.", "unit_price": 100.0},
]
totals = compute_invoice_totals(items, discount_percent=10.0)
data = _single_rate_invoice()
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["allowance_reason"] = "Aktionsrabatt"
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
xml_str = generate_compliant_xrechnung(data)
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"}
allowances = root.findall(".//cac:AllowanceCharge", ns)
assert len(allowances) == 2
percents = sorted(a.find("cac:TaxCategory/cbc:Percent", ns).text for a in allowances)
assert percents == ["19.00", "7.00"] or percents == ["7.00", "19.00"]
total_alloc = sum(float(a.find("cbc:Amount", ns).text) for a in allowances)
assert abs(total_alloc - 20.0) < 0.001
def test_extract_service_period_finds_range():
text = "Managed Services\nLeistungszeitraum: 01.03.2026 bis 31.03.2026\n1\n..."
start, end = extract_service_period(text)
assert start == "2026-03-01"
assert end == "2026-03-31"
def test_extract_service_period_returns_empty_when_absent():
assert extract_service_period("Keine Zeitangabe hier.") == ("", "")
def test_detect_contract_type_managed_services():
assert detect_contract_type("Wartungsvertrag Managed Services Tier 2") == "Managed Services"
def test_detect_contract_type_cloud_abo():
assert detect_contract_type("Ihr Cloud-Abonnement für 2026") == "Cloud-Abonnement"
def test_detect_contract_type_returns_empty_when_no_match():
assert detect_contract_type("Ganz normale Rechnung ohne Vertragstyp") == ""
def test_extract_hardware_serial_numbers_finds_sn_pattern():
text = "Server Dell R750\nS/N: ABCD-1234-XYZ\nSwitch Cisco\nSeriennummer: SN9988771"
serials = extract_hardware_serial_numbers(text)
assert "ABCD-1234-XYZ" in serials
assert "SN9988771" in serials
def test_extract_hardware_serial_numbers_empty_when_absent():
assert extract_hardware_serial_numbers("Keine Seriennummern hier.") == []
def test_full_invoice_with_all_phase1_features_combined():
"""Realistische IT-Systemhaus-Rechnung: gemischte Steuersaetze, Rabatt, Skonto,
dynamischer Zahlungsmittelcode und IT-spezifische Abrechnungseinheiten - alles zusammen."""
items = [
{"name": "Managed-Services-Pauschale", "description": "", "quantity": 1.0, "unit": "Monat(e)",
"unit_price": 500.0, "total_amount": 500.0, "vat_rate_percent": 19.0},
{"name": "Fachliteratur", "description": "", "quantity": 1.0, "unit": "Stk.",
"unit_price": 40.0, "total_amount": 40.0, "vat_rate_percent": 7.0},
]
totals = compute_invoice_totals(items, discount_percent=5.0)
data = _single_rate_invoice()
data["items"] = items
data["tax_buckets"] = totals["tax_buckets"]
data["line_net_total"] = totals["line_net_total"]
data["allowance_amount"] = totals["allowance_amount"]
data["allowance_reason"] = "Neukundenrabatt"
data["total_net_amount"] = totals["tax_basis_total"]
data["total_vat_amount"] = totals["total_vat_amount"]
data["total_gross_amount"] = totals["total_gross_amount"]
data["skonto_percent"] = 2.0
data["skonto_days"] = "14"
data["payment_means_code"] = "31"
ubl_xml = generate_compliant_xrechnung(data)
ubl_root = ET.fromstring(ubl_xml.encode("utf-8"))
ubl_ns = {"cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2",
"cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"}
assert len(ubl_root.findall(".//cac:TaxTotal/cac:TaxSubtotal", ubl_ns)) == 2
assert ubl_root.find(".//cac:PaymentMeans/cbc:PaymentMeansCode", ubl_ns).text == "31"
ubl_allowances = ubl_root.findall(".//cac:AllowanceCharge", ubl_ns)
assert len(ubl_allowances) == 2 # ein Rabatt-Block je Steuersatz (19% und 7%)
assert abs(sum(float(a.find("cbc:Amount", ubl_ns).text) for a in ubl_allowances) - 27.0) < 0.001
assert "Skonto" in ubl_root.find(".//cac:PaymentTerms/cbc:Note", ubl_ns).text
cii_xml = generate_compliant_zugferd_cii(data)
cii_root = ET.fromstring(cii_xml.encode("utf-8"))
cii_ns = {"ram": "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"}
assert len(cii_root.findall(".//ram:ApplicableHeaderTradeSettlement/ram:ApplicableTradeTax", cii_ns)) == 2
assert cii_root.find(".//ram:SpecifiedTradeSettlementPaymentMeans/ram:TypeCode", cii_ns).text == "31"
cii_allowances = cii_root.findall(".//ram:SpecifiedTradeAllowanceCharge", cii_ns)
assert len(cii_allowances) == 2 # ein Rabatt-Block je Steuersatz (19% und 7%)
assert abs(sum(float(a.find("ram:ActualAmount", cii_ns).text) for a in cii_allowances) - 27.0) < 0.001