diff --git a/invoice_logic.py b/invoice_logic.py index 6e93413..1b0ec4c 100644 --- a/invoice_logic.py +++ b/invoice_logic.py @@ -245,6 +245,108 @@ def extract_referenznummer(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()] + + 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 Kundenadresse) + 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_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 + + 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 + } + + def sanitize_filename_part(value): """Macht einen String dateisystem-sicher: entfernt in Dateinamen kritische Zeichen (Pfadtrenner/Sonderzeichen) und fasst Whitespace zu einem Bindestrich zusammen.""" diff --git a/main.py b/main.py index 120316d..61dd660 100644 --- a/main.py +++ b/main.py @@ -1,13 +1,9 @@ -import os import io import streamlit as st import json -import re import fitz -import requests from facturx import generate_from_binary as facturx_generate_from_binary from invoice_logic import ( - InvoiceMetadata, VERKAEUFER, VERKAEUFER_IBAN, PAYMENT_MEANS_CODES, @@ -35,6 +31,7 @@ from invoice_logic import ( extract_document_references, extract_labeled_header_value, extract_referenznummer, + extract_header_metadata_pure_python, ) # ========================================== @@ -44,58 +41,11 @@ if "invoice_counter" not in st.session_state: st.session_state.invoice_counter = 0 # ========================================== -# 1. API-CLIENT FÜR DAS vLLM-BACKEND +# 1. KOPFDATEN-EXTRAKTION (KI-frei, test-version3) # ========================================== -class LLMClient: - - def __init__(self, model="gemma431b", base_url="http://10.192.0.140:8002"): - self.model = model - - 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} - ] - return self._openai_compatible_request(messages, temperature, json_schema) - - 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}" - - - -# ========================================== -# 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') - st.session_state.llm_instance = LLMClient(model=llm_model, base_url=llm_url) - return st.session_state.llm_instance +# Diese Variante kommt ohne LLM/vLLM-Backend aus: die Kopfdaten werden vollstaendig +# deterministisch aus dem PDF-Text gelesen (siehe extract_header_metadata_pure_python in +# invoice_logic.py). Das ist schneller, offline-faehig und ohne Server-Abhaengigkeit. @st.cache_data(show_spinner=False) @@ -143,7 +93,6 @@ def build_zugferd_pdf(invoice_json: str, original_pdf_bytes: bytes) -> bytes: st.set_page_config(page_title="PDF zu E-Rechnung", layout="wide") st.title("📄 PDF zu E-Rechnung Konverter") -llm = get_cached_llm() col1, col2 = st.columns(2) with col1: @@ -173,63 +122,15 @@ with col2: if uploaded_file is not None and pdf_text.strip() != "": if st.button("Daten extrahieren", width="stretch"): - with st.spinner("Extrahiere Daten über vLLM-API..."): + with st.spinner("Extrahiere Daten..."): - 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" - "- customer_number\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" - "- delivery_note_id\n" - "- order_id\n" - "- contract_id\n" - "- service_start_date\n" - "- service_end_date\n" - "- payment_term_days\n" - "- invoice_note\n\n" - "WICHTIGE REGEL: Analysiere den Text direkt VOR der ersten Tabellenposition '1'. " - "Sortiere alle dort vorlaufenden Zeilen vollständig in die dafür vorgesehenen ID-, Datums- und Freitext-Felder ein. " - "Schreibe diese vorlaufenden Metadaten NIEMALS in die Beschreibung der ersten Position hinein!\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() + # KI-frei (test-version3): Kopfdaten vollstaendig deterministisch aus dem + # PDF-Text lesen - kein vLLM/LLM-Aufruf mehr. Liefert dasselbe Dict-Schema, + # auf das die nachfolgende Pipeline aufsetzt. + final_data = extract_header_metadata_pure_python(pdf_text) + # Sicherheitsnetz: falls doch die eigene scitech-Adresse als Kunden-E-Mail + # erkannt wurde, leeren (BT-49 ist die Adresse des Empfaengers, nicht unsere). if final_data.get("recipient_email") == VERKAEUFER["email"]: final_data["recipient_email"] = ""