from __future__ import annotations import re import math import subprocess import tempfile from dataclasses import dataclass from pathlib import Path from ollama_client import ContextLimitError, transcribe_image class PdfTextError(RuntimeError): pass @dataclass(frozen=True) class ExtractionResult: text: str page_count: int ocr_pages: tuple[int, ...] def run(args: list[str], timeout: int = 60) -> str: try: completed = subprocess.run(args, check=True, capture_output=True, text=True, timeout=timeout) except FileNotFoundError as exc: raise PdfTextError(f"Benötigtes Programm fehlt: {args[0]}") from exc except subprocess.CalledProcessError as exc: detail = (exc.stderr or exc.stdout or "").strip() raise PdfTextError(detail or f"Programm fehlgeschlagen: {args[0]}") from exc except subprocess.TimeoutExpired as exc: raise PdfTextError(f"Zeitüberschreitung bei {Path(args[0]).name}.") from exc return completed.stdout def page_count(pdf: Path, max_pages: int = 10) -> int: output = run(["pdfinfo", str(pdf)], timeout=20) match = re.search(r"^Pages:\s+(\d+)\s*$", output, re.MULTILINE) if not match: raise PdfTextError("Seitenzahl der PDF konnte nicht ermittelt werden.") count = int(match.group(1)) if count < 1 or count > max_pages: raise PdfTextError(f"PDF muss zwischen 1 und {max_pages} Seiten enthalten.") return count def _use_ocr(text: str) -> bool: visible = re.sub(r"\s+", "", text) alnum = re.sub(r"\W", "", text) return len(visible) < 60 or len(alnum) < 40 def _ocr_page(pdf: Path, page: int) -> str: info = run(["pdfinfo", "-box", "-f", str(page), "-l", str(page), str(pdf)]) size = re.search(r"Page\s+\d+\s+MediaBox:\s+([-\d.]+)\s+([-\d.]+)\s+([-\d.]+)\s+([-\d.]+)", info) if not size: raise PdfTextError(f"Abmessungen von Seite {page} konnten nicht gelesen werden.") left, top, right, bottom = map(float, size.groups()) width, height = right - left, bottom - top dpi = min(200, 72 * (4_000_000 / (width * height)) ** .5) rotation = re.search(r'Page\s+\d+\s+rot:\s+(-?\d+)', info) if rotation and int(rotation.group(1)) % 180: height = width with tempfile.TemporaryDirectory(prefix="contentpanel_ocr_") as temp: prefix = Path(temp) / "page" def read(top: int, pixels: int, depth: int = 0) -> str: run([ "pdftoppm", "-f", str(page), "-l", str(page), "-singlefile", "-r", str(dpi), "-jpeg", *(["-y", str(top), "-H", str(pixels)] if depth else []), str(pdf), str(prefix) ], timeout=90) image = prefix.with_suffix(".jpg") if not image.is_file(): raise PdfTextError(f"OCR-Vorbereitung für Seite {page} hat kein Bild erzeugt.") try: return transcribe_image(image.read_bytes()) except ContextLimitError: if depth >= 3 or pixels <= 256: raise # Overlap keeps text on a cut line readable; the editorial pass can # remove duplicates. Never drop a strip or publish a partial reading. half = pixels // 2 return (read(top, half + 48, depth + 1) + '\n\n' + read(top + half - 48, pixels - half + 48, depth + 1)) return read(0, math.ceil(height * dpi / 72)) def normalize_text(text: str) -> str: text = text.replace("\r\n", "\n").replace("\r", "\n").replace("\x0c", "\n") text = re.sub(r"(?<=\w)-[ \t]*\n[ \t]*(?=\w)", "", text) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n[ \t]+", "\n", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def extract_text(pdf: Path, max_pages: int = 10) -> ExtractionResult: count = page_count(pdf, max_pages=max_pages) pages: list[str] = [] ocr_pages: list[int] = [] for number in range(1, count + 1): extracted = normalize_text(run(["pdftotext", "-f", str(number), "-l", str(number), "-layout", "-enc", "UTF-8", str(pdf), "-"], timeout=30)) if _use_ocr(extracted): ocr = normalize_text(_ocr_page(pdf, number)) if len(ocr) > len(extracted): extracted = ocr ocr_pages.append(number) if extracted: pages.append(extracted) text = normalize_text("\n\n".join(pages)) if len(re.sub(r"\s+", "", text)) < 20: raise PdfTextError("Aus der PDF konnte auch per OCR kein ausreichender Text gewonnen werden.") return ExtractionResult(text=text, page_count=count, ocr_pages=tuple(ocr_pages))