Extrakt von Fortna gestartet um Standardsymbole und Basis für die ersten .rst Files zu erhalten.

This commit is contained in:
2026-04-29 22:55:30 +02:00
parent 367eb38cc8
commit cfc494d0dd
20 changed files with 1092 additions and 3 deletions
Binary file not shown.

After

Width:  |  Height:  |  Size: 50 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 227 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 29 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 96 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 60 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 57 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 62 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 13 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 9.3 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 33 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 183 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 4.8 KiB

+43
View File
@@ -0,0 +1,43 @@
@echo off
REM ================================================================
REM Fortna-Dokumente nach .rst konvertieren
REM
REM Konvertiert Word-Dokumente (.docx) und Word-erstellte PDFs
REM aus dem Fortna-Projektordner in .rst-Dateien und sortiert
REM sie in die chapters/-Struktur ein.
REM ================================================================
call "%~dp0setenv.bat"
REM Pruefe ob venv existiert
if not exist "%PROJECT%\.venv" (
echo FEHLER: .venv nicht gefunden. Bitte zuerst install_py.bat ausfuehren.
exit /b 1
)
call "%PROJECT%\.venv\Scripts\activate.bat"
REM Standard-Quellpfad (kann ueberschrieben werden)
set "FORTNA_DIR=C:\10-Develop\projekte\Fortna"
if not "%~1"=="" set "FORTNA_DIR=%~1"
REM Optionen weiterleiten
set "EXTRA_ARGS="
if "%~2"=="--dry-run" set "EXTRA_ARGS=--dry-run"
if "%~2"=="--verbose" set "EXTRA_ARGS=--verbose"
if "%~2"=="-v" set "EXTRA_ARGS=--verbose"
echo.
echo ================================================================
echo FORTNA Dokumenten-Konvertierung
echo ================================================================
echo Quelle: %FORTNA_DIR%
echo Ziel: %PV_CHAPTERS%
echo ================================================================
echo.
python "%PV_LIB%\convert_to_rst.py" "%FORTNA_DIR%" "%PV_CHAPTERS%" %EXTRA_ARGS%
echo.
echo Fertig.
deactivate
+35
View File
@@ -0,0 +1,35 @@
#!/usr/bin/env bash
# ================================================================
# Fortna-Dokumente nach .rst konvertieren
# ================================================================
# Dieses Skript muss gesourct werden: source bin/convert_fortna.sh
# oder direkt: bash bin/convert_fortna.sh [quellpfad] [--dry-run]
# ================================================================
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
source "$SCRIPT_DIR/setenv.sh"
if [ ! -d "$PROJECT/.venv" ]; then
echo "FEHLER: .venv nicht gefunden. Bitte zuerst install_py.sh ausfuehren."
exit 1
fi
source "$PROJECT/.venv/bin/activate"
FORTNA_DIR="${1:-C:/10-Develop/projekte/Fortna}"
EXTRA_ARGS="${2:-}"
echo ""
echo "================================================================"
echo "FORTNA Dokumenten-Konvertierung"
echo "================================================================"
echo "Quelle: $FORTNA_DIR"
echo "Ziel: $PV_CHAPTERS"
echo "================================================================"
echo ""
python "$PV_LIB/convert_to_rst.py" "$FORTNA_DIR" "$PV_CHAPTERS" $EXTRA_ARGS
echo ""
echo "Fertig."
deactivate
+20
View File
@@ -0,0 +1,20 @@
@echo off
REM ================================================================
REM DOCU_BUILD - Bilder und Text aus Produktprogramm-PDFs extrahieren
REM ================================================================
call "%~dp0setenv.bat"
if not exist "%PROJECT%\.venv" (
echo FEHLER: Virtual environment nicht gefunden.
echo Bitte zuerst bin\install_py.bat ausfuehren.
exit /b 1
)
call "%PROJECT%\.venv\Scripts\activate.bat"
echo.
echo Extrahiere Bilder und Text aus Produktprogramm-PDFs ...
echo.
python "%PV_LIB%\extract_products.py" --data-dir "%PV_DATA%" %*
+639
View File
@@ -0,0 +1,639 @@
"""
convert_to_rst.py - Konvertiert Word-Dokumente (.docx) und Word-erstellte PDFs nach .rst
Zwei-Stufen-Ansatz:
1. Textextraktion: .docx via mammoth -> HTML, PDF via pdfminer -> Text
2. Strukturerkennung: HTML/Text -> .rst mit Ueberschriften-Hierarchie
Verwendung:
python convert_to_rst.py <input_dir> <output_dir> [--dry-run] [--lang de]
"""
from __future__ import annotations
import argparse
import html as html_module
import logging
import os
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
# ---------------------------------------------------------------------------
# Konfiguration: Mapping Quelldatei -> Zielkapitel
# ---------------------------------------------------------------------------
# Muster im Dateinamen -> Zielkapitel-Pfad (relativ zu chapters/)
CHAPTER_MAPPING: list[tuple[re.Pattern[str], str]] = [
# 02_INSTRUCTIONS -> 04_betrieb, 05_wartung, 06_steuerung
(re.compile(r"betriebsanleitung", re.I), "04_betrieb/betriebsanleitung"),
(re.compile(r"wartungsanleitung|wartung", re.I), "05_wartung/wartungsanleitung"),
(re.compile(r"funktion.*(steuerung|control)|steuerungsbeschreibung", re.I), "06_steuerung"),
# 01_CERTIFICATES
(re.compile(r"konfigurationsbeschreibung|configuration.description", re.I), "01_zertifikate/konfigurationsbeschreibung"),
(re.compile(r"konformit|conformity|declaration", re.I), "01_zertifikate/konformitaetserklaerung"),
# 00_CONTENTS
(re.compile(r"contens|inhaltsverzeichnis|contents", re.I), "00_inhaltsverzeichnis"),
# 03_SPARE_PARTS -> 07_ersatzteile
(re.compile(r"spare.part|ersatzteil", re.I), "07_ersatzteile"),
# 05_REPORTS -> 10_anhang
(re.compile(r"wiederanlauf|drehrichtung|erprobung|protokoll|pr[uü]f", re.I), "10_anhang/pruefberichte"),
# Risikobewertung (RB_ prefix)
(re.compile(r"^RB_|risikobewertung|risikobeurteilung", re.I), "02_sicherheit/risikobewertung"),
]
# Benutzerdefinierte Word-Styles -> Heading-Level
CUSTOM_STYLE_MAP = {
"Ü1-ILS": 1, "Ü2-ILS": 2, "Ü3-ILS": 3, "Ü4-ILS": 4,
"Ü1": 1, "Ü2": 2, "Ü3": 3, "Ü4": 4,
}
# mammoth style_map String
MAMMOTH_STYLE_MAP = "\n".join([
"p[style-name='Ü1-ILS'] => h1:fresh",
"p[style-name='Ü2-ILS'] => h2:fresh",
"p[style-name='Ü3-ILS'] => h3:fresh",
"p[style-name='Ü4-ILS'] => h4:fresh",
"p[style-name='Heading 1'] => h1:fresh",
"p[style-name='Heading 2'] => h2:fresh",
"p[style-name='Heading 3'] => h3:fresh",
"p[style-name='Heading 4'] => h4:fresh",
"p[style-name='Doku-Text'] => p:fresh",
"p[style-name='Doku-1Aufzähl'] => ul > li:fresh",
"p[style-name='Doku-1Aufzählung'] => ul > li:fresh",
"p[style-name='Doku-2fettAufzählung'] => ul > li:fresh",
"p[style-name='Caption'] => p:fresh",
"p[style-name='Text'] => p:fresh",
"p[style-name='Doku-Text fett'] => p:fresh",
"p[style-name='Doku-Text12fett'] => p:fresh",
"p[style-name='Vorgabetext'] => p:fresh",
])
# rST Ueberschriften-Zeichen nach Level
RST_HEADING_CHARS = {1: "=", 2: "-", 3: "~", 4: "^", 5: '"'}
log = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Datenklassen
# ---------------------------------------------------------------------------
@dataclass
class RstBlock:
"""Ein Block im .rst-Dokument."""
kind: str # "heading", "paragraph", "list_item", "image", "table", "empty"
text: str = ""
level: int = 0 # nur fuer headings
image_path: str = "" # nur fuer images
@dataclass
class ConversionResult:
"""Ergebnis einer Konvertierung."""
source_path: Path
target_chapter: str
rst_filename: str
blocks: list[RstBlock] = field(default_factory=list)
images: dict[str, bytes] = field(default_factory=dict) # filename -> data
warnings: list[str] = field(default_factory=list)
# ---------------------------------------------------------------------------
# Kapitel-Erkennung
# ---------------------------------------------------------------------------
def detect_chapter(filename: str) -> str:
"""Bestimmt das Zielkapitel anhand des Dateinamens."""
for pattern, chapter in CHAPTER_MAPPING:
if pattern.search(filename):
return chapter
return "10_anhang" # Fallback
def detect_steuerung_subdir(filename: str) -> str:
"""Erkennt das Steuerungssystem fuer Unterordner in 06_steuerung/."""
lower = filename.lower()
if "loading" in lower or "loadingloop" in lower.replace("_", ""):
return "loading_loop"
if "omniflo" in lower:
return "omniflo"
if "ils" in lower:
return "ils"
return ""
def make_rst_filename(source_name: str) -> str:
"""Erzeugt einen sauberen .rst-Dateinamen."""
stem = Path(source_name).stem
# Versionsnummern und Datumsangaben vereinfachen
slug = stem.lower()
slug = re.sub(r"[^a-z0-9_]+", "_", slug)
slug = re.sub(r"_+", "_", slug).strip("_")
return slug + ".de.rst"
# ---------------------------------------------------------------------------
# Konvertierung: .docx -> HTML -> rST-Blocks
# ---------------------------------------------------------------------------
def convert_docx_to_blocks(docx_path: Path) -> tuple[list[RstBlock], dict[str, bytes], list[str]]:
"""Konvertiert .docx via mammoth nach rST-Blocks."""
import mammoth
blocks: list[RstBlock] = []
images: dict[str, bytes] = {}
warnings: list[str] = []
image_counter = 0
def handle_image(image):
nonlocal image_counter
image_counter += 1
ext = image.content_type.split("/")[-1] if image.content_type else "png"
if ext == "jpeg":
ext = "jpg"
img_name = f"img_{image_counter:03d}.{ext}"
with image.open() as img_file:
images[img_name] = img_file.read()
return {"src": img_name}
with open(docx_path, "rb") as f:
result = mammoth.convert_to_html(
f,
style_map=MAMMOTH_STYLE_MAP,
convert_image=mammoth.images.img_element(handle_image),
)
html = result.value
for msg in result.messages:
warnings.append(str(msg))
# HTML parsen -> Blocks
blocks = _html_to_blocks(html)
return blocks, images, warnings
def _html_to_blocks(html: str) -> list[RstBlock]:
"""Parst mammoth-HTML in RstBlock-Liste."""
blocks: list[RstBlock] = []
# Einfacher Regex-basierter Parser (mammoth erzeugt sauberes HTML)
# Aufteilung in Top-Level-Elemente
elements = re.split(r"(</?(?:h[1-6]|p|ul|ol|li|table|tr|td|th|img)[^>]*>)", html)
current_text = ""
current_kind = "paragraph"
current_level = 0
in_list = False
in_table = False
for elem in elements:
if not elem:
continue
# Oeffnende Tags
heading_match = re.match(r"<h([1-6])(?:\s[^>]*)?>", elem)
if heading_match:
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
current_kind = "heading"
current_level = int(heading_match.group(1))
continue
if re.match(r"</h[1-6]>", elem):
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
current_kind = "paragraph"
current_level = 0
continue
if re.match(r"<ul[^>]*>", elem):
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
in_list = True
continue
if re.match(r"</ul>", elem):
in_list = False
continue
if re.match(r"<li[^>]*>", elem):
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
current_kind = "list_item"
continue
if re.match(r"</li>", elem):
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
current_kind = "paragraph"
continue
if re.match(r"<p[^>]*>", elem):
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
if current_kind != "heading":
current_kind = "paragraph"
continue
if re.match(r"</p>", elem):
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
current_kind = "paragraph"
current_level = 0
continue
# Bilder
img_match = re.search(r'<img[^>]*src="([^"]+)"[^>]*/?\s*>', elem)
if img_match:
src = img_match.group(1)
if not src.startswith("data:"): # Skip base64 inline images
blocks.append(RstBlock(kind="image", image_path=src))
continue
# table-Tags: vereinfacht als Text-Block
if re.match(r"<table[^>]*>", elem):
_flush_block(blocks, current_kind, current_text, current_level)
current_text = ""
in_table = True
continue
if re.match(r"</table>", elem):
in_table = False
_flush_block(blocks, "paragraph", current_text, 0)
current_text = ""
continue
# Text-Inhalt: HTML-Tags entfernen
text = re.sub(r"<[^>]+>", "", elem)
text = html_module.unescape(text)
current_text += text
_flush_block(blocks, current_kind, current_text, current_level)
return blocks
def _flush_block(blocks: list[RstBlock], kind: str, text: str, level: int) -> None:
"""Fuegt einen Block hinzu wenn Text vorhanden."""
clean = text.strip()
if not clean:
return
# Tab-getrennte Inhalte bereinigen
clean = re.sub(r"\t+", " ", clean)
blocks.append(RstBlock(kind=kind, text=clean, level=level))
# ---------------------------------------------------------------------------
# Konvertierung: PDF -> Text -> rST-Blocks
# ---------------------------------------------------------------------------
def convert_pdf_to_blocks(pdf_path: Path) -> tuple[list[RstBlock], list[str]]:
"""Konvertiert PDF via pdfminer nach rST-Blocks."""
from pdfminer.high_level import extract_text
warnings: list[str] = []
try:
text = extract_text(str(pdf_path))
except Exception as e:
warnings.append(f"PDF-Extraktion fehlgeschlagen: {e}")
return [], warnings
blocks = _text_to_blocks(text)
return blocks, warnings
def _text_to_blocks(text: str) -> list[RstBlock]:
"""Heuristik: Plain-Text in Blocks mit Ueberschriften-Erkennung."""
blocks: list[RstBlock] = []
lines = text.split("\n")
i = 0
while i < len(lines):
line = lines[i].rstrip()
if not line.strip():
i += 1
continue
# Heuristik: Nummerierte Ueberschriften (1, 1.1, 1.1.1 etc.)
heading_match = re.match(r"^(\d+(?:\.\d+)*)\s+(.+)$", line.strip())
if heading_match:
num = heading_match.group(1)
title = heading_match.group(2).strip()
level = min(num.count(".") + 1, 4)
# Pruefen ob die naechste Zeile leer ist (typisch fuer Ueberschriften)
next_empty = (i + 1 < len(lines) and not lines[i + 1].strip())
# Oder ob der Text kurz genug fuer eine Ueberschrift ist
if next_empty or len(title) < 80:
blocks.append(RstBlock(kind="heading", text=f"{num} {title}", level=level))
i += 1
continue
# Aufzaehlungszeichen
list_match = re.match(r"^\s*[•\-]\s+(.+)$", line.strip())
if list_match:
blocks.append(RstBlock(kind="list_item", text=list_match.group(1).strip()))
i += 1
continue
# Normaler Absatz: Sammle zusammenhaengende Zeilen
para_lines = [line.strip()]
i += 1
while i < len(lines) and lines[i].strip() and not re.match(r"^\d+(?:\.\d+)*\s+", lines[i].strip()):
if re.match(r"^\s*[•\-]\s+", lines[i].strip()):
break
para_lines.append(lines[i].strip())
i += 1
para_text = " ".join(para_lines)
if para_text:
blocks.append(RstBlock(kind="paragraph", text=para_text))
return blocks
# ---------------------------------------------------------------------------
# Blocks -> .rst Rendering
# ---------------------------------------------------------------------------
def render_rst(blocks: list[RstBlock], title: str = "") -> str:
"""Rendert RstBlock-Liste als .rst-String."""
lines: list[str] = []
if title:
lines.append(RST_HEADING_CHARS[1] * len(title))
lines.append(title)
lines.append(RST_HEADING_CHARS[1] * len(title))
lines.append("")
prev_kind = ""
for block in blocks:
if block.kind == "heading":
text = block.text
if not text:
continue
level = max(1, min(block.level, 5))
char = RST_HEADING_CHARS[level]
lines.append("")
if level == 1:
lines.append(char * len(text))
lines.append(text)
lines.append(char * len(text))
lines.append("")
elif block.kind == "paragraph":
if prev_kind != "paragraph":
lines.append("")
lines.append(block.text)
lines.append("")
elif block.kind == "list_item":
lines.append(f"- {block.text}")
elif block.kind == "image":
lines.append("")
lines.append(f".. figure:: images/{block.image_path}")
lines.append(" :align: center")
lines.append("")
prev_kind = block.kind
# Bereinigung: Mehrfache Leerzeilen reduzieren
rst = "\n".join(lines)
rst = re.sub(r"\n{3,}", "\n\n", rst)
return rst.strip() + "\n"
# ---------------------------------------------------------------------------
# Erkennung: Ist PDF Word-erstellt?
# ---------------------------------------------------------------------------
def is_word_pdf(pdf_path: Path) -> bool:
"""Prueft ob ein PDF mit Microsoft Word erstellt wurde."""
from pdfminer.pdfparser import PDFParser
from pdfminer.pdfdocument import PDFDocument
try:
with open(pdf_path, "rb") as f:
parser = PDFParser(f)
doc = PDFDocument(parser)
if not doc.info:
return False
info = doc.info[0]
def decode(v: object) -> str:
if isinstance(v, bytes):
# UTF-16 BOM erkennen
if v[:2] in (b"\xff\xfe", b"\xfe\xff"):
return v.decode("utf-16", errors="replace")
return v.decode("utf-8", errors="replace")
return str(v) if v else ""
creator = decode(info.get("Creator", b""))
producer = decode(info.get("Producer", b""))
combined = (creator + " " + producer).lower()
return "word" in combined or "microsoft" in combined
except Exception:
return False
# ---------------------------------------------------------------------------
# Hauptlogik
# ---------------------------------------------------------------------------
def find_sources(input_dir: Path, lang: str = "de") -> list[Path]:
"""Findet alle konvertierbaren Quelldateien (DE)."""
sources: list[Path] = []
# 1. .docx am Root
for p in sorted(input_dir.glob("*.docx")):
sources.append(p)
# 2. Deutsche PDFs aus Unterordnern
lang_patterns = [f"*{lang}*", f"*{lang.upper()}*", f"*DE*", f"*_de_*"]
de_dirs = [f"*{lang}*", f"*{lang.upper()}*", "*DE*", "*original*"]
for pdf in sorted(input_dir.rglob("*.pdf")):
rel = pdf.relative_to(input_dir)
parts_lower = str(rel).lower()
# Nur DE-Dateien oder Dateien aus DE-Ordnern
is_de = False
for part in rel.parts:
pl = part.lower()
if "de" in pl.split("_") or pl.startswith("01_de") or pl.endswith("_de"):
is_de = True
break
if "_de." in parts_lower or "_de_" in parts_lower:
is_de = True
# Dateien im Root (ohne Sprachordner) sind auch DE
if len(rel.parts) <= 2:
is_de = True
if not is_de:
continue
# Nur Word-erstellte PDFs
if is_word_pdf(pdf):
sources.append(pdf)
return sources
def convert_file(source: Path, input_dir: Path, chapters_dir: Path) -> ConversionResult:
"""Konvertiert eine einzelne Datei und bestimmt Zielkapitel."""
filename = source.name
# Kapitel bestimmen
chapter = detect_chapter(filename)
# Spezialfall: Steuerung -> Unterordner
if chapter == "06_steuerung":
sub = detect_steuerung_subdir(filename)
if sub:
chapter = f"06_steuerung/{sub}"
rst_filename = make_rst_filename(filename)
result = ConversionResult(
source_path=source,
target_chapter=chapter,
rst_filename=rst_filename,
)
# Konvertierung
if source.suffix.lower() == ".docx":
blocks, images, warnings = convert_docx_to_blocks(source)
result.blocks = blocks
result.images = images
result.warnings = warnings
elif source.suffix.lower() == ".pdf":
blocks, warnings = convert_pdf_to_blocks(source)
result.blocks = blocks
result.warnings = warnings
else:
result.warnings.append(f"Unbekanntes Format: {source.suffix}")
return result
def write_result(result: ConversionResult, chapters_dir: Path, dry_run: bool = False) -> None:
"""Schreibt das Konvertierungsergebnis ins Dateisystem."""
target_dir = chapters_dir / result.target_chapter
target_file = target_dir / result.rst_filename
# Titel aus erstem Heading oder Dateiname
title = ""
for block in result.blocks:
if block.kind == "heading" and block.text:
title = block.text
break
if not title:
title = result.source_path.stem
rst_content = render_rst(result.blocks, title="")
if dry_run:
log.info("[DRY-RUN] %s -> %s (%d blocks, %d images)",
result.source_path.name, target_file.relative_to(chapters_dir),
len(result.blocks), len(result.images))
return
# Ordner anlegen
target_dir.mkdir(parents=True, exist_ok=True)
# .rst schreiben
target_file.write_text(rst_content, encoding="utf-8")
log.info("Geschrieben: %s (%d blocks)", target_file.relative_to(chapters_dir), len(result.blocks))
# Bilder speichern
if result.images:
img_dir = target_dir / "images"
img_dir.mkdir(exist_ok=True)
for img_name, img_data in result.images.items():
img_file = img_dir / img_name
img_file.write_bytes(img_data)
log.info(" %d Bilder nach %s", len(result.images), img_dir.relative_to(chapters_dir))
def main() -> None:
parser = argparse.ArgumentParser(
description="Konvertiert Word-Dokumente (.docx) und Word-PDFs nach .rst"
)
parser.add_argument("input_dir", type=Path,
help="Quellverzeichnis mit .docx/.pdf Dateien")
parser.add_argument("output_dir", type=Path,
help="Ziel-chapters/ Verzeichnis")
parser.add_argument("--dry-run", action="store_true",
help="Nur anzeigen, nichts schreiben")
parser.add_argument("--lang", default="de",
help="Zielsprache (default: de)")
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",
)
input_dir = args.input_dir.resolve()
chapters_dir = args.output_dir.resolve()
if not input_dir.exists():
log.error("Quellverzeichnis nicht gefunden: %s", input_dir)
sys.exit(1)
if not chapters_dir.exists():
log.error("Zielverzeichnis nicht gefunden: %s", chapters_dir)
sys.exit(1)
# Quellen finden
log.info("Suche Quelldateien in %s ...", input_dir)
sources = find_sources(input_dir, args.lang)
log.info("Gefunden: %d Dateien", len(sources))
if not sources:
log.warning("Keine konvertierbaren Dateien gefunden.")
sys.exit(0)
# Anzeige
for src in sources:
rel = src.relative_to(input_dir)
chapter = detect_chapter(src.name)
if chapter == "06_steuerung":
sub = detect_steuerung_subdir(src.name)
if sub:
chapter = f"06_steuerung/{sub}"
log.info(" %s -> %s", rel, chapter)
print()
# Konvertierung
success = 0
errors = 0
for src in sources:
try:
result = convert_file(src, input_dir, chapters_dir)
if result.blocks:
write_result(result, chapters_dir, dry_run=args.dry_run)
success += 1
else:
log.warning("Keine Inhalte extrahiert: %s", src.name)
errors += 1
for w in result.warnings:
log.warning(" %s: %s", src.name, w)
except Exception as e:
log.error("Fehler bei %s: %s", src.name, e)
errors += 1
print()
log.info("Fertig: %d erfolgreich, %d Fehler", success, errors)
if __name__ == "__main__":
main()
+348
View File
@@ -0,0 +1,348 @@
"""
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_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))
raw_lines: list[dict] = []
for page_num in range(len(doc)):
page = doc[page_num]
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", []):
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_header_footer(text, page_num + 1):
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.)
RE_HEADING = re.compile(r"^(\d+(?:\.\d+)*)\s+([A-Za-zÄÖÜäöüß].+)$")
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
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)
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 _is_header_footer(text: str, page_num: int) -> bool:
"""Erkennt typische Kopf-/Fusszeilen."""
lower = text.lower().strip()
# Copyright-Zeilen
if "copyright" in lower and "schönenberger" in lower:
return True
# Seitenzahlen
if re.match(r"^seite\s+\d+$", lower):
return True
# Version/Datum Fusszeilen
if re.match(r"^version\s+\d+", lower):
return True
# "Erstellt:" / "Freigabe:" Fusszeilen
if lower.startswith("erstellt:") or lower.startswith("freigabe:"):
return True
return False
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()
+7 -3
View File
@@ -1,6 +1,10 @@
# Python-Abhaengigkeiten
# Installieren mit: pip install -r requirements.txt
# Beispiel-Abhaengigkeiten anpassen nach Bedarf:
# pydantic >= 2.0.0
# pytest >= 9.0.0
# Dokumenten-Konvertierung
mammoth >= 1.8.0
pdfminer.six >= 20231228
PyMuPDF >= 1.24.0
# reStructuredText / Sphinx
docutils >= 0.21