#!/usr/bin/env python3 """ extract_buyers_poc.py - POC: extrahiert Buyer-Information-Sheet-Felder aus PDFs. Verbesserte Fassung mit: - HTTP-Timeout (kein unendliches Haengen mehr bei Problem-PDFs) - Seitenlimit gegen riesige Scans (z.B. SALADINO mit 59 Seiten) - hoehere Vision-Aufloesung + Gemma-Vision-Budget fuer Checkbox-Erkennung - Guided Decoding (JSON-Schema erzwungen) mit Loose-Parser als Fallback - Wahl zwischen nur-oberste-Ebene und rekursiv (--recursive) - Sortierung nach (aus Dateiname geparstem) Datum fuer Nachvollziehbarkeit Ablauf pro PDF: 1. Text-Extraktion (pypdf). Genug Text -> Text-Pfad. 2. Zu wenig Text (Scan) -> erste Seiten als Bilder -> Vision-Pfad. 3. VLM/LLM extrahiert Zielfelder als JSON (Schema-guided). 4. Eine Zeile pro PDF -> buyers_raw.jsonl 5. Am Ende: Business-Kategorien -> kategorien_roh.txt Zielfelder verteilen sich ueber ZWEI Seiten (Buyer Information Sheet + Confidentiality Agreement). Fehlt eine Seite -> ihre Felder bleiben null. Aufruf: python extract_buyers_poc.py \ --src "/mnt/bizmatch-nas/AA Buyers NDA's/Buyers NDA's A-Z/S" \ --out ./poc_out \ --api http://192.168.100.160:8000/v1 \ --model "google/gemma-4-12b-it" \ --limit 150 """ import os import io import sys import json import base64 import argparse import glob import re import datetime from openai import OpenAI from pypdf import PdfReader from pdf2image import convert_from_path # --------------------------------------------------------------------------- # Zielschema (fuer Guided Decoding) + Prompt # --------------------------------------------------------------------------- JSON_SCHEMA = { "type": "object", "properties": { "is_buyer_sheet": {"type": "boolean"}, "name_company": {"type": ["string", "null"]}, "prospective_buyer": {"type": ["string", "null"]}, "company": {"type": ["string", "null"]}, "phone": {"type": ["string", "null"]}, "cell": {"type": ["string", "null"]}, "email": {"type": ["string", "null"]}, "address": {"type": ["string", "null"]}, "state": {"type": ["string", "null"]}, "how_did_you_hear": {"type": ["string", "null"]}, "interested_in_updates": {"type": ["boolean", "null"]}, "types_of_business_raw": {"type": ["string", "null"]}, "types_of_business": {"type": "array", "items": {"type": "string"}}, "background_experience": {"type": ["string", "null"]}, "date_of_introduction": {"type": ["string", "null"]}, }, "required": [ "is_buyer_sheet", "name_company", "prospective_buyer", "company", "phone", "cell", "email", "address", "state", "how_did_you_hear", "interested_in_updates", "types_of_business_raw", "types_of_business", "background_experience", "date_of_introduction" ], } SYSTEM_PROMPT = """Du bist ein praezises Datenextraktions-System fuer Formulare der Firma "BizMatch Business Brokerage". Es gibt zwei relevante Seiten: 1. "BUYER INFORMATION SHEET" - enthaelt: NAME/COMPANY, PHONE/FAX/CELL, ADDRESS, EMAIL, "HOW DID YOU HEAR ABOUT US", die Frage "ARE YOU INTERESTED IN UPDATES ON NEW LISTINGS" mit ZWEI KAESTCHEN direkt vor den Woertern YES und NO, "TYPES OF BUSINESSES INTERESTED IN", "BACKGROUND / BUSINESS EXPERIENCE". 2. "CONFIDENTIALITY AGREEMENT" - enthaelt: "Prospective Buyer / Broker / Contact" (der ECHTE Name der Person), "Company (if applicable)", "Date of Introduction". Ignoriere juristischen Vertragstext, Brokerage Services, Dropbox Sign, Unterschriften und alles, was unten nicht verlangt wird. Wenn KEINE dieser beiden BizMatch-Seiten erkennbar ist, setze is_buyer_sheet=false und alle anderen Felder null bzw. []. CHECKBOX-ANWEISUNG (sehr wichtig): Bei "ARE YOU INTERESTED IN UPDATES" stehen zwei kleine Quadrate, je eines direkt VOR "YES" und vor "NO". Untersuche genau, welches Quadrat eine Markierung enthaelt (Haken, X, Kreuz, ausgefuellt, handschriftlich). Nur YES markiert -> true. Nur NO markiert -> false. Keines klar markiert -> null. REGELN fuer "types_of_business": - Extrahiere NUR die Geschaefts-KATEGORIE, nicht Umsatz/Groesse/Zusatz. - "High Volume Restaurants sales 5Mil" -> ["Restaurant"] - "Oil field service company, trucking" -> ["Oil Field", "Trucking"] - "any profitable business" -> ["Any"] - Singular, knapp, englisch. Leer/unklar -> []. REGELN allgemein: - Fehlendes/leeres Feld -> null (bzw. [] fuer Listen). - ERFINDE NICHTS. Nur was tatsaechlich im Dokument steht. - Bei "date_of_introduction": nach Moeglichkeit Format YYYY-MM-DD, sonst Originaltext, sonst null. - Antworte NUR mit dem JSON-Objekt. """ USER_TEXT_INTRO = "Hier ist das Dokument als extrahierter Text:\n\n" USER_IMG_INTRO = ("Hier sind die ersten Seiten des gescannten Dokuments. Achte " "besonders auf die angekreuzte YES/NO-Checkbox. Extrahiere die Felder:") # --------------------------------------------------------------------------- # Datum aus Dateinamen parsen (fuer Sortierung + als Datums-Fallback) # --------------------------------------------------------------------------- def date_from_filename(path): """ Sucht ein Datum im Dateinamen. Typische Muster: 120623, 030924, 071923 (MMDDYY) sowie 2024-01-01 o.ae. Gibt (datetime.date | None). """ name = os.path.basename(path) # ISO zuerst m = re.search(r"(20\d{2})[-_](\d{2})[-_](\d{2})", name) if m: try: return datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3))) except ValueError: pass # 6-stellig MMDDYY (haeufigstes Muster hier) for m in re.finditer(r"(? groesseres Datum = kleinerer Schluessel y, m, d = (int(x) for x in iso.split("-")) return -datetime.date(y, m, d).toordinal() def sort_key(path): d = date_from_filename(path) if d: # Gruppe 0 (mit Datum), neueste zuerst return (0, _invert_date(d.isoformat()), path) # ohne erkanntes Datum: nach mtime (neueste zuerst), ans Ende try: mt = -int(os.path.getmtime(path)) except OSError: mt = 0 return (1, mt, path) # --------------------------------------------------------------------------- # PDF-Handling # --------------------------------------------------------------------------- def pdf_page_count(path): try: return len(PdfReader(path).pages) except Exception: return -1 def extract_text(pdf_path, max_pages=4): try: reader = PdfReader(pdf_path) pages = reader.pages[:max_pages] return "\n".join(p.extract_text() or "" for p in pages) except Exception: return "" def pdf_to_base64_images(pdf_path, first_page=1, max_pages=2, max_dim=1600, quality=80, dpi=150): """ Seiten [first_page .. first_page+max_pages-1] -> Base64-JPEG. Moderate Aufloesung (dpi=150, max_dim=1600): reicht fuer Checkbox/Handschrift. first_page erlaubt das Ueberspringen einer fuehrenden Seite (z.B. die ueberfluessige handschriftliche Notiz-Seite bei "Notes"-Dateien). """ last_page = first_page + max_pages - 1 try: images = convert_from_path(pdf_path, first_page=first_page, last_page=last_page, dpi=dpi) except Exception as e: print(f" [Bildkonvertierung-Fehler] {e}") return [] out = [] for img in images: w, h = img.size scale = min(1.0, max_dim / max(w, h)) if scale < 1.0: img = img.resize((int(w * scale), int(h * scale))) buf = io.BytesIO() img.convert("RGB").save(buf, format="JPEG", quality=quality) out.append(base64.b64encode(buf.getvalue()).decode("utf-8")) return out # --------------------------------------------------------------------------- # LLM-Call (mit Guided Decoding + Vision-Budget) # --------------------------------------------------------------------------- def call_model(client, model, content_payload, vision_budget=None, max_tokens=768): # llama.cpp erzwingt JSON ueber response_format mit json_schema. # WICHTIG gegen Gemma-4 "-Flood"/Runaway: # - enable_thinking:false per Request (zuverlaessiger als nur Server-Flag) # - max_tokens moderat (768: genug fuer volle Datensaetze mit langem # Background/vielen Kategorien, aber begrenzt genug gegen Runaway) # - repeat_penalty gegen Wiederholschleifen resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": content_payload}, ], temperature=0.0, max_tokens=max_tokens, response_format={ "type": "json_schema", "json_schema": {"name": "buyer_sheet", "schema": JSON_SCHEMA}, }, extra_body={ "chat_template_kwargs": {"enable_thinking": False}, "repeat_penalty": 1.05, }, ) choice = resp.choices[0] content = choice.message.content # finish_reason "length" = Runaway (max_tokens erreicht) -> als Warnung markieren if getattr(choice, "finish_reason", None) == "length": return content, "length" return content, "stop" def parse_json_loose(txt): if not txt: return None txt = txt.strip() txt = re.sub(r"^```(?:json)?|```$", "", txt, flags=re.MULTILINE).strip() start, end = txt.find("{"), txt.rfind("}") if start == -1 or end == -1 or end < start: return None try: return json.loads(txt[start:end + 1]) except json.JSONDecodeError: return None # Seitenlimits TEXT_SCAN_THRESHOLD = 120 # < so viele Textzeichen -> als Scan behandeln MAX_SCAN_PAGES_TOTAL = 12 # PDFs mit mehr Seiten gelten als "kein reines Sheet" # Felder, bei denen wir Vision bevorzugen (handschriftlich/visuell auf dem Formular) _PREFER_VISION = { "prospective_buyer", "name_company", "company", "phone", "cell", "email", "address", "state", "how_did_you_hear", "interested_in_updates", "date_of_introduction", } # Felder, bei denen Text meist die sauberere (getippte) Quelle ist _PREFER_TEXT = {"background_experience", "types_of_business", "types_of_business_raw"} def _empty(v): return v in (None, "", [], {}) def merge_records(text_rec, vision_rec): """ Fuehrt Text- und Vision-Extraktion desselben PDF zusammen. Grundregel: nicht-leerer Wert schlaegt leeren; bei Konflikt entscheidet die bevorzugte Quelle je Feld (Vision fuer Handschrift, Text fuer Getipptes). """ if text_rec is None: return vision_rec if vision_rec is None: return text_rec merged = {} keys = set(text_rec) | set(vision_rec) for k in keys: tv, vv = text_rec.get(k), vision_rec.get(k) if _empty(tv) and _empty(vv): merged[k] = tv if k in text_rec else vv elif _empty(tv): merged[k] = vv elif _empty(vv): merged[k] = tv else: # beide gefuellt -> bevorzugte Quelle if k in _PREFER_VISION: merged[k] = vv elif k in _PREFER_TEXT: merged[k] = tv else: merged[k] = vv # Default: Vision return merged def _extract_via_text(client, model, text): payload = USER_TEXT_INTRO + text[:12000] raw, finish = call_model(client, model, payload) data = parse_json_loose(raw) if data is not None and finish == "length": data["_runaway"] = True # Antwort war abgeschnitten -> unvollstaendig moeglich return data def _extract_via_vision(client, model, pdf_path, max_pages, first_page=1): imgs = pdf_to_base64_images(pdf_path, first_page=first_page, max_pages=max_pages) if not imgs: return None payload = [{"type": "text", "text": USER_IMG_INTRO}] for b64 in imgs: payload.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}) raw, finish = call_model(client, model, payload) data = parse_json_loose(raw) if data is not None and finish == "length": data["_runaway"] = True return data def process_pdf(client, model, pdf_path, vision_budget=None): """ Hybrid-Strategie mit Notes-Seitenlogik: - "Notes"-Dateien: Seite 1 ist eine ueberfluessige handschriftliche Notiz. Die Buyer-Felder stehen auf Seite 2 (Info Sheet) + 3 (CA mit Datum/Name). Darum bei Notes-Dateien first_page=2. - Normale Dateien: Info Sheet = Seite 1, CA = Seite 2. first_page=1. - reiner Scan (keine Textebene): nur Vision - grosser Scan (> MAX_SCAN_PAGES_TOTAL): nur erste relevante Seite pruefen - getipptes PDF mit Textebene: Text UND Vision, dann mergen """ n_pages = pdf_page_count(pdf_path) text = extract_text(pdf_path) has_text = len(text.strip()) >= TEXT_SCAN_THRESHOLD # "Notes" im Dateinamen -> fuehrende Notiz-Seite ueberspringen. # Aber nur, wenn genug Seiten da sind (sonst normal ab Seite 1). is_notes = "notes" in os.path.basename(pdf_path).lower() vpage = 2 if (is_notes and n_pages >= 3) else 1 truncated_note = None try: if not has_text and n_pages > MAX_SCAN_PAGES_TOTAL: # grosser reiner Scan: nur die erste relevante Seite pruefen truncated_note = f"grosser Scan ({n_pages} Seiten), begrenzt geprueft" data = _extract_via_vision(client, model, pdf_path, max_pages=2, first_page=vpage) mode = "vision" elif not has_text: # reiner Scan data = _extract_via_vision(client, model, pdf_path, max_pages=2, first_page=vpage) mode = "vision" else: # Textebene vorhanden -> HYBRID: beide Pfade, dann mergen text_rec = _extract_via_text(client, model, text) vision_rec = None if n_pages <= MAX_SCAN_PAGES_TOTAL: vision_rec = _extract_via_vision(client, model, pdf_path, max_pages=2, first_page=vpage) data = merge_records(text_rec, vision_rec) mode = "hybrid" if vision_rec is not None else "text" except Exception as e: return {"_error": f"API: {type(e).__name__}: {e}", "n_pages": n_pages}, "?" if data is None: return {"_error": "JSON-Parse fehlgeschlagen", "n_pages": n_pages}, mode data["n_pages"] = n_pages if is_notes: data["_notes_file"] = True if truncated_note: data["_note"] = truncated_note return data, mode # --------------------------------------------------------------------------- # Hauptlauf # --------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser() ap.add_argument("--src", required=True) ap.add_argument("--out", default="./poc_out") ap.add_argument("--api", default="http://192.168.100.160:8000/v1") ap.add_argument("--model", default="gemma-4-12b") # --alias des llama-server ap.add_argument("--limit", type=int, default=150) ap.add_argument("--recursive", action="store_true", help="auch Unterordner durchsuchen (Default: nur oberste Ebene)") ap.add_argument("--timeout", type=float, default=60, help="HTTP-Timeout je Anfrage in Sekunden") ap.add_argument("--vision-budget", type=int, default=0, help="Gemma Vision-Token-Budget (0=aus/Default 280; sonst 70/140/280/560/1120). " "ACHTUNG: hohe Werte koennen auf manchen ROCm-Builds das Bild zerstoeren.") args = ap.parse_args() os.makedirs(args.out, exist_ok=True) jsonl_path = os.path.join(args.out, "buyers_raw.jsonl") cats_path = os.path.join(args.out, "kategorien_roh.txt") # Timeout am Client verhindert unendliches Haengen client = OpenAI(base_url=args.api, api_key="test-key", timeout=args.timeout, max_retries=1) if args.recursive: pdfs = glob.glob(os.path.join(args.src, "**", "*.pdf"), recursive=True) else: pdfs = glob.glob(os.path.join(args.src, "*.pdf")) # nach (Dateiname-)Datum sortieren pdfs = sorted(pdfs, key=sort_key) if args.limit: pdfs = pdfs[:args.limit] print(f"{len(pdfs)} PDFs im POC-Umfang (sortiert nach Datum, " f"{'rekursiv' if args.recursive else 'nur oberste Ebene'}).\n") all_categories = {} n_ok = n_err = n_text = n_vision = n_hybrid = n_nosheet = 0 with open(jsonl_path, "w", encoding="utf-8") as jf: for i, pdf in enumerate(pdfs, 1): rel = os.path.relpath(pdf, args.src) fdate = date_from_filename(pdf) print(f"[{i}/{len(pdfs)}] {fdate or '????-??-??'} {rel}") try: data, mode = process_pdf(client, args.model, pdf, args.vision_budget) # Bei Fehler EINMAL erneut versuchen (faengt seltene # nicht-deterministische Parse-Ausrutscher ab). if "_error" in data: print(f" (Fehler, ein Wiederholversuch...)") data, mode = process_pdf(client, args.model, pdf, args.vision_budget) except KeyboardInterrupt: print("\nAbbruch durch Nutzer. Bisheriges ist gesichert.") break except Exception as e: data, mode = {"_error": f"unerwartet: {type(e).__name__}: {e}"}, "?" n_text += (mode == "text") n_vision += (mode == "vision") n_hybrid += (mode == "hybrid") record = {"source_file": rel, "file_date": fdate.isoformat() if fdate else None, "extraction_mode": mode, **data} jf.write(json.dumps(record, ensure_ascii=False) + "\n") jf.flush() if "_error" in data: n_err += 1 print(f" FEHLER: {data['_error']}") continue n_ok += 1 if data.get("is_buyer_sheet") is False: n_nosheet += 1 for cat in (data.get("types_of_business") or []): key = cat.strip().lower() if key: all_categories[key] = all_categories.get(key, 0) + 1 with open(cats_path, "w", encoding="utf-8") as cf: for cat, cnt in sorted(all_categories.items(), key=lambda x: -x[1]): cf.write(f"{cnt:4d} {cat}\n") print("\n=== POC-Zusammenfassung ===") print(f" verarbeitet: {n_ok + n_err}") print(f" erfolgreich: {n_ok}") print(f" Fehler: {n_err}") print(f" Text-Pfad: {n_text}") print(f" Vision-Pfad: {n_vision}") print(f" Hybrid-Pfad: {n_hybrid}") print(f" kein Buyer-Sheet: {n_nosheet}") print(f" distinkte Kategorien (roh): {len(all_categories)}") print(f"\n JSONL: {jsonl_path}") print(f" Kategorien: {cats_path}") if __name__ == "__main__": main()