459 lines
14 KiB
Python
459 lines
14 KiB
Python
"""
|
|
extract_products.py - Extrahiert Bilder und Text aus Produktprogramm-PDFs.
|
|
|
|
Verwendet PyMuPDF (fitz) fuer zuverlaessige Bild- und Textextraktion.
|
|
|
|
Verwendung:
|
|
python extract_products.py [--data-dir <pfad>]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Mapping: PDF-Dateiname (Teilmatch) -> Zielordner fuer Bilder
|
|
PDF_TARGETS: list[tuple[str, str]] = [
|
|
("01_OMNIFLO", "omniflo"),
|
|
("02_ILS", "ils"),
|
|
("03_CPC", "cpc"),
|
|
("05_GERÜST", "geruest"),
|
|
("08_Trolley", "trolley"),
|
|
]
|
|
|
|
# rST Ueberschriften-Zeichen nach Level
|
|
RST_HEADING_CHARS = {1: "=", 2: "-", 3: "~", 4: "^"}
|
|
|
|
|
|
def find_target(pdf_name: str) -> str | None:
|
|
"""Findet den Zielordner-Namen fuer eine PDF-Datei."""
|
|
for pattern, target in PDF_TARGETS:
|
|
if pattern.lower() in pdf_name.lower():
|
|
return target
|
|
return None
|
|
|
|
|
|
def extract_images(pdf_path: Path, output_dir: Path) -> int:
|
|
"""Extrahiert alle Bilder aus einem PDF nach output_dir.
|
|
|
|
Returns:
|
|
Anzahl extrahierter Bilder.
|
|
"""
|
|
import fitz
|
|
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
doc = fitz.open(str(pdf_path))
|
|
count = 0
|
|
|
|
for page_num in range(len(doc)):
|
|
page = doc[page_num]
|
|
image_list = page.get_images(full=True)
|
|
|
|
for img_idx, img_info in enumerate(image_list):
|
|
xref = img_info[0]
|
|
base_image = doc.extract_image(xref)
|
|
if not base_image:
|
|
continue
|
|
|
|
image_bytes = base_image["image"]
|
|
ext = base_image.get("ext", "png")
|
|
|
|
# Kleine Bilder (Icons, Logos < 5KB) ueberspringen
|
|
if len(image_bytes) < 5000:
|
|
continue
|
|
|
|
count += 1
|
|
img_name = f"page{page_num + 1:03d}_img{img_idx + 1:02d}.{ext}"
|
|
img_path = output_dir / img_name
|
|
img_path.write_bytes(image_bytes)
|
|
log.debug(" Bild: %s (%d bytes)", img_name, len(image_bytes))
|
|
|
|
doc.close()
|
|
return count
|
|
|
|
|
|
def _extract_toc_numbers(doc) -> dict[str, str]:
|
|
"""Extrahiert Kapitelnummern aus dem Inhaltsverzeichnis des PDFs.
|
|
|
|
Sucht TOC-Seiten (Seiten mit '....' Punkt-Fuellern) und ordnet
|
|
Ueberschriftentexte ihren Kapitelnummern zu. Nur Eintraege mit
|
|
expliziter Nummer werden erfasst (bis Level 2).
|
|
|
|
Returns:
|
|
Dict title_text -> kapitelnummer, z.B. {"Weichen Omniflo": "2.3"}
|
|
"""
|
|
toc_entries: dict[str, str] = {}
|
|
|
|
for page_num in range(len(doc)):
|
|
page = doc[page_num]
|
|
text = page.get_text()
|
|
if "...." not in text:
|
|
continue
|
|
|
|
lines = text.splitlines()
|
|
current_number = ""
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
i += 1
|
|
if not line:
|
|
continue
|
|
|
|
# Kapitelnummer auf separater Zeile
|
|
num_match = re.match(r"^(\d+(?:\.\d+)*)\s*$", line)
|
|
if num_match:
|
|
current_number = num_match.group(1)
|
|
continue
|
|
|
|
# Zeile mit Punkt-Fuellern = TOC-Eintrag
|
|
dot_match = re.search(r"\.{4,}", line)
|
|
if not dot_match:
|
|
continue
|
|
|
|
title = line[:dot_match.start()].strip()
|
|
if not title:
|
|
continue
|
|
|
|
# Inline-Nummer am Anfang? "1.4 Abkürzungen ....."
|
|
nm = re.match(r"^(\d+(?:\.\d+)*)\s+(.+)", title)
|
|
if nm:
|
|
toc_entries[nm.group(2).strip()] = nm.group(1)
|
|
current_number = ""
|
|
elif current_number:
|
|
toc_entries[title] = current_number
|
|
current_number = ""
|
|
|
|
return toc_entries
|
|
|
|
|
|
RE_DATE = re.compile(r"^\d{2}\.\d{2}\.\d{4}$")
|
|
RE_COPYRIGHT = re.compile(r"^Copyright\s+\d{4}", re.IGNORECASE)
|
|
RE_VERSION_LINE = re.compile(r"^Version\s*[:.]?\s*\d", re.IGNORECASE)
|
|
|
|
# Bekannte Rausch-Muster: Autorinnen, Metadaten, Copyright-Zeilen
|
|
_NOISE_KEYWORDS = {
|
|
"annette schopper", "erstellt:", "freigabe:", "nicht freigegeben",
|
|
"nur zur information", "schönenberger systeme gmbh",
|
|
"justus-von-liebig-str", "info@schoenenberger.de",
|
|
"www.schoenenberger.de",
|
|
}
|
|
|
|
|
|
def _is_noise(text: str) -> bool:
|
|
"""Erkennt Rausch-Zeilen die nicht ins RST gehoeren."""
|
|
lower = text.lower().strip()
|
|
if RE_COPYRIGHT.match(text):
|
|
return True
|
|
if RE_DATE.match(text.strip()):
|
|
return True
|
|
if RE_VERSION_LINE.match(text):
|
|
return True
|
|
for kw in _NOISE_KEYWORDS:
|
|
if kw in lower:
|
|
return True
|
|
# Seitenzahlen
|
|
if re.match(r"^Seite\s+\d+$", text, re.IGNORECASE):
|
|
return True
|
|
# Einzelne kurze Jahreszahlen mit Autorin: "2020 Annette Schopper"
|
|
if re.match(r"^\d{4}\s+[A-Z][a-z]+\s+[A-Z][a-z]+$", text):
|
|
return True
|
|
return False
|
|
|
|
|
|
def extract_text_to_rst(pdf_path: Path, rst_path: Path) -> int:
|
|
"""Extrahiert Text aus PDF und schreibt ihn als .rst.
|
|
|
|
Returns:
|
|
Anzahl extrahierter Bloecke.
|
|
"""
|
|
import fitz
|
|
|
|
doc = fitz.open(str(pdf_path))
|
|
|
|
# TOC-Nummern vorab extrahieren
|
|
toc_numbers = _extract_toc_numbers(doc)
|
|
|
|
raw_lines: list[dict] = []
|
|
|
|
for page_num in range(len(doc)):
|
|
page = doc[page_num]
|
|
page_height = page.rect.height
|
|
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])
|
|
|
|
# Positionsbasierte Kopf-/Fusszeilen-Erkennung
|
|
# Kopfzeile: Y < 100pt (Logos, Kapitelname, OMNIFLO etc.)
|
|
if bbox[1] < 100:
|
|
continue
|
|
# Fusszeile: Y > page_height - 50pt
|
|
if bbox[3] > page_height - 50:
|
|
continue
|
|
|
|
text = ""
|
|
max_size = 0.0
|
|
for span in line.get("spans", []):
|
|
span_text = span.get("text", "").strip()
|
|
if span_text:
|
|
text += span_text + " "
|
|
size = span.get("size", 0)
|
|
if size > max_size:
|
|
max_size = size
|
|
|
|
text = text.strip()
|
|
if not text:
|
|
continue
|
|
if _is_noise(text):
|
|
continue
|
|
|
|
raw_lines.append({
|
|
"text": text,
|
|
"size": max_size,
|
|
"page": page_num + 1,
|
|
})
|
|
|
|
doc.close()
|
|
|
|
if not raw_lines:
|
|
return 0
|
|
|
|
# --- Phase 1: Nummern-Zeilen mit nachfolgender Text-Zeile zusammenfuehren ---
|
|
# Im PDF stehen "2.3.1" und "Antriebsstationen" auf separaten Zeilen.
|
|
# Zusammenfuehren nur wenn:
|
|
# - Die Nummer ein Abschnitts-Format hat (N oder N.N.N...)
|
|
# - Die naechste Zeile mit einem Buchstaben beginnt
|
|
# - Die Schriftgroesse >= 12pt (verhindert Artikelnummern)
|
|
RE_SECTION_NUM = re.compile(r"^(\d+(?:\.\d+)*)$")
|
|
merged: list[dict] = []
|
|
i = 0
|
|
while i < len(raw_lines):
|
|
m = RE_SECTION_NUM.match(raw_lines[i]["text"])
|
|
if m and i + 1 < len(raw_lines) and raw_lines[i]["size"] >= 12.0:
|
|
next_line = raw_lines[i + 1]
|
|
next_text = next_line["text"].strip()
|
|
# Naechste Zeile muss mit Buchstabe beginnen
|
|
if next_text and re.match(r"[A-Za-zÄÖÜäöüß]", next_text):
|
|
merged.append({
|
|
"text": f"{raw_lines[i]['text']} {next_line['text']}",
|
|
"size": max(raw_lines[i]["size"], next_line["size"]),
|
|
"page": raw_lines[i]["page"],
|
|
})
|
|
i += 2
|
|
continue
|
|
merged.append(raw_lines[i])
|
|
i += 1
|
|
|
|
# --- Phase 2: Inhaltsverzeichnis / Index entfernen ---
|
|
# Zuerst TOC-Seiten identifizieren (Seiten mit "...." Punkt-Fuellern)
|
|
RE_TOC_LINE = re.compile(r"\.{4,}")
|
|
toc_pages: set[int] = set()
|
|
for entry in merged:
|
|
if RE_TOC_LINE.search(entry["text"]):
|
|
toc_pages.add(entry["page"])
|
|
if entry["text"].strip().lower() in ("inhalt", "inhaltsverzeichnis"):
|
|
toc_pages.add(entry["page"])
|
|
|
|
# Alle Eintraege auf TOC-Seiten entfernen
|
|
filtered: list[dict] = []
|
|
for entry in merged:
|
|
if entry["page"] in toc_pages:
|
|
continue
|
|
filtered.append(entry)
|
|
|
|
# --- Phase 3: Ueberschriften erkennen und RST generieren ---
|
|
# Heading: Nummerierung wie "1 Text", "2.3 Text", "2.3.1.4 Text"
|
|
# Text muss mit einem Buchstaben beginnen (verhindert "610 014 001" etc.)
|
|
# Jeder Nummerierungsteil max 3 Stellen (verhindert Datumsangaben wie "10.12.2020")
|
|
# Titeltext muss mit einem Wort von mind. 3 Buchstaben beginnen
|
|
# (verhindert "2 B. def.", "19.8 mm (LW13)")
|
|
RE_HEADING = re.compile(
|
|
r"^(\d{1,3}(?:\.\d{1,3})*)\s+([A-Za-zÄÖÜäöüß]{3,}.*)$"
|
|
)
|
|
rst_lines: list[str] = []
|
|
title = pdf_path.stem
|
|
rst_lines.append(RST_HEADING_CHARS[1] * len(title))
|
|
rst_lines.append(title)
|
|
rst_lines.append(RST_HEADING_CHARS[1] * len(title))
|
|
rst_lines.append("")
|
|
|
|
prev_page = 0
|
|
block_count = 0
|
|
for entry in filtered:
|
|
text = entry["text"]
|
|
|
|
# Seitenumbruch-Kommentar
|
|
if entry["page"] != prev_page:
|
|
if prev_page > 0:
|
|
rst_lines.append("")
|
|
rst_lines.append(f".. Seite {entry['page']}")
|
|
rst_lines.append("")
|
|
prev_page = entry["page"]
|
|
|
|
# Nummerierte Ueberschrift?
|
|
# Fuer Level-1 (keine Punkte, z.B. "1 Text") zusaetzlich grosse
|
|
# Schrift (>= 12pt) verlangen um Fehlerkennungen zu vermeiden.
|
|
hm = RE_HEADING.match(text)
|
|
is_heading = False
|
|
if hm and len(hm.group(2)) < 120:
|
|
num = hm.group(1)
|
|
has_dots = "." in num
|
|
is_heading = has_dots or entry["size"] >= 12.0
|
|
|
|
# Phase 3b: Nicht-nummerierte Texte mit grosser Schrift pruefen,
|
|
# ob sie im TOC eine Kapitelnummer haben (fehlende PDF-Nummerierung)
|
|
if not is_heading and entry["size"] >= 12.0:
|
|
# Text gegen TOC abgleichen
|
|
toc_num = toc_numbers.get(text)
|
|
if toc_num:
|
|
# Nummer aus TOC voranstellen
|
|
candidate = f"{toc_num} {text}"
|
|
hm = RE_HEADING.match(candidate)
|
|
if hm:
|
|
text = candidate
|
|
is_heading = True
|
|
log.debug(" TOC-Nummer ergaenzt: %s", text)
|
|
|
|
if is_heading:
|
|
num = hm.group(1)
|
|
heading_text = f"{num} {hm.group(2)}"
|
|
# Level aus Tiefe der Nummerierung: 1=level1, 1.1=level2, etc.
|
|
depth = num.count(".") + 1
|
|
level = min(depth, 4)
|
|
char = RST_HEADING_CHARS[level]
|
|
rst_lines.append("")
|
|
if level == 1:
|
|
rst_lines.append(char * len(heading_text))
|
|
rst_lines.append(heading_text)
|
|
rst_lines.append(char * len(heading_text))
|
|
rst_lines.append("")
|
|
else:
|
|
rst_lines.append(text)
|
|
rst_lines.append("")
|
|
|
|
block_count += 1
|
|
|
|
# Mehrfache Leerzeilen reduzieren
|
|
rst = "\n".join(rst_lines)
|
|
rst = re.sub(r"\n{3,}", "\n\n", rst)
|
|
# 9-stellige Nummern mit Leerzeichen zusammenfuegen: "834 372 007" -> "834372007"
|
|
rst = re.sub(r"\b(\d{3})\s(\d{3})\s(\d{3})\b", r"\1\2\3", rst)
|
|
rst = re.sub(r"\b(\d{3})\s(\d{6})\b", r"\1\2", rst)
|
|
rst = re.sub(r"\b(\d{6})\s(\d{3})\b", r"\1\2", rst)
|
|
rst = rst.strip() + "\n"
|
|
|
|
rst_path.parent.mkdir(parents=True, exist_ok=True)
|
|
rst_path.write_text(rst, encoding="utf-8")
|
|
return block_count
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def dedup_images(img_dir: Path) -> int:
|
|
"""Entfernt doppelte Bilder per MD5-Hash aus einem Verzeichnis.
|
|
|
|
Returns:
|
|
Anzahl entfernter Duplikate.
|
|
"""
|
|
import hashlib
|
|
|
|
if not img_dir.exists():
|
|
return 0
|
|
|
|
seen: dict[str, Path] = {}
|
|
removed = 0
|
|
|
|
for img_path in sorted(img_dir.iterdir()):
|
|
if not img_path.is_file():
|
|
continue
|
|
md5 = hashlib.md5(img_path.read_bytes()).hexdigest()
|
|
if md5 in seen:
|
|
log.debug(" Duplikat entfernt: %s (== %s)", img_path.name, seen[md5].name)
|
|
img_path.unlink()
|
|
removed += 1
|
|
else:
|
|
seen[md5] = img_path
|
|
|
|
return removed
|
|
|
|
|
|
def process_all(data_dir: Path) -> None:
|
|
"""Verarbeitet alle konfigurierten PDFs."""
|
|
pp_dir = data_dir / "Produktprogramm"
|
|
images_base = pp_dir / "images"
|
|
|
|
if not pp_dir.exists():
|
|
log.error("Verzeichnis nicht gefunden: %s", pp_dir)
|
|
sys.exit(1)
|
|
|
|
pdf_files = list(pp_dir.glob("*.pdf"))
|
|
if not pdf_files:
|
|
log.error("Keine PDF-Dateien in %s gefunden", pp_dir)
|
|
sys.exit(1)
|
|
|
|
log.info("Gefunden: %d PDF-Dateien in %s", len(pdf_files), pp_dir)
|
|
|
|
for pdf_path in sorted(pdf_files):
|
|
target = find_target(pdf_path.name)
|
|
if target is None:
|
|
log.info("Uebersprungen (kein Ziel konfiguriert): %s", pdf_path.name)
|
|
continue
|
|
|
|
log.info("Verarbeite: %s -> %s", pdf_path.name, target)
|
|
|
|
# Bilder extrahieren und Duplikate entfernen
|
|
img_dir = images_base / target
|
|
img_count = extract_images(pdf_path, img_dir)
|
|
dup_count = dedup_images(img_dir)
|
|
log.info(" %d Bilder extrahiert, %d Duplikate entfernt -> %d einzigartig in %s",
|
|
img_count, dup_count, img_count - dup_count, img_dir)
|
|
|
|
# Text als RST extrahieren
|
|
rst_name = pdf_path.stem + ".de.rst"
|
|
rst_path = pp_dir / rst_name
|
|
block_count = extract_text_to_rst(pdf_path, rst_path)
|
|
log.info(" %d Textbloecke extrahiert nach %s", block_count, rst_path.name)
|
|
|
|
log.info("Fertig.")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Extrahiert Bilder und Text aus Produktprogramm-PDFs"
|
|
)
|
|
parser.add_argument("--data-dir", type=Path, default=None,
|
|
help="data/ Verzeichnis (default: aus PV_DATA oder ./data)")
|
|
parser.add_argument("--verbose", "-v", action="store_true",
|
|
help="Ausfuehrliche Ausgabe")
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(levelname)-7s %(message)s",
|
|
)
|
|
|
|
import os
|
|
if args.data_dir:
|
|
data_dir = args.data_dir.resolve()
|
|
elif os.environ.get("PV_DATA"):
|
|
data_dir = Path(os.environ["PV_DATA"]).resolve()
|
|
else:
|
|
data_dir = Path("data").resolve()
|
|
|
|
log.info("Data-Verzeichnis: %s", data_dir)
|
|
process_all(data_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|