Update für API & Docker

This commit is contained in:
dfuhrmann3 2026-07-02 14:19:25 +02:00
parent b9bbf841d9
commit 3bf38958ee
5 changed files with 35 additions and 359 deletions

2
.dockerignore Normal file
View File

@ -0,0 +1,2 @@
venv
models

20
Dockerfile Normal file
View File

@ -0,0 +1,20 @@
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip3 install --no-cache-dir --upgrade pip && \
pip3 install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8040
HEALTHCHECK CMD curl --fail http://localhost:8040/_stcore/health || exit 1
CMD ["streamlit", "run", "main.py", "--server.port=8040", "--server.address=0.0.0.0"]

358
app.py
View File

@ -1,358 +0,0 @@
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
from pydantic import BaseModel, Field
from llama_cpp import Llama, LlamaGrammar
# ==========================================
# 1. 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)")
# ==========================================
# 2. MATHEMATISCHE FUNKTIONEN
# ==========================================
def kaufmaennisches_runden(betrag, nachkommastellen=2):
"""
Rundet Beträge finanzmathematisch exakt nach kaufmännischer Art (ab 5 aufrunden).
Verhindert Pythons Banker's Rounding Fehler.
"""
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"
ET.register_namespace('', 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:xoev-de:kosit:standard:xrechnung_3.0"
ET.SubElement(root, f"{{{CBC}}}ProfileID").text = "urn:fdc:peppol.eu:poacc:bis:invoice:3"
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"
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"])
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=" ")
# ==========================================
# 3. KI MODELL (QWEN 3.5 INSTRUCT)
# ==========================================
def get_cached_llm():
if "llm_instance" not in st.session_state:
with st.spinner("🚀 INITIAL: Stammdaten-Modell wird geladen..."):
base_dir = os.path.dirname(os.path.abspath(__file__))
model_path = os.path.join(base_dir, "models", "Qwen3.5-4B-Instruct-Q8_0.gguf")
st.session_state.llm_instance = Llama(model_path=model_path, n_ctx=2048, n_gpu_layers=-1, n_threads=8)
return st.session_state.llm_instance
# ==========================================
# 4. STREAMLIT FRONTEND
# ==========================================
st.set_page_config(page_title="PDF zu E-Rechnung", layout="wide")
st.title("📄 Enterprise PDF zu E-Rechnung Konverter")
llm = get_cached_llm()
col1, col2 = st.columns(2)
with col1:
st.subheader("1. Dokument hochladen")
uploaded_file = st.file_uploader("Wähle eine Rechnungs-PDF", type=["pdf"])
pdf_text = ""
if uploaded_file is not None:
doc = fitz.open(stream=uploaded_file.read(), filetype="pdf")
for page in doc:
pdf_text += page.get_text("text") + "\n"
st.success("PDF eingelesen!")
with st.expander("Gefundener Text"): st.text(pdf_text)
with col2:
st.subheader("2. Strukturierte Daten validieren")
if "extracted_data" not in st.session_state: st.session_state.extracted_data = None
if uploaded_file is not None and pdf_text.strip() != "":
if st.button("🚀 Daten extrahieren (Kugelsichere Hybrid-Methode)", width="stretch"):
with st.spinner("Extrahiere Daten..."):
# STUFE 1: Stammdaten via KI
grammar_meta = LlamaGrammar.from_json_schema(json.dumps(InvoiceMetadata.model_json_schema()))
prompt_meta = f"""Du bist ein OCR-Parser für deutsche B2B-Rechnungen. Extrahiere die Metadaten fehlerfrei.
GIB DAS JSON REIN ALS EINZEILER AUS. KEINE ZEILENUMBRÜCHE!
Rechnungstext:\n{pdf_text}"""
res_meta = llm(prompt=prompt_meta, max_tokens=512, grammar=grammar_meta, temperature=0.0)
final_data = json.loads(res_meta["choices"][0]["text"])
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))
# Berechnung inklusive kaufmännischem Runden auf Positionsebene
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)
# Kaufmännisches Runden für die Umsatzsteuer
raw_vat_amount = computed_net_amount * (vat_rate / 100.0)
computed_vat_amount = kaufmaennisches_runden(raw_vat_amount)
# Brutto aus den gerundeten Netto- und MwSt-Werten
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")

12
docker-compose.yml Normal file
View File

@ -0,0 +1,12 @@
services:
invoice-converter:
build: .
image: pdf-xrechnung-converter:latest
container_name: invoice-converter-app
ports:
- "8040:8040"
environment:
- VLLM_URL=http://10.192.0.140:8002
- LLM_MODEL=gemma431b
- LLM_BACKEND=vllm
restart: unless-stopped

View File

@ -1,5 +1,5 @@
streamlit==1.58.0
llama_cpp_python==0.3.32
pydantic==2.13.4
pymupdf==1.28.0
requests==2.32.3