1588 lines
86 KiB
Python
1588 lines
86 KiB
Python
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. R00-000)")
|
||
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 (BT-46), unter der der Kunde beim Verkäufer geführt wird (z.B. der exakte Wert nach 'Kunden-Nr:'). Falls im Text keine Kundennummer erkennbar ist: leer lassen.")
|
||
leitweg_id: str = Field(default="", description="Leitweg-ID (BT-10) - NUR bei Rechnungen an Behörden (B2G), z.B. '04011000-1234-56'. Ist im Text keine Leitweg-ID erkennbar: leer lassen (NICHT die Kundennummer hier eintragen).")
|
||
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.Mustermann). 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. Musterfirma 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="Vertragsnummer (BT-12) VOR Position 1 - NUR ein echter Vertrag. NICHT ein Vergabeangebot/eine Ausschreibung (die gehören in tender_id) und NICHT ein 'Angebot' (z.B. 'ANGEBOT AN26-174 vom ...' - das wird separat als Freitext behandelt, hier KEIN Wert eintragen). Sonst leer lassen.")
|
||
tender_id: str = Field(default="", description="Nummer eines Vergabeangebots oder einer Ausschreibung (BT-17, z.B. AN26-130) VOR Position 1. NICHT die Vertragsnummer (die gehört in contract_id) und NICHT ein bloßes 'Angebot ... vom ...' (das ist KEIN Vergabeangebot). 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)
|
||
"bic": "DEUTDEDB160", # BT-86 (BIC der Hausbank, Deutsche Bank)
|
||
}
|
||
|
||
# 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)",
|
||
}
|
||
|
||
# Schema-Codes fuer die elektronische Adresse des Kaeufers (BT-49-1, Code-Liste EAS). "EM" (E-Mail)
|
||
# ist der Standard fuer den B2B-Versand per Mail; die weiteren gelten v.a. beim Peppol-Versand an
|
||
# Behoerden (B2G). Die Leitweg-ID hat ihr eigentliches "Zuhause" in BT-10, kann als elektronische
|
||
# Adresse aber ueber Schema 0204 zugestellt werden.
|
||
BUYER_ENDPOINT_SCHEMES = {
|
||
"EM": "E-Mail-Adresse (Standard, B2B)",
|
||
"0204": "Leitweg-ID (deutsche Behörden / B2G)",
|
||
"0088": "GLN (Global Location Number)",
|
||
}
|
||
|
||
|
||
def resolve_buyer_endpoint(recipient_email, recipient_scheme, leitweg_id):
|
||
"""Ermittelt die effektive elektronische Adresse des Kaeufers (BT-49) samt Schema (BT-49-1).
|
||
|
||
Zuerst wird der Wert bestimmt: das BT-49-Feld hat Vorrang; ist es leer, wird die Leitweg-ID
|
||
(BT-10) genommen (so muss sie nur einmal eingegeben werden). Dann das Schema:
|
||
1. Enthaelt der Wert ein '@', ist es immer eine E-Mail -> Schema 'EM' (egal aus welchem Feld
|
||
er stammt und unabhaengig vom Dropdown - ein falsch gewaehltes Schema schlaegt nicht durch).
|
||
2. Stammt der Wert aus der Leitweg-ID (BT-49-Feld war leer) -> Schema '0204'.
|
||
3. Sonst (Wert direkt in BT-49 eingetragen, ohne '@', z.B. GLN) -> das gewaehlte Schema.
|
||
Ist beides leer, bleibt die Adresse leer (Schema unveraendert). Gibt (adresse, schema) zurueck."""
|
||
email = str(recipient_email or "").strip()
|
||
if email:
|
||
value, from_leitweg = email, False
|
||
else:
|
||
value, from_leitweg = str(leitweg_id or "").strip(), True
|
||
|
||
if not value:
|
||
return "", recipient_scheme
|
||
if "@" in value:
|
||
return value, "EM"
|
||
if from_leitweg:
|
||
return value, "0204"
|
||
return value, recipient_scheme
|
||
|
||
# Fester Hinweistext fuer Vorkasse. Vorkasse ist eine Zahlungs*bedingung* (BT-20), kein
|
||
# Zahlungs*mittel* - das Zahlungsmittel bleibt die Ueberweisung. Der Text wird daher zu den
|
||
# Zahlungsbedingungen (BT-20) gelegt, nicht in den Zahlungsmittelcode (BT-81).
|
||
VORKASSE_HINWEIS = "Zahlung erfolgt per Vorkasse vor Lieferung."
|
||
|
||
|
||
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 format_payment_term_text(payment_term_days):
|
||
"""Klartext des Zahlungsziels fuer BT-20 ('14' -> 'Zahlbar innerhalb von 14 Tagen ohne
|
||
Abzug.'). Das Faelligkeitsdatum steht strukturiert in BT-9; der Klartext daneben ist bei
|
||
Lieferanten die Regel (4 von 5 Eingangsrechnungen liefern ihn) und fuer den Empfaenger
|
||
lesbar. Gibt "" zurueck, wenn kein gueltiges Zahlungsziel angegeben ist - es wird NICHTS
|
||
geraten."""
|
||
try:
|
||
days = int(str(payment_term_days).strip())
|
||
except (TypeError, ValueError):
|
||
return ""
|
||
if days <= 0:
|
||
return ""
|
||
return f"Zahlbar innerhalb von {days} Tagen ohne Abzug."
|
||
|
||
|
||
def resolve_payment_reference(data):
|
||
"""Zahlungsreferenz/Verwendungszweck (BT-83). Standard ist die Rechnungsnummer - so machen es
|
||
auch die Lieferanten, die das Feld ueberhaupt fuellen. Ist aber eine eigene Referenz
|
||
angegeben (z.B. eine SAP-Referenz, die die Buchhaltung des Empfaengers zur Zuordnung
|
||
verlangt), gewinnt diese. Ein leeres Feld bedeutet 'Standard nehmen', nicht 'leer lassen' -
|
||
ganz weglassen laesst sich BT-83 nur, wenn auch keine Rechnungsnummer vorliegt."""
|
||
explicit = str(data.get("payment_reference") or "").strip()
|
||
if explicit:
|
||
return explicit
|
||
return str(data.get("invoice_number") or "").strip()
|
||
|
||
|
||
def build_payment_terms_note(data):
|
||
"""Kombinierter Freitext fuer die Zahlungsbedingungen (BT-20): das Zahlungsziel im Klartext
|
||
ODER ein Vorkasse-Hinweis, jeweils optional ergaenzt um eine Skonto-Vereinbarung. Fuer all
|
||
das kennt EN16931 kein eigenes strukturiertes Feld. Gibt "" zurueck, wenn nichts davon
|
||
gesetzt ist.
|
||
|
||
Vorkasse und Zahlungsziel schliessen sich bewusst aus: 'Zahlung per Vorkasse vor Lieferung'
|
||
und 'zahlbar innerhalb von 14 Tagen' wuerden sich widersprechen."""
|
||
parts = []
|
||
if data.get("vorkasse"):
|
||
parts.append(VORKASSE_HINWEIS)
|
||
else:
|
||
term_text = format_payment_term_text(data.get("payment_term_days"))
|
||
if term_text:
|
||
parts.append(term_text)
|
||
skonto_text = format_skonto_text(
|
||
data.get("skonto_percent"), data.get("skonto_days"), data["invoice_date"]
|
||
)
|
||
if skonto_text:
|
||
parts.append(skonto_text)
|
||
return " ".join(parts)
|
||
|
||
|
||
ANGEBOT_PATTERN = re.compile(
|
||
r"\bAngebot\s+([A-Za-z0-9][A-Za-z0-9\-/.]*)\s+vom\s+(\d{1,2}[.\-/]\d{1,2}[.\-/]\d{2,4})",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def extract_angebot_reference(text):
|
||
"""Findet einen Angebotsbezug der Form 'Angebot <Nummer> vom <Datum>' im Rechnungstext.
|
||
EN16931 kennt kein eigenes Feld fuer einen Angebotsbezug, daher wird der Fund als Freitext
|
||
(BT-22) uebernommen. Gibt den normalisierten Text zurueck oder "" wenn nichts passt."""
|
||
m = ANGEBOT_PATTERN.search(text)
|
||
if not m:
|
||
return ""
|
||
return f"Angebot {m.group(1)} vom {m.group(2)}"
|
||
|
||
|
||
# Referenzzeilen, die ein eigenes EN16931-Feld haben und daher NICHT in den Vorlauf-Freitext
|
||
# gehoeren (Auftragsbestaetigung BT-14, Angebot/Vergabeangebot, Lieferschein BT-16, Vertrag BT-12).
|
||
_REFERENCE_LINE_PATTERN = re.compile(
|
||
r"^(AUFTRAGSBESTÄTIGUNG|VERGABEANGEBOT|ANGEBOT|AUSSCHREIBUNG|LIEFERSCHEIN|VERTRAG|BESTELLUNG)\b",
|
||
re.IGNORECASE,
|
||
)
|
||
# Positionsnummer: alleinstehend "1" / "12" oder gegliedert "2.1". Markiert das Ende des Vorlaufs.
|
||
_POSITION_NUMBER_PATTERN = re.compile(r"^\d{1,3}(?:\.\d{1,3})?$")
|
||
|
||
|
||
def extract_pre_position_freetext(text):
|
||
"""Liest den beschreibenden Vorlauftext zwischen dem Tabellenkopf ('Pos.') und der ersten
|
||
Position deterministisch aus dem PDF-Text - unabhaengig vom (oft unvollstaendigen) LLM. Alles
|
||
zwischen der ersten 'Pos.'-Zeile und der ersten Positionsnummer wird uebernommen, ausser den
|
||
Referenzzeilen (Auftragsbestaetigung/Angebot/Vergabeangebot/Lieferschein/Vertrag), die eigene
|
||
Felder haben, und der zwischen Seiten wiederholten Moeblierung (Seitenkopf, Bank-/Register-
|
||
zeilen, Uebertrag/Vortrag) - relevant bei mehrseitigen Rechnungen, deren erste 'Pos.'-Zeile auf
|
||
Seite 1 steht, die erste echte Position aber erst nach einem Seitenumbruch. Gibt den Freitext
|
||
zurueck ("" wenn keiner vorhanden) oder None, wenn die Tabellenstruktur nicht erkennbar ist
|
||
(dann bleibt ein evtl. LLM-Freitext unveraendert)."""
|
||
lines = [ln.strip() for ln in text.split("\n")]
|
||
|
||
header_idx = next((i for i, ln in enumerate(lines) if ln == "Pos."), None)
|
||
if header_idx is None:
|
||
return None
|
||
|
||
first_pos_idx = next(
|
||
(i for i in range(header_idx + 1, len(lines)) if _POSITION_NUMBER_PATTERN.match(lines[i])),
|
||
None,
|
||
)
|
||
if first_pos_idx is None:
|
||
return None
|
||
|
||
# Leere Zeilen raus, dann die Seiten-Moeblierung strippen (Seitenkopf 'RECHNUNG'..'Betrag €',
|
||
# Bank-/Registerzeilen, Uebertrag/Vortrag), zuletzt die Referenzzeilen mit eigenem Feld.
|
||
region = [ln for ln in lines[header_idx + 1:first_pos_idx] if ln]
|
||
region = _strip_bundle_furniture(region)
|
||
freetext_lines = [ln for ln in region if not _REFERENCE_LINE_PATTERN.match(ln)]
|
||
return "\n".join(freetext_lines)
|
||
|
||
|
||
# Beleg-Referenzen aus den Referenzzeilen: Schluesselwort -> Feld. Zuverlaessiger als eine
|
||
# LLM-Zuordnung, die z.B. ein Vergabeangebot faelschlich in BT-12 (Vertrag) statt BT-17 einsortiert.
|
||
_DOC_REF_PATTERNS = {
|
||
"order_id": re.compile(r"\bAUFTRAGSBESTÄTIGUNG\s+([A-Za-z0-9][A-Za-z0-9\-/.]*)", re.IGNORECASE),
|
||
"delivery_note_id": re.compile(r"\bLIEFERSCHEIN\s+([A-Za-z0-9][A-Za-z0-9\-/.]*)", re.IGNORECASE),
|
||
}
|
||
# BT-17: Vergabeangebot. '\bVERGABEANGEBOT' greift dank Wortgrenze NICHT in ein blosses 'ANGEBOT'.
|
||
_TENDER_VERGABE_PATTERN = re.compile(r"\bVERGABEANGEBOT\s+([A-Za-z0-9][A-Za-z0-9\-/.]*)", re.IGNORECASE)
|
||
|
||
|
||
def extract_document_references(text):
|
||
"""Liest Beleg-Referenzen deterministisch aus den Referenzzeilen des PDF-Texts:
|
||
Auftragsbestaetigung (BT-14), Lieferschein (BT-16) und - fuer BT-17 - Vergabeangebot ODER,
|
||
falls keines vorhanden ist, ein einfaches Angebot ('Angebot <Nr> vom <Datum>'). Pro Rechnung
|
||
kommt nur eines von beiden vor; beide gehoeren fachlich in BT-17. Gibt ein Dict nur mit den
|
||
gefundenen Feldern zurueck, damit fehlende Felder nicht ueberschrieben werden."""
|
||
result = {}
|
||
for field, pattern in _DOC_REF_PATTERNS.items():
|
||
m = pattern.search(text)
|
||
if m:
|
||
result[field] = m.group(1).strip()
|
||
|
||
# BT-17: Vergabeangebot bevorzugt; sonst ein einfaches 'Angebot <Nr> vom <Datum>'
|
||
# (ANGEBOT_PATTERN erzwingt die 'vom <Datum>'-Form und vermeidet so Fehltreffer auf
|
||
# Fliesstext wie 'Angebot erstellen').
|
||
m_vergabe = _TENDER_VERGABE_PATTERN.search(text)
|
||
if m_vergabe:
|
||
result["tender_id"] = m_vergabe.group(1).strip()
|
||
else:
|
||
m_angebot = ANGEBOT_PATTERN.search(text)
|
||
if m_angebot:
|
||
result["tender_id"] = m_angebot.group(1).strip()
|
||
return result
|
||
|
||
|
||
def extract_labeled_header_value(text, label):
|
||
"""Liest einen Kopf-Feldwert deterministisch aus: findet die Zeile, die exakt <label> ist
|
||
(z.B. 'Ihr Auftrag:' oder 'Bearbeiter:'), und gibt den Wert der naechsten nicht-leeren Zeile
|
||
zurueck. Gibt "" zurueck, wenn das Label nicht vorkommt."""
|
||
lines = [ln.strip() for ln in text.split("\n")]
|
||
for i, ln in enumerate(lines):
|
||
if ln == label:
|
||
for j in range(i + 1, len(lines)):
|
||
if lines[j]:
|
||
return lines[j]
|
||
return ""
|
||
return ""
|
||
|
||
|
||
_REFERENZNUMMER_PATTERN = re.compile(r"\bReferenznummer\s+([A-Za-z0-9][A-Za-z0-9\-/.]*)", re.IGNORECASE)
|
||
|
||
|
||
def extract_referenznummer(text):
|
||
"""Findet eine 'Referenznummer <Wert>'-Zeile (z.B. SAP-Referenz). EN16931 hat dafuer kein
|
||
eigenes Feld - der Wert wird als Freitext (BT-22) uebernommen. "" wenn nicht vorhanden."""
|
||
m = _REFERENZNUMMER_PATTERN.search(text)
|
||
return m.group(1).strip() if m else ""
|
||
|
||
|
||
# Erkennt eine "PLZ Ort"-Zeile (deutsche 5-stellige PLZ + Ort), z.B. "16761 Hennigsdorf".
|
||
_ZIP_CITY_PATTERN = re.compile(r"^(\d{5})\s+(.+)$")
|
||
# E-Mail-Adresse (grob), zum Herausfischen einer Kunden-E-Mail aus dem Adressbereich.
|
||
_EMAIL_PATTERN = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
|
||
|
||
|
||
def extract_recipient_address(text):
|
||
"""Zerlegt den Empfaenger-Adressblock deterministisch in (name, street, zip, city).
|
||
|
||
Aufbau der scitech-Rechnungen: Zeile 0 ist die eigene Absender-Rueckadresse
|
||
('scitech ... Str. 4 ... Hennigsdorf'). Danach folgt die Rechnungsadresse des Kunden bis
|
||
zur ersten 'Lieferadresse:'- oder 'RECHNUNG'-Zeile. Die 'PLZ Ort'-Zeile ist der Anker:
|
||
die Zeile davor ist die Strasse, alles davor der (ggf. mehrzeilige) Firmenname. Eine
|
||
nachgestellte USt-IdNr.-Zeile ('DE 123456789') steht hinter 'PLZ Ort' und wird ignoriert.
|
||
Gibt ("", "", "", "") zurueck, wenn keine 'PLZ Ort'-Zeile gefunden wird."""
|
||
lines = text.split("\n")
|
||
end = len(lines)
|
||
for i in range(1, len(lines)):
|
||
s = lines[i].strip()
|
||
if s.lower().startswith("lieferadresse") or s.upper().startswith("RECHNUNG"):
|
||
end = i
|
||
break
|
||
# Zeile 0 (eigene Absenderadresse) ueberspringen, leere Zeilen entfernen.
|
||
block = [ln.strip() for ln in lines[1:end] if ln.strip()]
|
||
return _address_from_block(block)
|
||
|
||
|
||
def _address_from_block(block):
|
||
"""Zerlegt einen Adressblock (Liste nicht-leerer Zeilen) in (name, street, zip, city) anhand
|
||
der 'PLZ Ort'-Zeile als Anker: die Zeile davor ist die Strasse, alles davor der (ggf.
|
||
mehrzeilige) Firmenname. Zeilen hinter 'PLZ Ort' (z.B. USt-IdNr., E-Mail) werden ignoriert.
|
||
Gibt ("", "", "", "") zurueck, wenn keine 'PLZ Ort'-Zeile gefunden wird."""
|
||
zip_city_idx = None
|
||
zip_val = city_val = ""
|
||
for idx, ln in enumerate(block):
|
||
m = _ZIP_CITY_PATTERN.match(ln)
|
||
if m:
|
||
zip_city_idx = idx # letzte Uebereinstimmung gewinnt (die echte Adresse)
|
||
zip_val, city_val = m.group(1), m.group(2).strip()
|
||
|
||
if zip_city_idx is None or zip_city_idx < 1:
|
||
return "", "", "", ""
|
||
|
||
street = block[zip_city_idx - 1]
|
||
name = " ".join(block[: zip_city_idx - 1]).strip()
|
||
return name, street, zip_val, city_val
|
||
|
||
|
||
def extract_delivery_address(text):
|
||
"""Zerlegt den 'Lieferadresse:'-Block (falls vorhanden) in ein Dict mit name/street/zip/city
|
||
(BG-15 Lieferadresse). Der Block steht zwischen der 'Lieferadresse:'-Zeile und der ersten
|
||
'RECHNUNG'-Zeile; E-Mail-Zeilen (bei manchen Kunden vorhanden) werden ignoriert. Gibt None zurueck, wenn
|
||
keine Lieferadresse vorhanden ist oder der Block keine 'PLZ Ort'-Zeile enthaelt."""
|
||
lines = text.split("\n")
|
||
start = next((i for i, ln in enumerate(lines) if ln.strip().lower().startswith("lieferadresse")), None)
|
||
if start is None:
|
||
return None
|
||
block = []
|
||
for ln in lines[start + 1:]:
|
||
s = ln.strip()
|
||
if s.upper().startswith("RECHNUNG"):
|
||
break
|
||
if s and not _EMAIL_PATTERN.search(s):
|
||
block.append(s)
|
||
name, street, zip_val, city_val = _address_from_block(block)
|
||
if not (name and city_val):
|
||
return None
|
||
return {"name": name, "street": street, "zip": zip_val, "city": city_val}
|
||
|
||
|
||
def _normalize_address_text(*parts):
|
||
"""Normalisiert Adressteile fuer den Vergleich: kleinschreiben, 'Str.'/'Straße' vereinheitlichen,
|
||
dann alle Nicht-Alphanumerik entfernen. So sind Zeilenumbrueche und Schreibweisen egal."""
|
||
s = " ".join(str(p or "") for p in parts).lower()
|
||
s = s.replace("straße", "strasse").replace("str.", "strasse").replace("ß", "ss")
|
||
return re.sub(r"[^a-z0-9]", "", s)
|
||
|
||
|
||
def delivery_address_differs(delivery, recipient_name, recipient_street, recipient_zip, recipient_city):
|
||
"""True, wenn die Lieferadresse (dict von extract_delivery_address) inhaltlich von der
|
||
Rechnungsadresse abweicht - unempfindlich gegen Zeilenumbrueche und 'Str.'/'Straße'. Nur dann
|
||
traegt BG-15 echte Information (sonst waere die Lieferadresse eine reine Dopplung)."""
|
||
if not delivery:
|
||
return False
|
||
d = _normalize_address_text(delivery.get("name"), delivery.get("street"), delivery.get("zip"), delivery.get("city"))
|
||
b = _normalize_address_text(recipient_name, recipient_street, recipient_zip, recipient_city)
|
||
return d != b
|
||
|
||
|
||
def extract_recipient_email(text):
|
||
"""Sucht im Adressbereich (bis zur 'RECHNUNG'-Zeile) nach einer Kunden-E-Mail. Die eigene
|
||
scitech-Adresse wird ausgeschlossen. Gibt "" zurueck, wenn keine gefunden wird."""
|
||
lines = text.split("\n")
|
||
end = len(lines)
|
||
for i, ln in enumerate(lines):
|
||
if ln.strip().upper().startswith("RECHNUNG"):
|
||
end = i
|
||
break
|
||
for ln in lines[:end]:
|
||
m = _EMAIL_PATTERN.search(ln)
|
||
if m and "scitech" not in m.group(0).lower():
|
||
return m.group(0)
|
||
return ""
|
||
|
||
|
||
# USt-Satz aus der Steuerzeile lesen, z.B. '19 % USt. auf € 2.680,00' -> 19.0.
|
||
_VAT_RATE_PATTERN = re.compile(r"(\d+(?:[.,]\d+)?)\s*%\s*USt", re.IGNORECASE)
|
||
|
||
|
||
def extract_header_metadata_pure_python(text):
|
||
"""Deterministischer Ersatz fuer die LLM-Kopfdatenextraktion (test-version3, KI-frei).
|
||
|
||
Liefert dasselbe Dict-Schema, das zuvor das vLLM/gemma-Modell produziert hat, damit die
|
||
nachgelagerte Pipeline in main.py (Positionen, Beleg-Referenzen, Freitext, Angebots-/
|
||
Referenzbezug usw.) unveraendert darauf aufsetzen kann. Alle Werte stammen aus den
|
||
regelmaessigen Kopf-Labeln ('Nummer:', 'Datum:', 'Kunden-Nr.:', ...) und dem Adressblock
|
||
der scitech-Rechnungen. Felder ohne dedizierte Label (leitweg_id, doc-refs, invoice_note)
|
||
bleiben leer - sie werden in main.py separat deterministisch befuellt bzw. im UI ergaenzt."""
|
||
name, street, zip_val, city_val = extract_recipient_address(text)
|
||
|
||
zahlungsziel = extract_labeled_header_value(text, "Zahlungsziel:")
|
||
m_days = re.search(r"\d+", zahlungsziel)
|
||
|
||
m_vat = _VAT_RATE_PATTERN.search(text)
|
||
vat_rate = float(m_vat.group(1).replace(",", ".")) if m_vat else 19.0
|
||
|
||
# BG-15 Lieferadresse nur vorbelegen, wenn sie von der Rechnungsadresse abweicht - sonst waere
|
||
# sie reine Dopplung (die Rechnungsadresse ist meist auch der Lieferort). Die UI-Felder bleiben
|
||
# dann leer und es wird keine BG-15 exportiert.
|
||
_deliv = extract_delivery_address(text)
|
||
_deliv = _deliv if delivery_address_differs(_deliv, name, street, zip_val, city_val) else None
|
||
|
||
return {
|
||
"invoice_number": extract_labeled_header_value(text, "Nummer:"),
|
||
"invoice_date": parse_to_strict_xml_date(extract_labeled_header_value(text, "Datum:")),
|
||
"delivery_date": parse_to_strict_xml_date(extract_labeled_header_value(text, "Lieferdatum:")),
|
||
"customer_number": extract_labeled_header_value(text, "Kunden-Nr.:"),
|
||
"leitweg_id": "", # BT-10: nur B2G, kein eigenes Label - im UI zu erfassen
|
||
"buyer_reference": extract_labeled_header_value(text, "Ihr Auftrag:"),
|
||
"recipient_name": name,
|
||
"recipient_street": street,
|
||
"recipient_zip": zip_val,
|
||
"recipient_city": city_val,
|
||
"recipient_email": extract_recipient_email(text),
|
||
"vat_rate_percent": vat_rate,
|
||
"delivery_note_id": "", # wird in main.py via extract_document_references gesetzt
|
||
"order_id": "",
|
||
"contract_id": "",
|
||
"tender_id": "",
|
||
"service_start_date": parse_to_strict_xml_date(extract_labeled_header_value(text, "Zeitraum von:")),
|
||
"service_end_date": parse_to_strict_xml_date(extract_labeled_header_value(text, "Zeitraum bis:")),
|
||
"payment_term_days": m_days.group(0) if m_days else "",
|
||
"invoice_note": "", # wird in main.py via extract_pre_position_freetext gesetzt
|
||
# BG-15 Lieferadresse (nur bei abweichendem Lieferort vorbelegt, sonst leer):
|
||
"delivery_name": _deliv["name"] if _deliv else "",
|
||
"delivery_street": _deliv["street"] if _deliv else "",
|
||
"delivery_zip": _deliv["zip"] if _deliv else "",
|
||
"delivery_city": _deliv["city"] if _deliv else "",
|
||
}
|
||
|
||
|
||
def sanitize_filename_part(value):
|
||
"""Macht einen String dateisystem-sicher: entfernt in Dateinamen kritische Zeichen
|
||
(Pfadtrenner/Sonderzeichen) und fasst Whitespace zu einem Bindestrich zusammen."""
|
||
s = str(value or "").strip()
|
||
s = re.sub(r'[\\/:*?"<>|]', "", s)
|
||
s = re.sub(r"\s+", "-", s)
|
||
return s.strip("-_")
|
||
|
||
|
||
# Maximale Laenge des Empfaengernamens im Dateinamen - lange Kundennamen wuerden ihn sonst
|
||
# unhandlich machen.
|
||
EXPORT_RECIPIENT_MAX_LEN = 15
|
||
|
||
# Bearbeiter-Kopfzeile hat die Form '<Name> / <Nummer>', z.B. 'Max Mustermann / 1234'.
|
||
_BEARBEITER_NUMBER_PATTERN = re.compile(r"/\s*([A-Za-z0-9][A-Za-z0-9\-.]*)\s*$")
|
||
|
||
|
||
def extract_bearbeiter_number(bearbeiter):
|
||
"""Holt die Nummer hinter dem Bearbeiternamen ('Max Mustermann / 1234' -> '1234').
|
||
Steht im Feld nur eine Zahl, wird diese direkt genommen; ein Name ohne Nummer ergibt "",
|
||
damit kein Personenname im Dateinamen landet."""
|
||
s = str(bearbeiter or "").strip()
|
||
if not s:
|
||
return ""
|
||
m = _BEARBEITER_NUMBER_PATTERN.search(s)
|
||
if m:
|
||
return m.group(1)
|
||
return s if s.isdigit() else ""
|
||
|
||
|
||
def build_export_basename(invoice_number, recipient_name, bearbeiter):
|
||
"""Dateiname-Basis fuer die Exporte: '<Rechnungsnummer>_<Empfaenger>_<Bearbeiternummer>'.
|
||
Der Empfaenger wird auf EXPORT_RECIPIENT_MAX_LEN Zeichen gekuerzt. Leere Teile werden
|
||
weggelassen (kein '__'); ist alles leer, bleibt 'Rechnung'."""
|
||
recipient = sanitize_filename_part(recipient_name)[:EXPORT_RECIPIENT_MAX_LEN].strip("-_")
|
||
parts = [
|
||
sanitize_filename_part(invoice_number),
|
||
recipient,
|
||
sanitize_filename_part(extract_bearbeiter_number(bearbeiter)),
|
||
]
|
||
base = "_".join(p for p in parts if p)
|
||
return base or "Rechnung"
|
||
|
||
|
||
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|"
|
||
r"Wartungspauschale|Pauschale|Pauschal|Pausch\.|Psch\.)$",
|
||
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
|
||
|
||
# Abbrechen bei der echten Positionsnummer:
|
||
# - gegliederte Nummer "N.M" (z.B. "2.1") ist eindeutig eine (Unter-)Position;
|
||
# - eine alleinstehende Zahl nur, wenn sie der erwarteten Positionsnummer
|
||
# (len(items)+1) entspricht - so werden Ziffern-Fragmente aufgesplitteter
|
||
# Seriennummern (z.B. "9") NICHT faelschlich als Positionsnummer gewertet.
|
||
if re.match(r"^\d{1,3}\.\d{1,3}$", prev_line):
|
||
break
|
||
if prev_line.isdigit() and len(prev_line) <= 2 and int(prev_line) == len(items) + 1:
|
||
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
|
||
|
||
|
||
# ==========================================
|
||
# BUNDLE-RECHNUNGEN (konfigurierte Positionen mit Komponentenliste, z.B. Dell-Notebooks)
|
||
# ==========================================
|
||
# Solche Rechnungen (Signal: "Menge Komponenten") haben einen voellig anderen Aufbau als die
|
||
# ueblichen Service-Rechnungen: jede Position ist ein konfiguriertes Buendel aus Produktname +
|
||
# Dutzenden Komponentenzeilen, verteilt ueber viele Seiten mit wiederholten Seitenkoepfen und
|
||
# "Uebertrag/Vortrag"-Zeilen. Der Preis steht als 3-Zeilen-Block (Menge+Einheit / Einzelpreis /
|
||
# Zeilensumme) mittendrin; die Service-Tags (Seriennummern) folgen erst danach. Dafuer gibt es
|
||
# diesen eigenen Parser - der uebrige (haeufigere) Rechnungstyp bleibt beim Standardparser.
|
||
|
||
_BUNDLE_MARKER = "Menge Komponenten"
|
||
_BUNDLE_MONEY_LINE = re.compile(r"^[\d.]+,\d{2}$")
|
||
_BUNDLE_UNIT_LINE = re.compile(
|
||
r"^(\d+(?:[.,]\d+)?)\s*(Stk\.|pausch\.|pauschal|Monat\(e\)|Tag\(e\)|Std\.|Stück)$",
|
||
re.IGNORECASE,
|
||
)
|
||
_STATED_TAX_BASIS_PATTERN = re.compile(r"%\s*USt\.?\s*auf\s*€\s*([\d.]+,\d{2})", re.IGNORECASE)
|
||
|
||
|
||
def is_bundle_invoice(text):
|
||
"""True, wenn die Rechnung dem Buendel-Layout entspricht (konfigurierte Positionen mit
|
||
Komponentenliste) - erkennbar am Marker 'Menge Komponenten'. Nur dann greift der Bundle-Parser."""
|
||
return _BUNDLE_MARKER in text
|
||
|
||
|
||
def _is_bundle_position_number(line):
|
||
"""Eine alleinstehende kleine Ganzzahl (Positions- oder Komponentennummer)."""
|
||
return line.isdigit() and len(line) <= 3
|
||
|
||
|
||
def _strip_bundle_furniture(lines):
|
||
"""Entfernt die zwischen den Seiten wiederholte Moeblierung aus dem Zeilenstrom: den ganzen
|
||
Seitenkopf ('RECHNUNG' ... 'Betrag €'), die Bank-/Registerzeilen sowie 'Uebertrag/Vortrag'
|
||
samt der jeweils angrenzenden laufenden Uebertragssumme (Geldzeile)."""
|
||
out = []
|
||
i, n = 0, len(lines)
|
||
while i < n:
|
||
ln = lines[i]
|
||
if ln == "RECHNUNG": # kompletten Seitenkopf bis 'Betrag €' ueberspringen
|
||
j = i + 1
|
||
while j < n and lines[j] != "Betrag €":
|
||
j += 1
|
||
i = j + 1
|
||
continue
|
||
if ln.startswith("Deutsche Bank ") or ln.startswith("Registergericht"):
|
||
i += 1
|
||
continue
|
||
if ln == "Übertrag:":
|
||
if out and _BUNDLE_MONEY_LINE.match(out[-1]):
|
||
out.pop() # laufende Uebertragssumme davor entfernen
|
||
i += 1
|
||
continue
|
||
if ln == "Vortrag:":
|
||
i += 1
|
||
if i < n and _BUNDLE_MONEY_LINE.match(lines[i]):
|
||
i += 1 # laufende Vortragssumme danach entfernen
|
||
continue
|
||
out.append(ln)
|
||
i += 1
|
||
return out
|
||
|
||
|
||
def extract_bundle_items(text):
|
||
"""Parst eine Buendel-Rechnung: liefert je Position (Produkt) name, description (volle
|
||
Komponentenliste + Service-Tags = BT-154), quantity, unit, unit_price. Ankerpunkt ist der
|
||
3-Zeilen-Preisblock; die Positionsgrenzen sind die Positions-Header (Name vor 'Menge
|
||
Komponenten' bzw. die Zeile hinter der Positionsnummer bei Positionen ohne Komponentenliste)."""
|
||
raw = [l.strip() for l in text.split("\n") if l.strip()]
|
||
lines = _strip_bundle_furniture(raw)
|
||
end = next((i for i, l in enumerate(lines) if l.startswith("Zwischensumme")), len(lines))
|
||
lines = lines[:end]
|
||
|
||
# 1) Preisbloecke: <Menge Einheit> gefolgt von 2 Geldzeilen (Einzelpreis, Zeilensumme).
|
||
price_blocks = []
|
||
for i in range(len(lines) - 2):
|
||
m = _BUNDLE_UNIT_LINE.match(lines[i])
|
||
if m and _BUNDLE_MONEY_LINE.match(lines[i + 1]) and _BUNDLE_MONEY_LINE.match(lines[i + 2]):
|
||
price_blocks.append((i, clean_german_float(m.group(1)), m.group(2),
|
||
clean_german_float(lines[i + 1])))
|
||
if not price_blocks:
|
||
return []
|
||
|
||
marker_idxs = [i for i, l in enumerate(lines) if l == _BUNDLE_MARKER]
|
||
|
||
# 2) Fuer jeden Preisblock den Positions-Header (Startindex + Name) bestimmen.
|
||
headers = [] # (start_idx, name, pb_idx, qty, unit, unit_price)
|
||
prev_boundary = 0
|
||
for (pb_idx, qty, unit, up) in price_blocks:
|
||
mk = next((m for m in reversed(marker_idxs) if prev_boundary <= m < pb_idx), None)
|
||
if mk is not None:
|
||
# Name = Zeilen direkt vor 'Menge Komponenten' bis zur Positionsnummer.
|
||
k = mk - 1
|
||
name_lines = []
|
||
while k >= prev_boundary and not _is_bundle_position_number(lines[k]):
|
||
name_lines.insert(0, lines[k])
|
||
k -= 1
|
||
start = k if (k >= prev_boundary and _is_bundle_position_number(lines[k])) else k + 1
|
||
name = " ".join(name_lines).strip()
|
||
else:
|
||
# Position ohne Komponentenliste (z.B. Serviceleistung): rueckwaerts vom Preisblock die
|
||
# Positionsnummer suchen; die Zeile dahinter ist der Name.
|
||
k = pb_idx
|
||
while k > prev_boundary and not _is_bundle_position_number(lines[k - 1]):
|
||
k -= 1
|
||
start = k - 1 if (k - 1 >= prev_boundary and _is_bundle_position_number(lines[k - 1])) else k
|
||
name = lines[k] if k < pb_idx else ""
|
||
headers.append((start, name, pb_idx, qty, unit, up))
|
||
prev_boundary = pb_idx + 3
|
||
|
||
# 3) Positionen bauen: Beschreibung = Segment vom Header bis zum naechsten Header,
|
||
# ohne Preisblock-Zeilen, fuehrende Positionsnummer und verirrte Geldzeilen.
|
||
items = []
|
||
for h_i, (start, name, pb_idx, qty, unit, up) in enumerate(headers):
|
||
seg_end = headers[h_i + 1][0] if h_i + 1 < len(headers) else len(lines)
|
||
seg = lines[start:seg_end]
|
||
skip = {pb_idx - start, pb_idx - start + 1, pb_idx - start + 2}
|
||
desc_lines = []
|
||
for idx, l in enumerate(seg):
|
||
if idx in skip:
|
||
continue
|
||
if idx == 0 and _is_bundle_position_number(l):
|
||
continue
|
||
if _BUNDLE_MONEY_LINE.match(l):
|
||
continue
|
||
desc_lines.append(l)
|
||
items.append({
|
||
"name": name,
|
||
"description": "\n".join(desc_lines).strip(),
|
||
"quantity": qty,
|
||
"unit": unit,
|
||
"unit_price": up,
|
||
})
|
||
return items
|
||
|
||
|
||
def extract_invoice_items(text):
|
||
"""Dispatcher: Buendel-Rechnungen (konfigurierte Positionen mit Komponentenliste) ueber den
|
||
Bundle-Parser, alle uebrigen ueber den Standardparser (unveraendertes Verhalten)."""
|
||
if is_bundle_invoice(text):
|
||
return extract_bundle_items(text)
|
||
return extract_invoice_items_pure_python(text)
|
||
|
||
|
||
def extract_stated_tax_basis(text):
|
||
"""Liest die auf der Rechnung ausgewiesene Bemessungsgrundlage aus der Steuerzeile
|
||
('19 % USt. auf € 217.500,55' -> 217500.55). Fuer den Summen-Abgleich als Sicherheitsnetz.
|
||
Gibt None zurueck, wenn keine solche Zeile gefunden wird."""
|
||
m = _STATED_TAX_BASIS_PATTERN.search(text)
|
||
return clean_german_float(m.group(1)) if m else None
|
||
|
||
|
||
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 ""
|
||
|
||
|
||
# Rechnungstyp (BT-3) an Schluesselwoertern im Titel/Text erkennen -> UN/UNTDID-1001-Code.
|
||
_INVOICE_TYPE_KEYWORDS = [
|
||
(re.compile(r"\bTeilrechnung\b", re.IGNORECASE), "326"), # Teilrechnung
|
||
(re.compile(r"\bKorrekturrechnung\b", re.IGNORECASE), "384"), # Korrekturrechnung
|
||
]
|
||
|
||
|
||
def detect_invoice_type_code(text):
|
||
"""Erkennt den Rechnungstyp (BT-3) an Schluesselwoertern im Rechnungstext (z.B. 'Teilrechnung'
|
||
-> 326, 'Korrekturrechnung' -> 384). Gibt sonst '380' (Handelsrechnung) zurueck."""
|
||
for pattern, code in _INVOICE_TYPE_KEYWORDS:
|
||
if pattern.search(text):
|
||
return code
|
||
return "380"
|
||
|
||
|
||
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 = Leitweg-ID: NUR bei Behörden-Rechnungen (B2G) vorhanden. Fehlt sie,
|
||
# wird das Element weggelassen. BT-10 ist in XRechnung Pflicht (BR-DE-15), darf aber nie leer
|
||
# sein: bei Behoerden (B2G) die Leitweg-ID, sonst als Fallback die Kundennummer.
|
||
_buyer_ref_bt10 = str(data.get("leitweg_id") or data.get("customer_number") or "").strip()
|
||
if _buyer_ref_bt10:
|
||
ET.SubElement(root, f"{{{CBC}}}BuyerReference").text = _buyer_ref_bt10
|
||
|
||
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()
|
||
|
||
# BT-17 Vergabeangebot/Ausschreibung. Muss laut UBL-Schema (cac:OriginatorDocumentReference)
|
||
# VOR cac:ContractDocumentReference (BT-12) stehen.
|
||
if data.get("tender_id") and str(data["tender_id"]).strip() != "":
|
||
originator_ref = ET.SubElement(root, f"{{{CAC}}}OriginatorDocumentReference")
|
||
ET.SubElement(originator_ref, f"{{{CBC}}}ID").text = str(data["tender_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 = str(data.get("seller_endpoint", 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 = str(data.get("seller_contact_name", VERKAEUFER["contact_name"]))
|
||
ET.SubElement(contact_supp, f"{{{CBC}}}Telephone").text = VERKAEUFER["phone"]
|
||
ET.SubElement(contact_supp, f"{{{CBC}}}ElectronicMail").text = str(data.get("seller_contact_email", VERKAEUFER["email"]))
|
||
|
||
# --- Empfänger ---
|
||
customer = ET.SubElement(root, f"{{{CAC}}}AccountingCustomerParty")
|
||
party_cust = ET.SubElement(customer, f"{{{CAC}}}Party")
|
||
# BT-49-1 Schema der elektronischen Kaeufer-Adresse: dynamisch (EM/0204/0088), Default EM.
|
||
ET.SubElement(party_cust, f"{{{CBC}}}EndpointID", {"schemeID": str(data.get("recipient_endpoint_scheme", "EM"))}).text = str(data.get("recipient_email", "")).strip()
|
||
# BT-46 Kundennummer (Käufer-Kennung). Muss laut UBL-Schema nach EndpointID und vor
|
||
# PostalAddress stehen.
|
||
if data.get("customer_number") and str(data["customer_number"]).strip() != "":
|
||
party_ident_cust = ET.SubElement(party_cust, f"{{{CAC}}}PartyIdentification")
|
||
ET.SubElement(party_ident_cust, f"{{{CBC}}}ID").text = str(data["customer_number"]).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"])
|
||
|
||
# BG-15 Lieferadresse nur, wenn Name + Ort gesetzt sind (im UI nur bei abweichendem Lieferort
|
||
# vorbelegt). UBL-Reihenfolge in cac:Delivery: ActualDeliveryDate, DeliveryLocation, DeliveryParty.
|
||
_has_deliv_addr = bool(str(data.get("delivery_name") or "").strip() and str(data.get("delivery_city") or "").strip())
|
||
if data.get("delivery_date") or _has_deliv_addr:
|
||
delivery = ET.SubElement(root, f"{{{CAC}}}Delivery")
|
||
if data.get("delivery_date"):
|
||
ET.SubElement(delivery, f"{{{CBC}}}ActualDeliveryDate").text = parse_to_strict_xml_date(data["delivery_date"])
|
||
if _has_deliv_addr:
|
||
deliv_loc = ET.SubElement(delivery, f"{{{CAC}}}DeliveryLocation")
|
||
deliv_addr = ET.SubElement(deliv_loc, f"{{{CAC}}}Address")
|
||
if str(data.get("delivery_street") or "").strip():
|
||
ET.SubElement(deliv_addr, f"{{{CBC}}}StreetName").text = str(data["delivery_street"]).strip() # BT-75
|
||
ET.SubElement(deliv_addr, f"{{{CBC}}}CityName").text = str(data["delivery_city"]).strip() # BT-77
|
||
if str(data.get("delivery_zip") or "").strip():
|
||
ET.SubElement(deliv_addr, f"{{{CBC}}}PostalZone").text = str(data["delivery_zip"]).strip() # BT-78
|
||
country = ET.SubElement(deliv_addr, f"{{{CAC}}}Country")
|
||
ET.SubElement(country, f"{{{CBC}}}IdentificationCode").text = "DE" # BT-80
|
||
deliv_party = ET.SubElement(delivery, f"{{{CAC}}}DeliveryParty")
|
||
party_name_deliv = ET.SubElement(deliv_party, f"{{{CAC}}}PartyName")
|
||
ET.SubElement(party_name_deliv, f"{{{CBC}}}Name").text = str(data["delivery_name"]).strip() # BT-70
|
||
|
||
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"))
|
||
# BT-83 Zahlungsreferenz (Verwendungszweck), damit der Empfaenger die Zahlung automatisch
|
||
# zuordnen kann - standardmaessig die Rechnungsnummer, im UI aber ueberschreibbar. Muss laut
|
||
# UBL-Sequenz NACH PaymentMeansCode und VOR PayeeFinancialAccount stehen.
|
||
_payment_ref = resolve_payment_reference(data)
|
||
if _payment_ref:
|
||
ET.SubElement(payment_means, f"{{{CBC}}}PaymentID").text = _payment_ref
|
||
payee_acc = ET.SubElement(payment_means, f"{{{CAC}}}PayeeFinancialAccount")
|
||
ET.SubElement(payee_acc, f"{{{CBC}}}ID").text = str(data["iban"]).replace(" ", "")
|
||
# BT-86 BIC der Verkaeuferbank (optional, aber von vielen Empfaengern erwartet).
|
||
_bic = str(data.get("bic") or VERKAEUFER["bic"]).replace(" ", "")
|
||
if _bic:
|
||
fin_branch = ET.SubElement(payee_acc, f"{{{CAC}}}FinancialInstitutionBranch")
|
||
ET.SubElement(fin_branch, f"{{{CBC}}}ID").text = _bic
|
||
|
||
# BT-20 Zahlungsbedingungen als Freitext in cac:PaymentTerms/cbc:Note (Vorkasse und/oder
|
||
# Skonto) - EN16931 kennt dafuer kein eigenes strukturiertes Feld. Muss laut UBL-Schema NACH
|
||
# PaymentMeans und VOR AllowanceCharge/TaxTotal stehen.
|
||
_payment_terms_note = build_payment_terms_note(data)
|
||
if _payment_terms_note:
|
||
payment_terms = ET.SubElement(root, f"{{{CAC}}}PaymentTerms")
|
||
ET.SubElement(payment_terms, f"{{{CBC}}}Note").text = _payment_terms_note
|
||
|
||
# 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}"
|
||
# BT-113 Vorauszahlung: mangels Anzahlungsdaten immer 0,00 - macht die Summensektion
|
||
# vollstaendig (viele Empfaenger/Validatoren erwarten das Feld). Muss VOR PayableAmount stehen.
|
||
ET.SubElement(monetary_total, f"{{{CBC}}}PrepaidAmount", {"currencyID": "EUR"}).text = "0.00"
|
||
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")
|
||
# BT-24 Profil: ZUGFeRD/Factur-X wird B2B verschickt - daher das EN16931-Profil (COMFORT)
|
||
# deklarieren, NICHT das strengere KoSIT-XRechnung-Behoerdenprofil. So passt die
|
||
# Profildeklaration zum factur-x-Level 'en16931' und zur B2B-Ausrichtung (vgl. Lieferanten-XML).
|
||
ET.SubElement(guideline, f"{{{RAM}}}ID").text = "urn:cen.eu:en16931:2017"
|
||
|
||
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."))
|
||
# Menge identisch zum UBL-Generator formatieren: ganze Zahlen mit 2 Nachkommastellen
|
||
# (z.B. "1.00"), gebrochene Mengen nur mit den noetigen Stellen (z.B. "1.375") - so
|
||
# erscheinen Stueckzahlen nicht als "1.0000", ohne dass Nachkommastellen verloren gehen.
|
||
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_deliv, f"{{{RAM}}}BilledQuantity", {"unitCode": u_code}).text = qty_formatted
|
||
|
||
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: bei Behörden (B2G) die Leitweg-ID, sonst als Fallback die Kundennummer,
|
||
# damit das in XRechnung verpflichtende Feld (BR-DE-15) nie leer bleibt.
|
||
_buyer_ref_bt10 = str(data.get("leitweg_id") or data.get("customer_number") or "").strip()
|
||
if _buyer_ref_bt10:
|
||
ET.SubElement(header_agree, f"{{{RAM}}}BuyerReference").text = _buyer_ref_bt10
|
||
|
||
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 = str(data.get("seller_contact_name", 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 = str(data.get("seller_contact_email", 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 = str(data.get("seller_endpoint", 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")
|
||
# BT-46 Kundennummer (Käufer-Kennung). Muss laut CII-Schema als ram:ID VOR ram:Name stehen.
|
||
if data.get("customer_number") and str(data["customer_number"]).strip() != "":
|
||
ET.SubElement(customer_party, f"{{{RAM}}}ID").text = str(data["customer_number"]).strip()
|
||
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. Muss laut XSD nach PostalTradeAddress stehen.
|
||
# Bei XRechnung Pflicht (UI blockiert leer), bei ZUGFeRD (B2B) optional - fehlt sie, wird der
|
||
# Block ganz weggelassen statt als leeres Element geschrieben.
|
||
if str(data.get("recipient_email", "")).strip():
|
||
cust_uri = ET.SubElement(customer_party, f"{{{RAM}}}URIUniversalCommunication")
|
||
# BT-49-1 Schema der elektronischen Kaeufer-Adresse: dynamisch (EM/0204/0088), Default EM.
|
||
ET.SubElement(cust_uri, f"{{{RAM}}}URIID", {"schemeID": str(data.get("recipient_endpoint_scheme", "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()
|
||
|
||
# BT-17 Vergabeangebot/Ausschreibung. In CII ein ram:AdditionalReferencedDocument mit
|
||
# ram:TypeCode "50"; muss laut XSD NACH ContractReferencedDocument stehen.
|
||
if data.get("tender_id") and str(data["tender_id"]).strip() != "":
|
||
tender_ref = ET.SubElement(header_agree, f"{{{RAM}}}AdditionalReferencedDocument")
|
||
ET.SubElement(tender_ref, f"{{{RAM}}}IssuerAssignedID").text = str(data["tender_id"]).strip()
|
||
ET.SubElement(tender_ref, f"{{{RAM}}}TypeCode").text = "50"
|
||
|
||
header_deliv = ET.SubElement(trade_trans, f"{{{RAM}}}ApplicableHeaderTradeDelivery")
|
||
# BG-15 Lieferadresse (ShipToTradeParty) nur bei gesetztem Name + Ort. Muss laut CII-Sequenz
|
||
# VOR dem Lieferdatum (ActualDeliverySupplyChainEvent) stehen.
|
||
if str(data.get("delivery_name") or "").strip() and str(data.get("delivery_city") or "").strip():
|
||
ship_to = ET.SubElement(header_deliv, f"{{{RAM}}}ShipToTradeParty")
|
||
ET.SubElement(ship_to, f"{{{RAM}}}Name").text = str(data["delivery_name"]).strip() # BT-70
|
||
deliv_addr = ET.SubElement(ship_to, f"{{{RAM}}}PostalTradeAddress")
|
||
if str(data.get("delivery_zip") or "").strip():
|
||
ET.SubElement(deliv_addr, f"{{{RAM}}}PostcodeCode").text = str(data["delivery_zip"]).strip() # BT-78
|
||
if str(data.get("delivery_street") or "").strip():
|
||
ET.SubElement(deliv_addr, f"{{{RAM}}}LineOne").text = str(data["delivery_street"]).strip() # BT-75
|
||
ET.SubElement(deliv_addr, f"{{{RAM}}}CityName").text = str(data["delivery_city"]).strip() # BT-77
|
||
ET.SubElement(deliv_addr, f"{{{RAM}}}CountryID").text = "DE" # BT-80
|
||
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")
|
||
# BT-83 Zahlungsreferenz (Verwendungszweck), damit der Empfaenger die Zahlung automatisch
|
||
# zuordnen kann - standardmaessig die Rechnungsnummer, im UI aber ueberschreibbar. Muss laut
|
||
# CII-Sequenz VOR InvoiceCurrencyCode stehen (die Sequenz ist CreditorReferenceID ->
|
||
# PaymentReference -> ... -> InvoiceCurrencyCode).
|
||
_payment_ref = resolve_payment_reference(data)
|
||
if _payment_ref:
|
||
ET.SubElement(header_settle, f"{{{RAM}}}PaymentReference").text = _payment_ref
|
||
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(" ", "")
|
||
# BT-86 BIC der Verkaeuferbank (nach dem Konto, laut CII-Schema).
|
||
_bic = str(data.get("bic") or VERKAEUFER["bic"]).replace(" ", "")
|
||
if _bic:
|
||
fin_inst = ET.SubElement(pay_means, f"{{{RAM}}}PayeeSpecifiedCreditorFinancialInstitution")
|
||
ET.SubElement(fin_inst, f"{{{RAM}}}BICID").text = _bic
|
||
|
||
# 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"))
|
||
_payment_terms_note = build_payment_terms_note(data)
|
||
if _cii_due or _payment_terms_note:
|
||
payment_terms = ET.SubElement(header_settle, f"{{{RAM}}}SpecifiedTradePaymentTerms")
|
||
if _payment_terms_note:
|
||
ET.SubElement(payment_terms, f"{{{RAM}}}Description").text = _payment_terms_note
|
||
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}"
|
||
# BT-113 Vorauszahlung: mangels Anzahlungsdaten immer 0,00. Muss laut CII-Schema NACH
|
||
# GrandTotalAmount und VOR DuePayableAmount stehen (DuePayable = GrandTotal - Vorauszahlung).
|
||
ET.SubElement(header_sum, f"{{{RAM}}}TotalPrepaidAmount").text = "0.00"
|
||
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 normalize_pdf_bytes(pdf_bytes: bytes) -> bytes:
|
||
"""Oeffnet die PDF mit pikepdf und speichert sie neu. Das erzeugt eine saubere,
|
||
standardkonforme PDF-Struktur (Seitenbaum/XRef). Manche Ursprungs-PDFs (z.B. von PDF-XChange
|
||
erzeugte, die bereits ein OutputIntent tragen) haben eine Struktur, an der factur-x sonst
|
||
scheitert: die factur-x.xml wird nicht eingebettet und der Seitenbaum geht verloren
|
||
('unable to find page tree'). Schlaegt das Normalisieren selbst fehl, werden die Original-Bytes
|
||
unveraendert zurueckgegeben, damit der Export nicht komplett blockiert."""
|
||
try:
|
||
pdf = pikepdf.open(io.BytesIO(pdf_bytes))
|
||
buf = io.BytesIO()
|
||
pdf.save(buf)
|
||
pdf.close()
|
||
return buf.getvalue()
|
||
except Exception:
|
||
return pdf_bytes
|
||
|
||
|
||
# Wie viele Leistungsnachweise maximal an eine Rechnung gehaengt werden duerfen. Begrenzt,
|
||
# damit die Export-PDF handhabbar bleibt und ein versehentlicher Massen-Upload auffaellt.
|
||
MAX_ATTACHMENT_PDFS = 3
|
||
|
||
|
||
def append_pdf_pages(base_pdf_bytes: bytes, *extra_pdf_bytes: bytes) -> bytes:
|
||
"""Haengt die Seiten aller uebergebenen PDFs in Reihenfolge hinten an base_pdf_bytes an
|
||
(z.B. Leistungsnachweise an die Rechnung) und gibt die zusammengefuehrte PDF zurueck. Aus
|
||
den angehaengten PDFs wird nichts ausgelesen - sie sind reiner Seiteninhalt. Ohne Anhang
|
||
kommen die Basis-Bytes unveraendert zurueck.
|
||
|
||
Bewusst OHNE try/except: schlaegt das Zusammenfuehren fehl (kaputte oder passwortgeschuetzte
|
||
Datei), muss der Export sichtbar scheitern. Eine stillschweigend ohne Leistungsnachweis
|
||
exportierte Rechnung waere schlimmer als eine Fehlermeldung - anders als bei
|
||
normalize_pdf_bytes, wo das Zurueckfallen auf die Eingabe harmlos ist."""
|
||
if not extra_pdf_bytes:
|
||
return base_pdf_bytes
|
||
with pikepdf.open(io.BytesIO(base_pdf_bytes)) as base:
|
||
for extra_bytes in extra_pdf_bytes:
|
||
with pikepdf.open(io.BytesIO(extra_bytes)) as extra:
|
||
base.pages.extend(extra.pages)
|
||
buf = io.BytesIO()
|
||
base.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def count_pdf_pages(pdf_bytes: bytes) -> int:
|
||
"""Seitenzahl einer PDF. Dient nur einer Anzeige-Meldung und gibt bei unlesbaren Bytes 0
|
||
zurueck, statt den bereits erfolgreichen Export mit einer Exception zu kippen."""
|
||
try:
|
||
with pikepdf.open(io.BytesIO(pdf_bytes)) as pdf:
|
||
return len(pdf.pages)
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
def pdf_has_embedded_facturx(pdf_bytes: bytes) -> bool:
|
||
"""True, wenn die PDF wieder lesbar ist UND die eingebettete 'factur-x.xml' enthaelt. Dient als
|
||
Erfolgskontrolle nach der ZUGFeRD-Erzeugung - factur-x kann bei manchen PDF-Strukturen die XML
|
||
still weglassen bzw. eine kaputte PDF produzieren."""
|
||
try:
|
||
with pikepdf.open(io.BytesIO(pdf_bytes)) as pdf:
|
||
return "factur-x.xml" in pdf.attachments
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
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()
|