""" PDF-Processor Modul Text-Extraktion, OCR und ZUGFeRD-Erkennung """ import subprocess from pathlib import Path from typing import Dict, Optional, Tuple import logging import re logger = logging.getLogger(__name__) # Versuche Libraries zu importieren try: import pdfplumber PDFPLUMBER_AVAILABLE = True except ImportError: PDFPLUMBER_AVAILABLE = False logger.warning("pdfplumber nicht installiert") try: from pypdf import PdfReader PYPDF_AVAILABLE = True except ImportError: PYPDF_AVAILABLE = False logger.warning("pypdf nicht installiert") class PDFProcessor: """Verarbeitet PDFs: Text-Extraktion, OCR, ZUGFeRD-Erkennung""" def __init__(self, ocr_language: str = "deu", ocr_dpi: int = 300): self.ocr_language = ocr_language self.ocr_dpi = ocr_dpi def verarbeite(self, pdf_pfad: str) -> Dict: """ Vollständige PDF-Verarbeitung Returns: Dict mit: text, ist_zugferd, zugferd_xml, hat_text, ocr_durchgefuehrt """ pfad = Path(pdf_pfad) if not pfad.exists(): return {"fehler": f"Datei nicht gefunden: {pdf_pfad}"} ergebnis = { "pfad": str(pfad), "text": "", "ist_zugferd": False, "zugferd_xml": None, "hat_text": False, "ocr_durchgefuehrt": False, "seiten": 0 } # 1. ZUGFeRD prüfen zugferd_result = self.pruefe_zugferd(pdf_pfad) ergebnis["ist_zugferd"] = zugferd_result["ist_zugferd"] ergebnis["zugferd_xml"] = zugferd_result.get("xml") # 2. Text extrahieren text, seiten = self.extrahiere_text(pdf_pfad) ergebnis["text"] = text ergebnis["seiten"] = seiten ergebnis["hat_text"] = bool(text and len(text.strip()) > 50) # 3. OCR falls kein Text (aber NICHT bei ZUGFeRD!) if not ergebnis["hat_text"] and not ergebnis["ist_zugferd"]: logger.info(f"Kein Text gefunden, starte OCR für {pfad.name}") ocr_text, ocr_erfolg = self.fuehre_ocr_aus(pdf_pfad) if ocr_erfolg: ergebnis["text"] = ocr_text ergebnis["hat_text"] = bool(ocr_text and len(ocr_text.strip()) > 50) ergebnis["ocr_durchgefuehrt"] = True return ergebnis def extrahiere_text(self, pdf_pfad: str) -> Tuple[str, int]: """ Extrahiert Text aus PDF Returns: Tuple von (text, seitenanzahl) """ text_parts = [] seiten = 0 # Methode 1: pdfplumber (besser für Tabellen) if PDFPLUMBER_AVAILABLE: try: with pdfplumber.open(pdf_pfad) as pdf: seiten = len(pdf.pages) for page in pdf.pages: page_text = page.extract_text() if page_text: text_parts.append(page_text) if text_parts: return "\n\n".join(text_parts), seiten except Exception as e: logger.debug(f"pdfplumber Fehler: {e}") # Methode 2: pypdf (Fallback) if PYPDF_AVAILABLE: try: reader = PdfReader(pdf_pfad) seiten = len(reader.pages) for page in reader.pages: page_text = page.extract_text() if page_text: text_parts.append(page_text) if text_parts: return "\n\n".join(text_parts), seiten except Exception as e: logger.debug(f"pypdf Fehler: {e}") # Methode 3: pdftotext CLI (Fallback) try: result = subprocess.run( ["pdftotext", "-layout", pdf_pfad, "-"], capture_output=True, text=True, timeout=30 ) if result.returncode == 0 and result.stdout.strip(): return result.stdout, seiten except Exception as e: logger.debug(f"pdftotext Fehler: {e}") return "", seiten def pruefe_zugferd(self, pdf_pfad: str) -> Dict: """ Prüft ob PDF eine ZUGFeRD/Factur-X Rechnung ist Returns: Dict mit: ist_zugferd, xml (falls vorhanden) """ ergebnis = {"ist_zugferd": False, "xml": None} # Methode 1: factur-x Library try: from facturx import get_facturx_xml_from_pdf xml_bytes = get_facturx_xml_from_pdf(pdf_pfad) if xml_bytes: ergebnis["ist_zugferd"] = True ergebnis["xml"] = xml_bytes.decode("utf-8") if isinstance(xml_bytes, bytes) else xml_bytes logger.info(f"ZUGFeRD erkannt: {Path(pdf_pfad).name}") return ergebnis except ImportError: logger.debug("factur-x nicht installiert") except Exception as e: logger.debug(f"factur-x Fehler: {e}") # Methode 2: Manuell nach XML-Attachment suchen if PYPDF_AVAILABLE: try: reader = PdfReader(pdf_pfad) if "/Names" in reader.trailer.get("/Root", {}): # Embedded Files prüfen pass # Komplexere Logik hier # Alternativ: Im Text nach ZUGFeRD-Markern suchen for page in reader.pages[:1]: # Nur erste Seite text = page.extract_text() or "" if any(marker in text.upper() for marker in ["ZUGFERD", "FACTUR-X", "EN 16931"]): ergebnis["ist_zugferd"] = True logger.info(f"ZUGFeRD-Marker gefunden: {Path(pdf_pfad).name}") break except Exception as e: logger.debug(f"ZUGFeRD-Prüfung Fehler: {e}") return ergebnis def fuehre_ocr_aus(self, pdf_pfad: str) -> Tuple[str, bool]: """ Führt OCR mit ocrmypdf durch Returns: Tuple von (text, erfolg) """ pfad = Path(pdf_pfad) temp_pfad = pfad.with_suffix(".ocr.pdf") try: # ocrmypdf ausführen result = subprocess.run( [ "ocrmypdf", "--language", self.ocr_language, "--deskew", # Schräge Scans korrigieren "--clean", # Bild verbessern "--skip-text", # Seiten mit Text überspringen "--force-ocr", # OCR erzwingen falls nötig str(pfad), str(temp_pfad) ], capture_output=True, text=True, timeout=120 # 2 Minuten Timeout ) if result.returncode == 0 and temp_pfad.exists(): # Original mit OCR-Version ersetzen pfad.unlink() temp_pfad.rename(pfad) # Text aus OCR-PDF extrahieren text, _ = self.extrahiere_text(str(pfad)) return text, True else: logger.error(f"OCR Fehler: {result.stderr}") if temp_pfad.exists(): temp_pfad.unlink() return "", False except subprocess.TimeoutExpired: logger.error(f"OCR Timeout für {pfad.name}") if temp_pfad.exists(): temp_pfad.unlink() return "", False except FileNotFoundError: logger.error("ocrmypdf nicht installiert") return "", False except Exception as e: logger.error(f"OCR Fehler: {e}") if temp_pfad.exists(): temp_pfad.unlink() return "", False def extrahiere_metadaten(self, pdf_pfad: str) -> Dict: """Extrahiert PDF-Metadaten""" metadaten = {} if PYPDF_AVAILABLE: try: reader = PdfReader(pdf_pfad) if reader.metadata: metadaten = { "titel": reader.metadata.get("/Title", ""), "autor": reader.metadata.get("/Author", ""), "ersteller": reader.metadata.get("/Creator", ""), "erstellt": reader.metadata.get("/CreationDate", ""), } except Exception as e: logger.debug(f"Metadaten-Fehler: {e}") return metadaten