dfuhrmann
/
E-Rechnung
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
E-Rechnung/main.py

464 lines
23 KiB
Python
Raw Blame History

import os
import streamlit as st
import json
import re
import xml.etree.ElementTree as ET
from xml.dom import minidom
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_HALF_UP
import fitz
import requests
from pydantic import BaseModel, Field
# ==========================================
# 1. API-CLIENT FÜR EXTERNE BACKENDS (vLLM / Ollama)
# ==========================================
class LLMClient:
"""Client für Ollama und vLLM (OpenAI-kompatibel) mit JSON-Schema-Unterstützung."""
def __init__(self, backend_type="vllm", model="gemma431b", base_url=None):
self.backend_type = backend_type.lower()
self.model = model
if not base_url:
base_url = "http://10.192.0.140:8002" if self.backend_type == "vllm" else "http://localhost:11434"
# KUGELSICHERER REINIGER: Filtert versehentliche Markdown-Links [...] vorab heraus
base_url = base_url.strip()
if base_url.startswith("[") and ")" in base_url:
match = re.search(r"\((https?://.*?)\)", base_url)
if match:
base_url = match.group(1)
self.base_url = base_url.rstrip("/")
def chat(self, prompt, system_prompt="You are a helpful assistant.", temperature=0.0, json_schema=None):
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
]
if self.backend_type == "ollama":
return self._ollama_request(messages, temperature, json_schema)
else:
return self._openai_compatible_request(messages, temperature, json_schema)
def _ollama_request(self, messages, temperature, json_schema):
url = f"{self.base_url}/api/chat"
payload = {
"model": self.model,
"messages": messages,
"stream": False,
"options": {
"temperature": temperature
}
}
if json_schema:
payload["format"] = json_schema
try:
response = requests.post(url, json=payload, timeout=120)
response.raise_for_status()
return response.json().get("message", {}).get("content", "")
except Exception as e:
return f"Ollama Error: {e}"
def _openai_compatible_request(self, messages, temperature, json_schema):
url = f"{self.base_url}/v1/chat/completions"
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature
}
if json_schema:
payload["guided_json"] = json.dumps(json_schema) if isinstance(json_schema, dict) else json_schema
try:
response = requests.post(url, json=payload, timeout=120)
response.raise_for_status()
data = response.json()
return data.get("choices", [{}])[0].get("message", {}).get("content", "")
except Exception as e:
return f"vLLM/OpenAI Error: {e}"
# ==========================================
# 2. METADATEN SCHREIBSCHEMA
# ==========================================
class InvoiceMetadata(BaseModel):
invoice_number: str = Field(description="Die Rechnungsnummer nach 'Nummer:' (z.B. R26-219)")
invoice_date: str = Field(description="Das Rechnungsdatum nach 'Datum:' im Format YYYY-MM-DD")
delivery_date: str = Field(description="Das Lieferdatum nach 'Lieferdatum:' im Format YYYY-MM-DD")
buyer_reference: str = Field(description="Der exakte Wert nach 'Ihr Auftrag:' (z.B. 05.06.2026-Herr.Knospe)")
recipient_name: str = Field(description="Vollständige Firma des Kunden/Empfängers (z.B. Stadtwerke Hennigsdorf GmbH)")
recipient_street: str = Field(description="Straße und Hausnummer des Kunden")
recipient_zip: str = Field(description="Postleitzahl (PLZ) des Kunden")
recipient_city: str = Field(description="Stadt des Kunden")
recipient_email: str = Field(description="E-Mail-Adresse des Kunden (LEER lassen, wenn nicht explizit im Text!)")
vat_rate_percent: float = Field(description="Der primäre Umsatzsteuersatz in Prozent (z.B. 19.0)")
# ==========================================
# 3. MATHEMATISCHE FUNKTIONEN
# ==========================================
def kaufmaennisches_runden(betrag, nachkommastellen=2):
if not betrag:
return 0.0
quantizer = Decimal('1.' + '0' * nachkommastellen)
return float(Decimal(f"{betrag:.6f}").quantize(quantizer, rounding=ROUND_HALF_UP))
def clean_german_float(val_str):
if not val_str:
return 0.0
s = str(val_str).strip().replace("€", "").replace(" ", "").replace("'", "").replace("\n", "").replace("\r", "")
if not s:
return 0.0
if "." in s and "," in s:
s = s.replace(".", "").replace(",", ".")
elif "," in s:
s = s.replace(",", ".")
try:
return float(s)
except ValueError:
return 0.0
def extract_invoice_items_pure_python(text):
items = []
lines = [line.strip() for line in text.split("\n") if line.strip()]
unit_pattern = re.compile(r"^(\d+(?:[\.,]\d+)?)\s*(Tag\(e\)|Std\.|Stk\.|Stunden|Stück|stk|stck|stück)$", re.IGNORECASE)
price_stop_pattern = re.compile(r"^[\d\.]+,[\d]{2}$")
for i, line in enumerate(lines):
match = unit_pattern.match(line)
if match:
raw_qty = match.group(1)
raw_unit = match.group(2)
raw_price = "0,00"
for k in range(1, 3):
if i + k < len(lines):
candidate = lines[i+k].replace("€", "").replace(" ", "").strip()
if re.match(r"^[\d.,]+$", candidate):
raw_price = candidate
break
desc_lines = []
for j in range(i - 1, -1, -1):
prev_line = lines[j]
if unit_pattern.match(prev_line) or price_stop_pattern.match(prev_line):
break
prev_lower = prev_line.lower()
if prev_lower in ["pos.", "pos", "bezeichnung", "beschreibung", "menge", "einheit", "einzelpreis", "gesamtpreis", "betrag €", "betrag", "artikel"]:
break
if any(prev_lower.startswith(w) for w in ["zwischensumme", "endbetrag", "gesamtbetrag", "umsatzsteuer"]):
break
if prev_line.isdigit() and len(prev_line) <= 2:
continue
desc_lines.insert(0, prev_line)
description = " ".join(desc_lines).strip()
if description:
items.append({
"description": description,
"quantity": clean_german_float(raw_qty),
"unit": raw_unit,
"unit_price": clean_german_float(raw_price)
})
return items
def map_unece_unit_code(unit_str):
u = str(unit_str).lower()
if "tag" in u or "day" in u: return "DAY"
elif "std" in u or "stund" in u or "hour" in u or "hur" in u: return "HUR"
return "C62"
def parse_to_strict_xml_date(date_str):
clean_str = str(date_str).strip()
for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d-%m-%Y"):
try: return datetime.strptime(clean_str, fmt).strftime("%Y-%m-%d")
except ValueError: continue
return datetime.today().strftime("%Y-%m-%d")
def generate_compliant_xrechnung(data):
INVOICE_NS = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
CBC = "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2"
CAC = "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
# Namespaces explizit registrieren
ET.register_namespace('ubl', INVOICE_NS)
ET.register_namespace('cbc', CBC)
ET.register_namespace('cac', CAC)
root = ET.Element(f"{{{INVOICE_NS}}}Invoice")
# Offizielle Profil-IDs für XRechnung 3.0 nach Kosit-Vorgabe
ET.SubElement(root, f"{{{CBC}}}CustomizationID").text = "urn:cen.eu:en16931:2017#compliant#urn:xeinkauf.de:kosit:xrechnung_3.0"
ET.SubElement(root, f"{{{CBC}}}ProfileID").text = "urn:fdc:peppol.eu:2017:poacc:billing:01:1.0"
ET.SubElement(root, f"{{{CBC}}}ID").text = str(data["invoice_number"])
ET.SubElement(root, f"{{{CBC}}}IssueDate").text = parse_to_strict_xml_date(data["invoice_date"])
try:
base_date = datetime.strptime(parse_to_strict_xml_date(data["invoice_date"]), "%Y-%m-%d")
computed_due = (base_date + timedelta(days=14)).strftime("%Y-%m-%d")
except Exception:
computed_due = (datetime.today() + timedelta(days=14)).strftime("%Y-%m-%d")
ET.SubElement(root, f"{{{CBC}}}DueDate").text = computed_due
ET.SubElement(root, f"{{{CBC}}}InvoiceTypeCode").text = "380"
ET.SubElement(root, f"{{{CBC}}}DocumentCurrencyCode").text = "EUR"
ET.SubElement(root, f"{{{CBC}}}BuyerReference").text = str(data["buyer_reference"])
vat_rate = float(data['vat_rate_percent'])
vat_category = "S" if vat_rate > 0.0 else "E"
# --- SENDER (SCITECH) ---
supplier = ET.SubElement(root, f"{{{CAC}}}AccountingSupplierParty")
party_supp = ET.SubElement(supplier, f"{{{CAC}}}Party")
ET.SubElement(party_supp, f"{{{CBC}}}EndpointID", {"schemeID": "EM"}).text = "office@scitech-gmbh.de"
party_name_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyName")
ET.SubElement(party_name_supp, f"{{{CBC}}}Name").text = "scitech it solutions GmbH"
addr_supp = ET.SubElement(party_supp, f"{{{CAC}}}PostalAddress")
ET.SubElement(addr_supp, f"{{{CBC}}}StreetName").text = "Hermann-Schumann-Str. 4"
ET.SubElement(addr_supp, f"{{{CBC}}}CityName").text = "Hennigsdorf"
ET.SubElement(addr_supp, f"{{{CBC}}}PostalZone").text = "16761"
country_supp = ET.SubElement(addr_supp, f"{{{CAC}}}Country")
ET.SubElement(country_supp, f"{{{CBC}}}IdentificationCode").text = "DE"
tax_scheme_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyTaxScheme")
ET.SubElement(tax_scheme_supp, f"{{{CBC}}}CompanyID").text = "DE218436692"
scheme_supp = ET.SubElement(tax_scheme_supp, f"{{{CAC}}}TaxScheme")
ET.SubElement(scheme_supp, f"{{{CBC}}}ID").text = "VAT"
legal_supp = ET.SubElement(party_supp, f"{{{CAC}}}PartyLegalEntity")
ET.SubElement(legal_supp, f"{{{CBC}}}RegistrationName").text = "scitech it solutions GmbH"
# KORREKTUR: Saubere, valide Erzeugung des Contact-Blocks mit cbc-Namespaces
contact_supp = ET.SubElement(party_supp, f"{{{CAC}}}Contact")
ET.SubElement(contact_supp, f"{{{CBC}}}Name").text = "Geschäftsführung"
ET.SubElement(contact_supp, f"{{{CBC}}}Telephone").text = "+49 3302 4940537"
ET.SubElement(contact_supp, f"{{{CBC}}}ElectronicMail").text = "office@scitech-gmbh.de"
# --- RECIPIENT ---
customer = ET.SubElement(root, f"{{{CAC}}}AccountingCustomerParty")
party_cust = ET.SubElement(customer, f"{{{CAC}}}Party")
ET.SubElement(party_cust, f"{{{CBC}}}EndpointID", {"schemeID": "EM"}).text = str(data["recipient_email"]) if data.get("recipient_email") else "rechnung@kunde.de"
party_name_cust = ET.SubElement(party_cust, f"{{{CAC}}}PartyName")
ET.SubElement(party_name_cust, f"{{{CBC}}}Name").text = str(data["recipient_name"])
addr_cust = ET.SubElement(party_cust, f"{{{CAC}}}PostalAddress")
ET.SubElement(addr_cust, f"{{{CBC}}}StreetName").text = str(data["recipient_street"])
ET.SubElement(addr_cust, f"{{{CBC}}}CityName").text = str(data["recipient_city"])
ET.SubElement(addr_cust, f"{{{CBC}}}PostalZone").text = str(data["recipient_zip"])
country_cust = ET.SubElement(addr_cust, f"{{{CAC}}}Country")
ET.SubElement(country_cust, f"{{{CBC}}}IdentificationCode").text = "DE"
legal_cust = ET.SubElement(party_cust, f"{{{CAC}}}PartyLegalEntity")
ET.SubElement(legal_cust, f"{{{CBC}}}RegistrationName").text = str(data["recipient_name"])
if data.get("delivery_date"):
delivery = ET.SubElement(root, f"{{{CAC}}}Delivery")
ET.SubElement(delivery, f"{{{CBC}}}ActualDeliveryDate").text = parse_to_strict_xml_date(data["delivery_date"])
payment_means = ET.SubElement(root, f"{{{CAC}}}PaymentMeans")
ET.SubElement(payment_means, f"{{{CBC}}}PaymentMeansCode").text = "30"
payee_acc = ET.SubElement(payment_means, f"{{{CAC}}}PayeeFinancialAccount")
ET.SubElement(payee_acc, f"{{{CBC}}}ID").text = str(data["iban"]).replace(" ", "")
tax_total = ET.SubElement(root, f"{{{CAC}}}TaxTotal")
ET.SubElement(tax_total, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}"
tax_subtotal = ET.SubElement(tax_total, f"{{{CAC}}}TaxSubtotal")
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxableAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}"
ET.SubElement(tax_subtotal, f"{{{CBC}}}TaxAmount", {"currencyID": "EUR"}).text = f"{data['total_vat_amount']:.2f}"
tax_cat = ET.SubElement(tax_subtotal, f"{{{CAC}}}TaxCategory")
ET.SubElement(tax_cat, f"{{{CBC}}}ID").text = vat_category
ET.SubElement(tax_cat, f"{{{CBC}}}Percent").text = f"{vat_rate:.2f}"
tax_scheme = ET.SubElement(tax_cat, f"{{{CAC}}}TaxScheme")
ET.SubElement(tax_scheme, f"{{{CBC}}}ID").text = "VAT"
monetary_total = ET.SubElement(root, f"{{{CAC}}}LegalMonetaryTotal")
ET.SubElement(monetary_total, f"{{{CBC}}}LineExtensionAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}TaxExclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_net_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}TaxInclusiveAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}"
ET.SubElement(monetary_total, f"{{{CBC}}}PayableAmount", {"currencyID": "EUR"}).text = f"{data['total_gross_amount']:.2f}"
for index, item in enumerate(data["items"], start=1):
line = ET.SubElement(root, f"{{{CAC}}}InvoiceLine")
ET.SubElement(line, f"{{{CBC}}}ID").text = str(index)
u_code = map_unece_unit_code(item.get("unit", "C62"))
q_val = float(item.get("quantity") or 0.0)
qty_formatted = f"{q_val:.4f}".rstrip('0').rstrip('.') if not q_val.is_integer() else f"{q_val:.2f}"
if qty_formatted == "" or qty_formatted == "0": qty_formatted = "0.00"
ET.SubElement(line, f"{{{CBC}}}InvoicedQuantity", {"unitCode": u_code}).text = qty_formatted
ET.SubElement(line, f"{{{CBC}}}LineExtensionAmount", {"currencyID": "EUR"}).text = f"{item['total_amount']:.2f}"
item_elem = ET.SubElement(line, f"{{{CAC}}}Item")
ET.SubElement(item_elem, f"{{{CBC}}}Name").text = str(item["description"])
# Saubere, UBL-konforme Schachtelung der Steuerkategorie innerhalb des Items
item_tax = ET.SubElement(item_elem, f"{{{CAC}}}ClassifiedTaxCategory")
ET.SubElement(item_tax, f"{{{CBC}}}ID").text = vat_category
ET.SubElement(item_tax, f"{{{CBC}}}Percent").text = f"{vat_rate:.2f}"
item_scheme = ET.SubElement(item_tax, f"{{{CAC}}}TaxScheme")
ET.SubElement(item_scheme, f"{{{CBC}}}ID").text = "VAT"
price_elem = ET.SubElement(line, f"{{{CAC}}}Price")
ET.SubElement(price_elem, f"{{{CBC}}}PriceAmount", {"currencyID": "EUR"}).text = f"{item['unit_price']:.2f}"
xml_str = ET.tostring(root, encoding="utf-8")
return minidom.parseString(xml_str).toprettyxml(indent=" ")
# ==========================================
# 4. API CLIENT INITIALISIERUNG
# ==========================================
def get_cached_llm():
if "llm_instance" not in st.session_state:
llm_url = os.environ.get('VLLM_URL', 'http://10.192.0.140:8002')
llm_model = os.environ.get('LLM_MODEL', 'gemma431b')
llm_backend = os.environ.get('LLM_BACKEND', 'vllm')
st.session_state.llm_instance = LLMClient(backend_type=llm_backend, model=llm_model, base_url=llm_url)
return st.session_state.llm_instance
# ==========================================
# 5. STREAMLIT FRONTEND
# ==========================================
st.set_page_config(page_title="PDF zu E-Rechnung", layout="wide")
st.title("📄 Enterprise PDF zu E-Rechnung Konverter (API-Version)")
llm = get_cached_llm()
col1, col2 = st.columns(2)
with col1:
st.subheader("1. Dokument hochladen")
uploaded_file = st.file_uploader("Wähle eine Rechnungs-PDF", type=["pdf"])
pdf_text = ""
if uploaded_file is not None:
doc = fitz.open(stream=uploaded_file.read(), filetype="pdf")
for page in doc:
pdf_text += page.get_text("text") + "\n"
st.success("PDF eingelesen!")
with st.expander("Gefundener Text"): st.text(pdf_text)
with col2:
st.subheader("2. Strukturierte Daten validieren")
if "extracted_data" not in st.session_state: st.session_state.extracted_data = None
if uploaded_file is not None and pdf_text.strip() != "":
if st.button("🚀 Daten extrahieren (Kugelsichere Hybrid-Methode via API)", width="stretch"):
with st.spinner("Extrahiere Daten über vLLM-API..."):
# STUFE 1: Stammdaten via API mit Schema-Zwang
schema_dict = InvoiceMetadata.model_json_schema()
sys_prompt = (
"Du bist ein präziser OCR-Parser für deutsche B2B-Rechnungen. "
"Extrahiere die Metadaten fehlerfrei als valides JSON-Objekt. "
"Das JSON MUSS zwingend exakt folgende Schlüssel (Keys) enthalten:\n"
"- invoice_number\n"
"- invoice_date (im Format YYYY-MM-DD)\n"
"- delivery_date (im Format YYYY-MM-DD)\n"
"- buyer_reference\n"
"- recipient_name\n"
"- recipient_street\n"
"- recipient_zip\n"
"- recipient_city\n"
"- recipient_email\n"
"- vat_rate_percent (als Zahl, z.B. 19.0)\n\n"
"Antworte ausschließlich mit dem reinen JSON-Objekt. Schreibe keinen anderen Text."
)
prompt_meta = f"Rechnungstext:\n{pdf_text}"
res_meta_text = llm.chat(
prompt=prompt_meta,
system_prompt=sys_prompt,
temperature=0.0,
json_schema=schema_dict
)
clean_json_text = res_meta_text.strip()
backtick_trigger = chr(96) * 3
if clean_json_text.startswith(backtick_trigger):
lines = clean_json_text.splitlines()
if lines[0].startswith(backtick_trigger):
lines = lines[1:]
if lines and lines[-1].startswith(backtick_trigger):
lines = lines[:-1]
clean_json_text = "\n".join(lines).strip()
try:
final_data = json.loads(clean_json_text)
except Exception as e:
st.error(f"Fehler beim Parsen der API-Antwort: {e}")
st.code(res_meta_text)
st.stop()
if final_data.get("recipient_email") == "office@scitech-gmbh.de":
final_data["recipient_email"] = ""
# STUFE 2: Zeilenbasiertes Python-Parsing für die Tabelle
final_data["items"] = extract_invoice_items_pure_python(pdf_text)
st.session_state.extracted_data = final_data
if st.session_state.extracted_data:
data = st.session_state.extracted_data
inv_num = st.text_input("Rechnungsnummer", value=data.get("invoice_number", ""))
inv_date = st.text_input("Rechnungsdatum (YYYY-MM-DD)", value=data.get("invoice_date", ""))
inv_delivery = st.text_input("Lieferdatum (YYYY-MM-DD)", value=data.get("delivery_date", ""))
buyer_ref = st.text_input("Ihr Auftrag / Referenz", value=data.get("buyer_reference", ""))
c_s, c_r = st.columns(2)
with c_s:
st.markdown("**Absender (Verkäufer - Fest hinterlegt):**")
st.text_input("Firma (Sender)", value="scitech it solutions GmbH", disabled=True)
st.text_input("E-Mail (Sender)", value="office@scitech-gmbh.de", disabled=True)
st.text_input("Telefon (Sender)", value="+49 3302 4940537", disabled=True)
st.text_input("Straße (Sender)", value="Hermann-Schumann-Str. 4", disabled=True)
st.text_input("PLZ (Sender)", value="16761", disabled=True)
st.text_input("Stadt (Sender)", value="Hennigsdorf", disabled=True)
with c_r:
st.markdown("**Empfänger (Kunde):**")
r_name = st.text_input("Firma (Kunde)", value=data.get("recipient_name", ""))
r_email = st.text_input("E-Mail (Kunde)", value=data.get("recipient_email", ""))
r_street = st.text_input("Straße (Kunde)", value=data.get("recipient_street", ""))
r_zip = st.text_input("PLZ (Kunde)", value=data.get("recipient_zip", ""))
r_city = st.text_input("Stadt (Kunde)", value=data.get("recipient_city", ""))
st.markdown("**3. Positionen:**")
edited_items = st.data_editor(data.get("items", []), num_rows="dynamic", width="stretch")
vat_rate = st.number_input("Steuersatz (%)", value=float(data.get("vat_rate_percent") or 19.0))
for item in edited_items:
raw_total = float(item.get("quantity") or 0.0) * float(item.get("unit_price") or 0.0)
item["total_amount"] = kaufmaennisches_runden(raw_total)
computed_net_amount = sum(item["total_amount"] for item in edited_items)
raw_vat_amount = computed_net_amount * (vat_rate / 100.0)
computed_vat_amount = kaufmaennisches_runden(raw_vat_amount)
computed_gross_amount = kaufmaennisches_runden(computed_net_amount + computed_vat_amount)
st.markdown("**Berechnete Summen:**")
c_net_ui, c_vat_ui, c_gross_ui = st.columns(3)
with c_net_ui: st.metric("Netto-Gesamt (€)", f"{computed_net_amount:.2f}")
with c_vat_ui: st.metric("MwSt-Gesamt (€)", f"{computed_vat_amount:.2f}")
with c_gross_ui: st.metric("Brutto-Zahlbetrag (€)", f"{computed_gross_amount:.2f}")
st.markdown("---")
if st.button("🔒 XRechnung exportieren", type="primary", width="stretch"):
final_invoice_dict = {
"invoice_number": inv_num, "invoice_date": inv_date, "delivery_date": inv_delivery,
"buyer_reference": buyer_ref, "recipient_name": r_name, "recipient_email": r_email,
"recipient_street": r_street, "recipient_zip": r_zip, "recipient_city": r_city,
"items": edited_items, "total_net_amount": computed_net_amount, "total_vat_amount": computed_vat_amount,
"vat_rate_percent": vat_rate, "total_gross_amount": computed_gross_amount, "iban": "DE06120700240456207000"
}
st.download_button(label="📥 Download XML", data=generate_compliant_xrechnung(final_invoice_dict), file_name=f"XRechnung_{inv_num}.xml", mime="text/xml")
Reference in New Issue View Git Blame Copy Permalink