"""Positionsbasierte PDF-Extraktion fuer die Fortna-Hauptanleitungen. Erzeugt aus einem Word-generierten PDF eine .rst-Datei, in der Text und Bilder in der richtigen Lesereihenfolge (nach Y-Position pro Seite) verwoben sind. Verbesserungen gegenueber extract_products.extract_text_to_rst: 1. Ueberschriften: Die im PDF vorhandene Nummerierung (z.B. "1.7 Wichtige Informationen") wird NUR zur Bestimmung der Gliederungstiefe ausgewertet, im RST-Ausgabetext aber ENTFERNT. Sphinx/LaTeX nummeriert selbst - dadurch keine doppelte Nummerierung (alt 1.7 + neu 3.2.1) mehr. 2. Bilder: werden positionsbasiert (bbox) pro Seite eingelesen, dedupliziert und an der korrekten Textstelle als `.. figure::`-Direktive referenziert. Bildunterschriften ("Abbildung N: ...") werden als figure-caption uebernommen. 3. Kopf-/Fusszeilen und Inhaltsverzeichnis-Seiten werden herausgefiltert. Oeffentliche Funktion: extract_manual(pdf_path, rst_path, img_dir) -> (blocks, images) """ from __future__ import annotations import hashlib import io import logging import re from pathlib import Path log = logging.getLogger(__name__) # Schwellwerte fuer die Symbol-/Icon-Erkennung SYMBOL_MAX_EDGE = 260 # Bilder <= dieser Kantenlaenge gelten als Symbol-Kandidat SYMBOL_MIN_COUNT = 2 # ... wenn dasselbe Bild mind. so oft vorkommt AHASH_THRESHOLD = 6 # max. Hamming-Distanz, ab der zwei aHashes als gleich gelten def average_hash(image_bytes: bytes, size: int = 8) -> int: """Perceptual Average-Hash (aHash) eines Bildes als Integer. Graustufen -> size x size -> Bit je Pixel (>= Mittelwert). Zwei visuell aehnliche Bilder (auch bei leicht anderer Skalierung/Kompression, wie mehrfach eingebettete Warnsymbole aus Word) haben eine kleine Hamming- Distanz. Robuster als ein exakter Byte-/Pixel-Hash. """ from PIL import Image try: im = Image.open(io.BytesIO(image_bytes)).convert("L").resize((size, size)) px = list(im.getdata()) avg = sum(px) / len(px) bits = 0 for i, p in enumerate(px): if p >= avg: bits |= (1 << i) return bits except Exception: # Fallback: Byte-Hash als Integer, wenn PIL das Bild nicht oeffnen kann return int(hashlib.md5(image_bytes).hexdigest()[:16], 16) def hamming(a: int, b: int) -> int: """Hamming-Distanz zweier Integer-Hashes (Anzahl unterschiedlicher Bits).""" return bin(a ^ b).count("1") # --- Bild-Klassifizierung ------------------------------------------------- SKETCH_MIN_BYTES = 10_000 # SW-Grafik groesser als ~10 KB gilt als Sketch def is_junk_fragment(image_bytes: bytes, w: int, h: int) -> bool: """True fuer wertlose Text-/Nummern-Fragmente (kein echtes Symbol/Bild). Typisch: flache (Hoehe <= 55 px), breite, schwarz-weisse Schnipsel, die in Wahrheit gerenderte Kapitelnummern/Beschriftungen sind (z.B. "5.2.6", zwei Quadrate + Zahlen). Diese werden verworfen. """ if h <= 55 and w / max(h, 1) >= 1.6 and is_grayscale(image_bytes): return True return False def is_grayscale(image_bytes: bytes) -> bool: """True, wenn das Bild (nahezu) schwarz-weiss/graustufig ist.""" from PIL import Image try: rgb = Image.open(io.BytesIO(image_bytes)).convert("RGB") data = list(rgb.getdata()) step = max(1, len(data) // 200) for r, g, b in data[::step]: if abs(r - g) > 12 or abs(g - b) > 12: return False return True except Exception: return False def classify_image(image_bytes: bytes, ext: str, w: int, h: int) -> str: """Klassifiziert ein Bild: 'photo' | 'sketch' | 'graphic'. - photo: JPEG/JPG (Fotos sind meist projektspezifisch) - sketch: schwarz-weisse Grafik groesser als ~10 KB (Skizze/Zeichnung) - graphic: alles uebrige (farbige PNG-Grafik, Diagramm) """ if ext.lower() in ("jpg", "jpeg"): return "photo" if len(image_bytes) > SKETCH_MIN_BYTES and is_grayscale(image_bytes): return "sketch" return "graphic" # Logo-Erkennung: grosse, breite Grafik (Schriftzug-Logo) auf oberen/unteren # Seitenbereichen. Wir erkennen sie ueber Seitenposition + Seitenverhaeltnis, # da OCR des Schriftzugs zu aufwendig waere. def looks_like_logo(w: int, h: int, y_top: float, page_height: float) -> bool: """Heuristik: breites Bild (ratio > 2.2) im Kopf-/Fussbereich der Seite.""" if h == 0: return False ratio = w / h in_margin = y_top < page_height * 0.18 or y_top > page_height * 0.82 return ratio > 2.2 and in_margin and w >= 200 RST_H = {1: "=", 2: "-", 3: "~", 4: "^", 5: '"'} # Symbolfont-Zeichen (Private Use Area) auf Unicode abbilden bzw. entfernen. PUA_MAP = { "": "•", "": "•", "": "•", "": "→", "": "→", "": "←", "": "✓", "": " ", } _PUA_RE = re.compile("|".join(re.escape(k) for k in PUA_MAP)) _PUA_REST_RE = re.compile(r"[-]") # Rausch-Muster (Kopf/Fuss/Metadaten), die nicht in den Text gehoeren. _NOISE = { "services cz s.r.o", "schönenberger systeme gmbh", "justus-von-liebig-str", "info@schoenenberger.de", "www.schoenenberger.de", "www.schoenenberger-systems.com", "karolinska 661/4", } RE_DATE = re.compile(r"^\d{2}\.\d{2}\.\d{4}$") RE_PAGENO = re.compile(r"^(seite|page)\s+\d+$", re.I) RE_PHONE = re.compile(r"^[TF]:\s*\+?\d") # Ueberschrift mit fuehrender Nummer: "1.7 Wichtige Informationen" # Nummernteile max 3-stellig (verhindert Datum/Artikelnummer als Ueberschrift). RE_HEADING = re.compile(r"^(\d{1,3}(?:\.\d{1,3}){0,4})\s+([A-Za-zÄÖÜäöüß].{2,})$") # reine Nummernzeile (Nummer und Titel stehen im PDF auf getrennten Zeilen) RE_NUM_ONLY = re.compile(r"^(\d{1,3}(?:\.\d{1,3}){0,4})\.?$") # Bildunterschrift "Abbildung 17: ..." / "Abb. 17 ..." / "Figure 3: ..." RE_CAPTION = re.compile(r"^(Abbildung|Abb\.?|Bild|Figure|Tabelle)\s*\d+\s*[:.]?\s*(.*)$", re.I) def clean_pua(text: str) -> str: """Symbolfont-Zeichen ersetzen, restliche PUA-Zeichen entfernen.""" text = _PUA_RE.sub(lambda m: PUA_MAP[m.group(0)], text) return _PUA_REST_RE.sub("", text) def _is_noise(text: str) -> bool: low = text.lower().strip() if not low: return True if RE_DATE.match(low) or RE_PAGENO.match(low) or RE_PHONE.match(text): return True return any(kw in low for kw in _NOISE) def _strip_heading_number(num: str, rest: str) -> str: """Gibt den Ueberschriftentext ohne fuehrende Gliederungsnummer zurueck.""" return rest.strip() def caption_slug(caption_text: str) -> str: """Leitet ein kurzes Datei-Stichwort aus einer Bildunterschrift ab. "Abbildung 17: Weiche Typ A links" -> "Weiche_Typ_A" Nimmt die ersten Woerter nach der "Abbildung N:"-Kennung. """ m = RE_CAPTION.match(caption_text) body = (m.group(2) if m else caption_text).strip() # Umlaute/Sonderzeichen -> ASCII-nah, Wortgrenzen zu _ body = (body.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue") .replace("Ä", "Ae").replace("Ö", "Oe").replace("Ü", "Ue") .replace("ß", "ss")) words = re.findall(r"[A-Za-z0-9]+", body) if not words: return "" slug = "_".join(words[:3]) # max. 3 Woerter fuer kompakten Namen return slug[:40] def extract_manual(pdf_path: Path, rst_path: Path, img_dir: Path, title: str | None = None, symbols_dir: Path | None = None, symbol_registry: dict | None = None, logos_dir: Path | None = None, logo_registry: dict | None = None, image_dest: Path | None = None, photo_dest: Path | None = None) -> tuple[int, int]: """Extrahiert Text + Bilder eines Anleitungs-PDF in eine .rst-Datei. Bild-Routing (siehe doc/plan_fortna_extraktion.md): - Symbole (klein, mehrfach): einmal zentral -> symbols_dir (assets/shared/symbols), im RST als `.. image::` referenziert. - Logos (breite Schriftzug-Grafik im Seitenrand): pro visueller Variante alle Aufloesungen -> logos_dir (assets/shared/logos), NICHT inline referenziert (reine Deko). - Inhaltsbilder: nach image_dest (assets/... fuer generelle Kapitel bzw. projekte/fortna/images/... fuer projektspezifische). Fotos (.jpg) landen IMMER projektspezifisch; das steuert der Aufrufer ueber image_dest. Benennung: PNG mit direkt folgender Bildunterschrift -> _NN.png, Fotos -> img_NNN.jpg, SW-Skizzen (>10KB) -> sketch_NNN.png, Rest img_NNN.png. symbol_registry / logo_registry (aHash -> Datei) werden dokumentuebergreifend wiederverwendet. Returns: (anzahl_textbloecke, anzahl_bilder) """ import fitz if symbol_registry is None: symbol_registry = {} if logo_registry is None: logo_registry = {} # image_dest: Zielordner fuer Grafiken/Skizzen. photo_dest: fuer Fotos (.jpg, # immer projektspezifisch). Default beider: img_dir. if image_dest is None: image_dest = img_dir if photo_dest is None: photo_dest = image_dest doc = fitz.open(str(pdf_path)) image_dest.mkdir(parents=True, exist_ok=True) # --- Seiten identifizieren, die uebersprungen werden --- # a) TOC-Seiten (Punktfueller / "Inhaltsverzeichnis") # b) Aenderungshistorie-/Dokumentinfo-Tabellen ("Aenderungsinformation", # Spaltenkombination Version+Datum+Aenderungsbeschreibung). Diese Tabellen # zerfallen beim linearen Auslesen in sinnlose Einzelzellen. skip_pages: set[int] = set() for pn in range(len(doc)): t = doc[pn].get_text() low = t.lower() if t.count("....") >= 3 or "\ninhaltsverzeichnis\n" in low: skip_pages.add(pn) elif "änderungsinformation" in low or "dokumenteninformation" in low: skip_pages.add(pn) elif ("änderungsbeschreibung" in low and "version" in low and ("datum" in low or "kapitel" in low)): skip_pages.add(pn) # --- Pass 1: Text- und Bild-Elemente sammeln (Bilder nur puffern) --- # Bilder werden zunaechst nicht geschrieben, sondern per Perceptual-Hash # (aHash) einem Cluster zugeordnet. Visuell aehnliche Bilder (gleiches Symbol # in anderer Skalierung) landen im selben Cluster. Erst nach dem Scan steht # fest, welches Cluster mehrfach vorkommt (Symbol-Kriterium). elements: list[dict] = [] clusters: list[dict] = [] # [{ahash, data, ext, w, h, count, cls, margin_hits, variants}] def _find_cluster(ah: int) -> int: for ci, c in enumerate(clusters): if hamming(ah, c["ahash"]) <= AHASH_THRESHOLD: return ci return -1 for pn in range(len(doc)): if pn in skip_pages: continue page = doc[pn] ph = page.rect.height pw = page.rect.width # Textzeilen mit Position page_dict = page.get_text("dict", flags=fitz.TEXT_PRESERVE_WHITESPACE) for block in page_dict.get("blocks", []): if block.get("type") != 0: continue for line in block.get("lines", []): bbox = line.get("bbox", [0, 0, 0, 0]) if bbox[1] < 70 or bbox[3] > ph - 45: # Kopf-/Fusszeile continue text = "" max_size = 0.0 bold = False for span in line.get("spans", []): st = clean_pua(span.get("text", "")) if st.strip(): text += st max_size = max(max_size, span.get("size", 0)) if span.get("flags", 0) & 2**4: # bold flag bold = True text = text.strip() if not text or _is_noise(text): continue elements.append({ "kind": "text", "y": bbox[1], "page": pn + 1, "text": text, "size": round(max_size, 1), "bold": bold, }) # Bilder mit Position. Titelseiten (erste 2 Seiten) = nur Logo/Deko. if pn < 2: continue for im in page.get_image_info(xrefs=True): xref = im.get("xref", 0) if not xref: continue b = im["bbox"] w, h = im.get("width", 0), im.get("height", 0) # Vollseiten-Deko / Hintergrundbilder ueberspringen if (b[2] - b[0]) > pw * 0.95 and (b[3] - b[1]) > ph * 0.95: continue if w < 40 or h < 40: # winzige Icons continue base = doc.extract_image(xref) if not base: continue data = base["image"] if len(data) < 3000: continue # Wertlose Text-/Nummern-Fragmente (flach, SW) verwerfen if is_junk_fragment(data, w, h): continue ext = base.get("ext", "png") ah = average_hash(data) in_margin = looks_like_logo(w, h, b[1], ph) ci = _find_cluster(ah) if ci < 0: clusters.append({"ahash": ah, "data": data, "ext": ext, "w": w, "h": h, "count": 1, "cls": classify_image(data, ext, w, h), "margin_hits": 1 if in_margin else 0, # groesste Auflvariante fuer Logo-Ablage merken "variants": [(w * h, w, h, data, ext)]}) ci = len(clusters) - 1 else: clusters[ci]["count"] += 1 if in_margin: clusters[ci]["margin_hits"] += 1 clusters[ci]["variants"].append((w * h, w, h, data, ext)) elements.append({ "kind": "image", "y": b[1], "page": pn + 1, "cluster": ci, }) doc.close() import os as _os # --- Zielverzeichnisse + relative Pfad-Praefixe (vom RST aus) --- rst_parent = rst_path.parent if symbols_dir is None: symbols_dir = rst_parent.parent.parent / "assets" / "shared" / "symbols" if logos_dir is None: logos_dir = symbols_dir.parent / "logos" def _rel(target: Path) -> str: try: return Path(_os.path.relpath(target, rst_parent)).as_posix() except ValueError: return target.as_posix() sym_rel_prefix = _rel(symbols_dir) dest_rel_prefix = _rel(image_dest) photo_rel_prefix = _rel(photo_dest) # --- Pass 2a: Cluster klassifizieren (Logo / Symbol / Inhaltsbild) --- # Logo: breit (ratio > 2.2), im Seitenrand UND wiederkehrend (>=3x) - das # unterscheidet ein Kopf-/Fusszeilen-Logo von einem einmaligen breiten # Diagramm. Alle Aufloesungen -> assets/shared/logos, NICHT inline. # Symbol: klein (<=260px) UND mehrfach (>=2x) -> assets/shared/symbols. cluster_is_symbol: list[bool] = [False] * len(clusters) cluster_is_logo: list[bool] = [False] * len(clusters) cluster_sympath: list[str] = [""] * len(clusters) for ci, c in enumerate(clusters): ratio = c["w"] / c["h"] if c["h"] else 0 is_logo = (ratio > 2.2 and c["margin_hits"] >= 3 and c["count"] >= 3 and c["w"] >= 200) if is_logo: cluster_is_logo[ci] = True # bereits bekannte Logo-Variante? -> nicht neu schreiben if any(hamming(c["ahash"], rah) <= AHASH_THRESHOLD + 2 for rah in logo_registry): continue logos_dir.mkdir(parents=True, exist_ok=True) idx = len(logo_registry) + 1 for (_area, w, h, data, ext) in sorted(set(c["variants"]), reverse=True): (logos_dir / f"logo_{idx:02d}_{w}px.{ext}").write_bytes(data) logo_registry[c["ahash"]] = idx continue is_symbol = (max(c["w"], c["h"]) <= SYMBOL_MAX_EDGE and c["count"] >= SYMBOL_MIN_COUNT) cluster_is_symbol[ci] = is_symbol if is_symbol: match = None for reg_ah, reg_file in symbol_registry.items(): if hamming(c["ahash"], reg_ah) <= AHASH_THRESHOLD: match = reg_file break if match is None: symbols_dir.mkdir(parents=True, exist_ok=True) match = f"symbol_{c['ahash'] & 0xffffffff:08x}.{c['ext']}" (symbols_dir / match).write_bytes(c["data"]) symbol_registry[c["ahash"]] = match cluster_sympath[ci] = f"{sym_rel_prefix}/{match}" # --- pro Seite nach Y-Position sortieren (Lesereihenfolge) --- elements.sort(key=lambda e: (e["page"], e["y"])) # --- Zeilen zusammenfuehren, die im PDF getrennt stehen --- # a) Nummer-only-Zeile ("2.3.1") + folgende Titelzeile # b) Bullet-only-Zeile ("•") + folgende Textzeile -> "• Text" merged: list[dict] = [] i = 0 while i < len(elements): e = elements[i] if e["kind"] == "text": nxt = elements[i + 1] if i + 1 < len(elements) else None # a) Gliederungsnummer + Titel m = RE_NUM_ONLY.match(e["text"]) if (m and nxt and nxt["kind"] == "text" and re.match(r"[A-Za-zÄÖÜäöüß]", nxt["text"])): merged.append({**e, "text": f"{m.group(1)} {nxt['text']}", "size": max(e["size"], nxt["size"])}) i += 2 continue # b) einzelnes Bullet-Zeichen + Text der Folgezeile if e["text"].strip() in ("•", "-", "–", "*") and nxt and nxt["kind"] == "text": merged.append({**e, "text": f"• {nxt['text']}"}) i += 2 continue merged.append(e) i += 1 # --- Pass 2b: Inhaltsbilder schreiben (mit Caption-Benennung + Klasse) --- # Jetzt (nach dem Merge) ist die Folgezeile jedes Bildes bekannt -> Caption. # Pro Cluster genau einmal schreiben; Zielname aus Caption oder Klasse. # Fotos (.jpg) -> photo_dest (projektspezifisch), Rest -> image_dest. image_dest.mkdir(parents=True, exist_ok=True) photo_dest.mkdir(parents=True, exist_ok=True) counters = {"img": 0, "sketch": 0} used_names: set[str] = set() slug_counters: dict[str, int] = {} cluster_path: dict[int, str] = {} # ci -> RST-Pfad (relativ) def _next_name(cls: str, ext: str, slug: str) -> str: if slug: slug_counters[slug] = slug_counters.get(slug, 0) + 1 name = f"{slug}_{slug_counters[slug]:02d}.{ext}" elif cls == "sketch": counters["sketch"] += 1 name = f"sketch_{counters['sketch']:03d}.{ext}" else: # photo / graphic counters["img"] += 1 name = f"img_{counters['img']:03d}.{ext}" while name in used_names: counters["img"] += 1 name = f"img_{counters['img']:03d}.{ext}" used_names.add(name) return name n_m = len(merged) for idx, e in enumerate(merged): if e["kind"] != "image": continue ci = e["cluster"] if cluster_is_logo[ci]: e["path"] = "" # Logos werden nicht inline referenziert e["is_symbol"] = False continue if cluster_is_symbol[ci]: e["path"] = cluster_sympath[ci] e["is_symbol"] = True continue e["is_symbol"] = False if ci in cluster_path: # Duplikat -> bereits geschrieben e["path"] = cluster_path[ci] continue c = clusters[ci] # Caption der Folgezeile? -> Stichwort fuer Benennung slug = "" if idx + 1 < n_m and merged[idx + 1]["kind"] == "text": cm = RE_CAPTION.match(merged[idx + 1]["text"]) if cm: slug = caption_slug(merged[idx + 1]["text"]) name = _next_name(c["cls"], c["ext"], slug) # Fotos immer projektspezifisch (photo_dest), sonst image_dest if c["cls"] == "photo": (photo_dest / name).write_bytes(c["data"]) rel = f"{photo_rel_prefix}/{name}" else: (image_dest / name).write_bytes(c["data"]) rel = f"{dest_rel_prefix}/{name}" cluster_path[ci] = rel e["path"] = rel img_counter = len(cluster_path) # --- RST rendern --- if title is None: title = pdf_path.stem out: list[str] = [RST_H[1] * len(title), title, RST_H[1] * len(title), ""] # Absatz-Puffer: aufeinanderfolgende Fliesstextzeilen zu einem Absatz # zusammenfuehren (PDF-Zeilenumbrueche sind kein Absatzende). para: list[str] = [] def flush_para() -> None: if para: out.append(" ".join(para).strip()) out.append("") para.clear() RE_BULLET = re.compile(r"^\s*[•\-–*]\s+") blocks = 0 n = len(merged) for idx, e in enumerate(merged): if e["kind"] == "image": flush_para() path = e.get("path", "") if not path: continue if e.get("is_symbol"): # Symbol/Warnzeichen: kleine, zentrierte Grafik ohne Caption-Box, # verweist auf die eine zentrale Quelle in assets/shared/symbols. out.append("") out.append(f".. image:: {path}") out.append(" :align: center") out.append(" :height: 12mm") out.append("") blocks += 1 continue caption = "" if idx + 1 < n and merged[idx + 1]["kind"] == "text": cm = RE_CAPTION.match(merged[idx + 1]["text"]) if cm: caption = merged[idx + 1]["text"] merged[idx + 1]["_consumed"] = True out.append("") out.append(f".. figure:: {path}") out.append(" :align: center") out.append(" :width: 80%") if caption: out.append("") out.append(f" {caption}") out.append("") blocks += 1 continue if e.get("_consumed"): continue text = e["text"] hm = RE_HEADING.match(text) # Ueberschrift NUR mit fuehrender PDF-Gliederungsnummer akzeptieren und # offensichtliche Fragmente/Bildlegenden ausschliessen (sonst werden # Zellen wie "POS", "li.", "re.", "bar, max. 10 bar" zu Ueberschriften). is_heading = False if hm and len(hm.group(2)) < 120: num, rest = hm.group(1), hm.group(2) rest_stripped = rest.strip() is_frag = ( len(rest_stripped) < 4 # zu kurz ("POS", "li.") or rest_stripped.endswith((",",)) # abgeschnittene Zelle or re.match(r"^(mm|cm|m|kg|x|l|h|min|s|bar|POS|li|re|Pos)\b", rest_stripped, re.I) or re.match(r"^(li|re|oben|unten|vorne|hinten|links|rechts)\.?$", rest_stripped, re.I) ) if not is_frag: is_heading = True if is_heading: flush_para() num, rest = hm.group(1), hm.group(2) heading = _strip_heading_number(num, rest) # NUMMER ENTFERNEN level = min(num.count(".") + 1, 5) char = RST_H[level] out.append("") if level == 1: out.append(char * max(len(heading), 3)) out.append(heading) out.append(char * max(len(heading), 3)) out.append("") blocks += 1 continue # Aufzaehlung: Zeile beginnt mit Bullet ODER enthaelt mehrere Bullets # (mehrere Listenpunkte sind im PDF oft in eine Zeile zusammengelaufen). if RE_BULLET.match(text) or text.count("•") >= 2: flush_para() # An Bullet-Zeichen aufsplitten, jeder Teil wird ein Listenpunkt. parts = re.split(r"\s*•\s*", text) for part in parts: item = RE_BULLET.sub("", part).strip().rstrip(",") if item: out.append(f"- {item}") out.append("") blocks += 1 continue # Fliesstext: an Absatz anhaengen. Endet die Zeile mit Satzzeichen, gilt # der Absatz als abgeschlossen und wird geflusht. para.append(text) if re.search(r"[.:!?]$", text): flush_para() blocks += 1 flush_para() rst = "\n".join(out) rst = re.sub(r"\n{3,}", "\n\n", rst).strip() + "\n" rst_path.parent.mkdir(parents=True, exist_ok=True) rst_path.write_text(rst, encoding="utf-8") return blocks, img_counter