7962999e8d
- "## 90", "## xxiv" (Fußnotenzeichen, römische Seitenzahlen ii–xcix)
wie Seitenzahlen entfernen, außer direkt vor einer Titel-Überschrift
oder in der PDF-Gliederung; "## IV." und Register-Buchstaben bleiben
- _mixed_block: zeilengetreue Blöcke zeilenweise prüfen; Konfiguration
([dependencies], name = "wert"), Befehle (cargo build, $ …) und
Programmausgaben (… ok, running n tests) als Codeblock, Anleitungen
als Text; kurze umbrochene Codezeilen erkennen
- Registereinträge ("match(), 247") nie als Code
- _raw_text: Leerzeichen bei Wortabstand ohne Leerzeichen-Zeichen
- Tests, README, Bedienungsanleitung, TODO
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2294 lines
87 KiB
Python
2294 lines
87 KiB
Python
#!/usr/bin/env python3
|
||
"""PDF (digital oder gescannt) nach Markdown und optional EPUB konvertieren.
|
||
|
||
Ablauf:
|
||
1. Jede Seite wird klassifiziert: Textseite, Scan ohne Text, Scan mit
|
||
vorhandener OCR-Ebene oder leer.
|
||
2. Seitengruppen gehen mit passender OCR-Einstellung an pymupdf4llm
|
||
(optional parallel).
|
||
3. normalize_markdown() gliedert Fließtext nach Satzenden, setzt
|
||
Überschriften anhand der PDF-Gliederung, fügt Silbentrennungen
|
||
zusammen und entfernt Seitenzahlen und Kolumnentitel.
|
||
4. Optional erzeugt pandoc daraus ein EPUB.
|
||
"""
|
||
|
||
import argparse
|
||
import contextlib
|
||
import io
|
||
import json
|
||
import multiprocessing
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import unicodedata
|
||
import urllib.parse
|
||
from collections import Counter
|
||
from concurrent.futures import ProcessPoolExecutor
|
||
from pathlib import Path
|
||
|
||
import pymupdf
|
||
import pymupdf4llm
|
||
from pymupdf4llm.helpers import document_layout
|
||
from pymupdf4llm.ocr import tesseract_api
|
||
|
||
|
||
# ===========================================================================
|
||
# Markdown-Muster
|
||
# ===========================================================================
|
||
|
||
HEADING = re.compile(r"^\s{0,3}(#{1,6})\s*(\S.*?)\s*#*\s*$")
|
||
|
||
# Schließende Anführungszeichen (deutsch „…“ ‚…‘ »…«, englisch, Schweiz «…»),
|
||
# Klammern und Markdown-Hervorhebungen.
|
||
CLOSERS = r"""”“’‘"'»«›‹)\]*_"""
|
||
|
||
# Satzzeichen, optional gefolgt von schließenden Zeichen (z. B. "…da?«").
|
||
SENTENCE_END = re.compile(rf"""[.!?…]+[{CLOSERS}]*$""")
|
||
|
||
# Typische Endungen von Sätzen, die fälschlich als Überschrift erkannt wurden:
|
||
# Punkt/Auslassung oder Frage/Ausruf mit schließendem Anführungszeichen.
|
||
SENTENCE_LIKE_END = re.compile(rf"""(?:[.…][{CLOSERS}]*|[!?]+[{CLOSERS}]+)$""")
|
||
|
||
# Überschriften, die legitim mit Punkt enden: "3.", "IV.", "Kapitel 3."
|
||
NUMBERING_ONLY = re.compile(
|
||
r"^(?:(?:kapitel|teil|chapter|part)\s+)?(?:\d+|[ivxlcdm]+)\.$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
FENCE = re.compile(r"^\s{0,3}(`{3,}|~{3,})")
|
||
LIST_ITEM = re.compile(r"^\s{0,3}(?:[-+*]|\d+[.)])\s+")
|
||
ORDERED_ITEM = re.compile(r"^\s{0,3}\d+[.)]\s+")
|
||
BLOCKQUOTE = re.compile(r"^\s{0,3}>\s?")
|
||
RULE = re.compile(r"^\s{0,3}(?:-\s*){3,}$|^\s{0,3}(?:\*\s*){3,}$")
|
||
IMAGE = re.compile(r"^\s*!\[.*?\]\(.*\)\s*$")
|
||
|
||
# Beispiel: **==> picture [426 x 480] intentionally omitted <==**
|
||
OMITTED_IMAGE = re.compile(
|
||
r"^\s*(?:\*\*|__)?\s*==>\s*"
|
||
r"(?:picture|image|bild)\b.*?"
|
||
r"(?:intentionally omitted|omitted|ausgelassen)\s*"
|
||
r"<==\s*(?:\*\*|__)?\s*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# Text, den pymupdf4llm innerhalb von Bildern findet (Fotos, OCR-Reste).
|
||
PICTURE_TEXT = re.compile(
|
||
r"\*\*----- Start of picture text -----\*\*<br>.*?"
|
||
r"\*\*----- End of picture text -----\*\*<br>\n?",
|
||
re.DOTALL,
|
||
)
|
||
|
||
CONTENTS_TITLE = re.compile(
|
||
r"^(?:inhaltsverzeichnis|inhalt|contents|table of contents)$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
TOC_WITH_PAGE = re.compile(
|
||
r"^\s*(?:[-*]\s+)?\S.+?"
|
||
r"(?:\.{2,}|\s{2,})"
|
||
r"(?:\d+|[ivxlcdm]+)\s*$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# Beginn eines nummerierten Eintrags mitten in einer Inhaltsverzeichniszeile.
|
||
NEXT_TOC_NUMBER = re.compile(r"\s(?=\d{1,3}\s+[^\W\d_])")
|
||
|
||
HTML_BLOCK = re.compile(
|
||
r"^\s*</?(?:table|tr|td|th|div|pre|p|h[1-6])\b",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# Seitenzahl allein in einer Zeile: "12", "- 12 -", "Seite 12"
|
||
PAGE_NUMBER = re.compile(
|
||
r"^(?:[-–—]\s*)?(?:(?:seite|page|s\.)\s*)?\d{1,4}(?:\s*[-–—])?$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
QUOTE_CHARS = set("«»„“”\"‚‘’›‹")
|
||
DASHES = str.maketrans({"–": "-", "—": "-", "‐": "-", "‑": "-"})
|
||
|
||
# Silbentrennung am Zeilen-/Blockende (Bindestrich, weicher Trennstrich, ¬).
|
||
HYPHEN_END = re.compile(r"[^\W\d_]{2,}[-\u00ad¬]$")
|
||
FIRST_WORD = re.compile(r"[^\W\d_]+")
|
||
|
||
# Nach "Arbeits-" folgt "und Krieger-Skills": Bindestrich + Leerzeichen bleiben.
|
||
CONJUNCTIONS = {
|
||
"und", "oder", "bis", "sowie", "als", "wie", "bzw", "noch",
|
||
"beziehungsweise", "respektive", "and", "or", "to",
|
||
}
|
||
|
||
# Wörter, mit denen eine echte Überschrift praktisch nie endet (Artikel,
|
||
# Konjunktionen). Präpositionen fehlen bewusst: "Die Spur nehmen sie auf".
|
||
TRAILING_STOPWORDS = {
|
||
"der", "die", "das", "den", "dem", "des", "ein", "eine", "einen", "einem",
|
||
"einer", "eines", "und", "oder", "aber", "dass", "daß", "sowie",
|
||
"the", "a", "an", "and", "or", "of",
|
||
}
|
||
|
||
# Gliederungseinträge für Teile bzw. nummerierte Kapitel.
|
||
PART_TITLE = re.compile(
|
||
r"^(?:teil|buch|band|part|book)\s+(?:[ivxlcdm]+|\d+)\b",
|
||
re.IGNORECASE,
|
||
)
|
||
# Zeile, die wie ein Gliederungseintrag aussieht ("Teil V: …", "32 Oberst …").
|
||
OUTLINE_ENTRY = re.compile(
|
||
r"^(?:\d{1,3}\.?\s+\S|(?:teil|buch|band|part|book|kapitel|chapter)\s+(?:[ivxlcdm]+|\d+)\b)",
|
||
re.IGNORECASE,
|
||
)
|
||
NUMBERED_TITLE = re.compile(
|
||
r"^(?:\d+\.?|kapitel\s+\d+|chapter\s+\d+)\s",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
# ===========================================================================
|
||
# Hilfsfunktionen für Text
|
||
# ===========================================================================
|
||
|
||
# Markdown-Sonderzeichen im PDF-Text maskieren, damit z. B. "__add__" nicht fett
|
||
# wird. "_" mitten im Wort ("snake_case") bleibt unmaskiert.
|
||
MD_SPECIAL = "\\`*_[]<>$~^"
|
||
MD_ESCAPE = re.compile(r"[\\`*\[\]<>$~^]|(?<![^\W_])_|_(?![^\W_])")
|
||
MD_UNESCAPE = re.compile(r"\\([\\`*_\[\]<>$~^])")
|
||
|
||
|
||
def md_escape(text: str) -> str:
|
||
return MD_ESCAPE.sub(lambda m: "\\" + m.group(0), text)
|
||
|
||
|
||
def md_unescape(text: str) -> str:
|
||
return MD_UNESCAPE.sub(r"\1", text)
|
||
|
||
|
||
def ends_sentence(text: str) -> bool:
|
||
return bool(SENTENCE_END.search(text.rstrip()))
|
||
|
||
|
||
def clean_inline(text: str) -> str:
|
||
"""Markdown-Hervorhebungen entfernen und Leerraum vereinheitlichen."""
|
||
text = re.sub(r"(?<!\\)\*\*|(?<!\\)__|(?<![\w\\])_|(?<!\\)_(?!\w)", "", text)
|
||
return re.sub(r"\s+", " ", text).strip()
|
||
|
||
|
||
def norm_title(text: str) -> str:
|
||
"""Vergleichsschlüssel für Überschriften und Gliederungseinträge."""
|
||
return md_unescape(clean_inline(text)).translate(DASHES).lower()
|
||
|
||
|
||
def is_other_structure(line: str) -> bool:
|
||
"""Markdown-Strukturen, die nicht zu Fließtext verbunden werden."""
|
||
return bool(
|
||
LIST_ITEM.match(line)
|
||
or BLOCKQUOTE.match(line)
|
||
or RULE.match(line)
|
||
or IMAGE.match(line)
|
||
or HTML_BLOCK.match(line)
|
||
or line.lstrip().startswith("|")
|
||
or line.startswith((" ", "\t"))
|
||
)
|
||
|
||
|
||
def join_parts(prev: str, nxt: str) -> str:
|
||
"""Zwei Teile eines Absatzes verbinden und Silbentrennung auflösen.
|
||
|
||
"Geheim-" + "dienst" -> "Geheimdienst"
|
||
"Militär-" + "Geheimdienst" -> "Militär-Geheimdienst"
|
||
"Arbeits-" + "und …" -> "Arbeits- und …"
|
||
"""
|
||
if not HYPHEN_END.search(prev) or not nxt:
|
||
return f"{prev} {nxt}"
|
||
|
||
stem = prev[:-1]
|
||
word = FIRST_WORD.match(nxt)
|
||
if word and word.group(0).lower() in CONJUNCTIONS:
|
||
return f"{stem}- {nxt}"
|
||
if nxt[0].islower():
|
||
return stem + nxt
|
||
return f"{stem}-{nxt}"
|
||
|
||
|
||
def is_false_heading(text: str, has_outline: bool, paragraph_open: bool) -> bool:
|
||
"""Erkennen, ob eine "Überschrift" in Wahrheit ein Satz(-rest) ist."""
|
||
if not text:
|
||
return True
|
||
if text[0].islower() or text[0] in "–—-…,;" or text[-1] in ",;":
|
||
return True
|
||
if NUMBERING_ONLY.match(text):
|
||
return False
|
||
if paragraph_open and ends_sentence(text):
|
||
return True
|
||
if paragraph_open and text.split()[-1].lower() in TRAILING_STOPWORDS:
|
||
return True # "Mitglied des Künstlerverbandes – damals eine"
|
||
if SENTENCE_LIKE_END.search(text):
|
||
return True
|
||
# Mit Gliederung stehen echte Überschriften im Outline; was dort fehlt
|
||
# und wie ein Satz endet, ist Fließtext.
|
||
return has_outline and ends_sentence(text)
|
||
|
||
|
||
def running_head_key(line: str) -> str | None:
|
||
"""Vergleichsschlüssel für mögliche Kolumnentitel (sonst None)."""
|
||
if (
|
||
HEADING.match(line)
|
||
or OMITTED_IMAGE.match(line)
|
||
or is_other_structure(line)
|
||
or FENCE.match(line)
|
||
):
|
||
return None
|
||
text = clean_inline(line)
|
||
if not text or len(text) > 60 or ends_sentence(text):
|
||
return None
|
||
if QUOTE_CHARS & set(text):
|
||
return None # wörtliche Rede ist kein Kolumnentitel
|
||
key = re.sub(r"\d+", "", text)
|
||
key = re.sub(r"\s+", " ", key).strip(" .-–—|·").lower()
|
||
if sum(ch.isalpha() for ch in key) < 4:
|
||
return None
|
||
return key
|
||
|
||
|
||
def find_running_heads(lines: list[str], min_repeats: int = 3) -> set[str]:
|
||
"""Kurze Zeilen, die (ohne Ziffern) sehr oft vorkommen."""
|
||
counts = Counter(k for k in map(running_head_key, lines) if k)
|
||
return {key for key, count in counts.items() if count >= min_repeats}
|
||
|
||
|
||
def split_contents_line(line: str, titles: list[str]) -> list[str]:
|
||
"""Zusammengeklebte Inhaltsverzeichniseinträge trennen.
|
||
|
||
titles: normalisierte Gliederungstitel, längste zuerst.
|
||
"""
|
||
text = clean_inline(line)
|
||
entries = []
|
||
|
||
while text:
|
||
low = text.translate(DASHES).lower()
|
||
|
||
match = next(
|
||
(
|
||
t for t in titles
|
||
if low.startswith(t) and (len(low) == len(t) or low[len(t)] == " ")
|
||
),
|
||
None,
|
||
)
|
||
if match:
|
||
entries.append(text[: len(match)])
|
||
text = text[len(match):].strip()
|
||
continue
|
||
|
||
# Nächsten bekannten Titel oder nächste Kapitelnummer suchen.
|
||
cuts = [
|
||
pos for t in titles
|
||
if (pos := low.find(" " + t, 1)) > 0
|
||
and (pos + 1 + len(t) == len(low) or low[pos + 1 + len(t)] == " ")
|
||
]
|
||
number = NEXT_TOC_NUMBER.search(text)
|
||
if number:
|
||
cuts.append(number.start())
|
||
if not cuts:
|
||
entries.append(text)
|
||
break
|
||
cut = min(cuts)
|
||
entries.append(text[:cut].strip())
|
||
text = text[cut:].strip()
|
||
|
||
return [e for e in entries if e]
|
||
|
||
|
||
# ===========================================================================
|
||
# Markdown normalisieren
|
||
# ===========================================================================
|
||
|
||
def normalize_markdown(
|
||
markdown: str,
|
||
outline: dict[str, int] | None = None,
|
||
*,
|
||
drop_page_numbers: bool = True,
|
||
drop_running_heads: bool = True,
|
||
running_head_min: int = 3,
|
||
) -> str:
|
||
outline = outline or {}
|
||
titles = sorted(outline, key=len, reverse=True)
|
||
|
||
markdown = PICTURE_TEXT.sub("", markdown)
|
||
lines = markdown.replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
||
running_heads = (
|
||
find_running_heads(lines, running_head_min) if drop_running_heads else set()
|
||
)
|
||
|
||
result: list[str] = []
|
||
paragraph: list[str] = []
|
||
|
||
in_fence = False
|
||
fence_char = ""
|
||
fence_length = 0
|
||
in_contents = False
|
||
contents_seen: set[str] = set() # Titel, die im Inhaltsverzeichnis standen
|
||
contents_table = False # Inhaltsverzeichnis liegt als Markdown-Tabelle vor
|
||
headings_seen: set[str] = set()
|
||
|
||
def paragraph_open() -> bool:
|
||
return bool(paragraph) and not ends_sentence(paragraph[-1])
|
||
|
||
def blank_line() -> None:
|
||
if result and result[-1] != "":
|
||
result.append("")
|
||
|
||
def flush_paragraph() -> None:
|
||
if paragraph:
|
||
text = paragraph[0]
|
||
for part in paragraph[1:]:
|
||
text = join_parts(text, part)
|
||
if result and result[-1].endswith(HARD_BREAK):
|
||
result.append("") # Fließtext nach zeilengetreuen Zeilen
|
||
result.append(text)
|
||
paragraph.clear()
|
||
|
||
def standalone(line: str) -> None:
|
||
flush_paragraph()
|
||
blank_line()
|
||
result.append(line)
|
||
result.append("")
|
||
|
||
def heading(level: int, text: str) -> None:
|
||
headings_seen.add(norm_title(text))
|
||
standalone("#" * min(max(level, 1), 6) + " " + text)
|
||
|
||
for index, original in enumerate(lines):
|
||
line = original.rstrip()
|
||
marker = FENCE.match(line)
|
||
if marker and not in_fence and MARKERS.search(line):
|
||
# Text aus einem Absatz, der mit ``` beginnt: kein Codezaun.
|
||
line = "\\" + line
|
||
marker = None
|
||
|
||
if in_fence:
|
||
result.append(line)
|
||
if marker:
|
||
token = marker.group(1)
|
||
if token[0] == fence_char and len(token) >= fence_length:
|
||
in_fence = False
|
||
continue
|
||
|
||
if marker:
|
||
flush_paragraph()
|
||
in_fence = True
|
||
fence_char = marker.group(1)[0]
|
||
fence_length = len(marker.group(1))
|
||
# Listing über eine Seitengrenze: an den vorigen Codeblock anhängen.
|
||
while result and result[-1] == "":
|
||
result.pop()
|
||
if result and result[-1] == line.strip() and line.strip() == fence_char * fence_length:
|
||
result.pop()
|
||
continue
|
||
blank_line()
|
||
result.append(line)
|
||
continue
|
||
|
||
if end_mark := BLOCK_END_MARK.search(line):
|
||
# Blockende ohne Satzende: Hätte das erste Wort des nächsten Blocks
|
||
# noch in die Zeile gepasst, war der Umbruch hart.
|
||
line = line[: end_mark.start()]
|
||
if not continues_block(lines[index + 1:], int(end_mark.group(1)), running_heads):
|
||
# Mehrzeiliger Absatz: Absatzende; einzelne Zeile: harter Umbruch
|
||
line += PARAGRAPH_END if end_mark.group(2) else HARD_BREAK
|
||
|
||
if in_contents and line.endswith((HARD_BREAK, PARAGRAPH_END)):
|
||
# Inhaltsverzeichnis: jeder Eintrag ein eigener Absatz (einheitlich,
|
||
# auch wenn das Verzeichnis in der PDF zeilengetreu erscheint).
|
||
line = line.rstrip(HARD_BREAK + PARAGRAPH_END)
|
||
elif line.endswith((HARD_BREAK, PARAGRAPH_END)) and OUTLINE_ENTRY.match(
|
||
clean_inline(line.rstrip(HARD_BREAK + PARAGRAPH_END))
|
||
):
|
||
# Verzeichnis ohne Überschrift "Inhalt" (z. B. "Vorschau Band Zwei"):
|
||
# Einträge als eigene Absätze statt als Zeilen mit "\".
|
||
flush_paragraph()
|
||
standalone(clean_inline(line.rstrip(HARD_BREAK + PARAGRAPH_END)))
|
||
continue
|
||
|
||
if line.endswith(PARAGRAPH_END):
|
||
content = line.removesuffix(PARAGRAPH_END)
|
||
if not paragraph and result and result[-1].endswith(HARD_BREAK):
|
||
result.append(content) # letzte Zeile eines zeilengetreuen Blocks
|
||
else:
|
||
if not paragraph_open():
|
||
flush_paragraph()
|
||
blank_line()
|
||
paragraph.append(content.strip())
|
||
flush_paragraph()
|
||
blank_line()
|
||
continue
|
||
|
||
if line.endswith(HARD_BREAK):
|
||
# Gliederungstitel oder "Inhalt" bleiben Überschriften.
|
||
check = clean_inline(line.removesuffix(HARD_BREAK))
|
||
key = norm_title(check)
|
||
if CONTENTS_TITLE.match(check) or (
|
||
key in outline and key not in headings_seen and not in_contents
|
||
):
|
||
line = line.removesuffix(HARD_BREAK)
|
||
|
||
if line.endswith(HARD_BREAK) or line == STANZA_BREAK:
|
||
# Zeilengetreue Zeile: Einrückung (geschützte Leerzeichen) behalten.
|
||
content = line.removesuffix(HARD_BREAK)
|
||
check = clean_inline(content)
|
||
if in_contents:
|
||
contents_seen.add(norm_title(check))
|
||
if drop_page_numbers and PAGE_NUMBER.match(check):
|
||
continue
|
||
if running_heads and running_head_key(check) in running_heads:
|
||
continue
|
||
if paragraph_open() and line != STANZA_BREAK:
|
||
# Letzte Zeile eines Absatzes aus dem vorigen Block
|
||
paragraph.append(content.strip())
|
||
flush_paragraph()
|
||
result[-1] += HARD_BREAK
|
||
continue
|
||
flush_paragraph()
|
||
while result and result[-1] == "" and len(result) > 1 and result[-2].endswith(HARD_BREAK):
|
||
result.pop() # Zeilen aus getrennten Blöcken zusammenhalten
|
||
if line == STANZA_BREAK:
|
||
if result and result[-1].endswith(HARD_BREAK):
|
||
result[-1] = result[-1].removesuffix(HARD_BREAK)
|
||
blank_line()
|
||
continue
|
||
if not result or result[-1] == "" or result[-1].endswith(HARD_BREAK):
|
||
result.append(line)
|
||
else:
|
||
blank_line()
|
||
result.append(line)
|
||
continue
|
||
|
||
if not line.strip(" \t"):
|
||
# Leerzeile nur dann als Absatzgrenze behandeln,
|
||
# wenn der bisherige Fließtext mit einem Satzende endet.
|
||
if paragraph and ends_sentence(paragraph[-1]):
|
||
flush_paragraph()
|
||
blank_line()
|
||
elif not paragraph:
|
||
blank_line()
|
||
continue
|
||
|
||
stripped = line.strip()
|
||
|
||
if not in_contents:
|
||
if drop_page_numbers and PAGE_NUMBER.match(clean_inline(stripped)):
|
||
continue
|
||
if running_heads and running_head_key(stripped) in running_heads:
|
||
continue
|
||
|
||
match = HEADING.match(line)
|
||
text = clean_inline(match.group(2)) if match else clean_inline(stripped)
|
||
|
||
if CONTENTS_TITLE.match(text):
|
||
heading(len(match.group(1)) if match else 2, text)
|
||
in_contents = True
|
||
contents_table = False
|
||
continue
|
||
|
||
if match:
|
||
key = norm_title(text)
|
||
level = outline.get(key)
|
||
|
||
if in_contents and level is not None and key not in contents_seen:
|
||
# Überschrift-formatierter Eintrag im Inhaltsverzeichnis
|
||
contents_seen.add(key)
|
||
standalone(text)
|
||
continue
|
||
|
||
if level is None and NUMBER_HEADING.match(text) and not heading_follows(
|
||
lines, index
|
||
):
|
||
# "## 90", "## XXIV": Fußnotenzeichen oder Seitenzahl, keine
|
||
# Kapitelnummer (die steht direkt vor einer Titel-Überschrift).
|
||
if drop_page_numbers:
|
||
continue
|
||
standalone(text)
|
||
continue
|
||
|
||
if level is None and is_false_heading(
|
||
text, bool(outline), paragraph_open()
|
||
):
|
||
# Satz(-rest) zurück in den Fließtext: setzt einen offenen
|
||
# Absatz fort, sonst beginnt er einen neuen.
|
||
in_contents = False
|
||
if paragraph and not paragraph_open():
|
||
flush_paragraph()
|
||
blank_line()
|
||
paragraph.append(text)
|
||
continue
|
||
|
||
in_contents = False
|
||
heading(level or len(match.group(1)), text)
|
||
continue
|
||
|
||
if OMITTED_IMAGE.match(line):
|
||
standalone(line)
|
||
continue
|
||
|
||
if in_contents:
|
||
if stripped.startswith("|"):
|
||
# Verzeichnis als Tabelle: Zeilen unverändert übernehmen,
|
||
# sonst zerfällt die Tabelle in einzelne Absätze.
|
||
if not contents_table:
|
||
flush_paragraph()
|
||
blank_line()
|
||
contents_table = True
|
||
for cell in stripped.strip("|").split("|"):
|
||
contents_seen.add(norm_title(cell))
|
||
result.append(line)
|
||
continue
|
||
if contents_table:
|
||
in_contents = False # Tabelle zu Ende: Fließtext folgt
|
||
blank_line()
|
||
elif ends_sentence(text) and len(text) > 80:
|
||
in_contents = False # erster Fließtext nach dem Verzeichnis
|
||
else:
|
||
for entry in split_contents_line(stripped, titles):
|
||
contents_seen.add(norm_title(entry))
|
||
standalone(entry)
|
||
continue
|
||
|
||
if TOC_WITH_PAGE.match(line):
|
||
standalone(line)
|
||
continue
|
||
|
||
# Eigenständige Zeile, die genau einem Gliederungstitel entspricht
|
||
# (z. B. nur fett gesetzt): zur Überschrift machen.
|
||
key = norm_title(stripped)
|
||
if key in outline and key not in headings_seen and len(stripped) <= 150:
|
||
heading(outline[key], text)
|
||
continue
|
||
|
||
# "Die Mauer fiel am" + "9. November …" ist keine Liste,
|
||
# "Liste:" + "1. Erstens" dagegen schon.
|
||
continues_sentence = paragraph_open() and paragraph[-1][-1:].isalnum()
|
||
if is_other_structure(line) and not (
|
||
ORDERED_ITEM.match(line) and continues_sentence
|
||
):
|
||
if paragraph:
|
||
flush_paragraph()
|
||
blank_line() # pandoc braucht eine Leerzeile vor Listen/Tabellen
|
||
result.append(line)
|
||
continue
|
||
|
||
# Einfache Zeilenumbrüche und nicht bestätigte Absatzgrenzen
|
||
# werden beim Flush durch Leerzeichen ersetzt.
|
||
paragraph.append(stripped)
|
||
|
||
flush_paragraph()
|
||
return finish_hard_breaks("\n".join(result)).strip() + "\n"
|
||
|
||
|
||
# Überschrift nur aus einer Zahl ohne Punkt: arabisch oder römisch ii–xcix
|
||
# (mindestens zwei Zeichen, damit Register-Buchstaben "C", "D", "M" bleiben).
|
||
NUMBER_HEADING = re.compile(
|
||
r"^(?:\d{1,4}|(?=[ivxlc]{2})(?:xc|xl|l?x{0,3})(?:ix|iv|v?i{0,3}))$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def heading_follows(lines: list[str], index: int) -> bool:
|
||
"""Ist die nächste nicht leere Zeile eine Überschrift?"""
|
||
following = next((line for line in lines[index + 1:] if line.strip()), "")
|
||
return bool(HEADING.match(following))
|
||
|
||
|
||
def continues_block(following: list[str], free: int, running_heads: set[str]) -> bool:
|
||
"""Geht der Satz im nächsten Block weiter? (erstes Wort passte nicht mehr)"""
|
||
for line in following:
|
||
text = line.strip(" \t")
|
||
if not text or PAGE_NUMBER.match(clean_inline(text)):
|
||
continue
|
||
if running_heads and running_head_key(text) in running_heads:
|
||
continue
|
||
if heading := HEADING.match(text):
|
||
text = heading.group(2) # oft falsch erkannt ("## trösten.")
|
||
elif text.startswith(("|", "!", ">", "```", "~~~", "**==>")):
|
||
return False
|
||
elif LIST_ITEM.match(text) and not ORDERED_ITEM.match(text):
|
||
return False # "14. Jahrhundert" ist dagegen oft keine Liste
|
||
word = md_unescape(clean_inline(text)).replace(NBSP, " ").split()
|
||
# Breite aus der vorigen Zeile geschätzt: ein Zeichen Spielraum.
|
||
return bool(word) and len(word[0]) + 2 > free
|
||
return False
|
||
|
||
|
||
def finish_hard_breaks(markdown: str) -> str:
|
||
"""Markierungen harter Umbrüche in Markdown umsetzen.
|
||
|
||
Folgt eine weitere zeilengetreue Zeile, wird daraus "\" (harter Umbruch);
|
||
die letzte Zeile eines Abschnitts endet ohne.
|
||
"""
|
||
lines = markdown.split("\n")
|
||
for i, line in enumerate(lines):
|
||
if line.endswith(HARD_BREAK):
|
||
text = line.removesuffix(HARD_BREAK).rstrip()
|
||
following = lines[i + 1] if i + 1 < len(lines) else ""
|
||
if following.strip(" \t") and not FENCE.match(following):
|
||
text += "\\"
|
||
lines[i] = text
|
||
return MARKERS.sub("", "\n".join(lines)) # Sicherheitsnetz
|
||
|
||
|
||
# ===========================================================================
|
||
# PDF-Gliederung
|
||
# ===========================================================================
|
||
|
||
def load_outline(doc: pymupdf.Document) -> dict[str, int]:
|
||
"""PDF-Lesezeichen als {normalisierter Titel: Überschriftenebene}."""
|
||
original: dict[str, int] = {}
|
||
for level, title, _page in doc.get_toc(simple=True):
|
||
key = norm_title(title)
|
||
if key:
|
||
original.setdefault(key, level)
|
||
|
||
# Teile und nummerierte Kapitel stehen oft auf derselben Ebene;
|
||
# dann rücken die Kapitel eine Ebene tiefer.
|
||
levels = dict(original)
|
||
for level in set(original.values()):
|
||
keys = [k for k, v in original.items() if v == level]
|
||
if any(PART_TITLE.match(k) for k in keys):
|
||
for key in keys:
|
||
if NUMBERED_TITLE.match(key) and not PART_TITLE.match(key):
|
||
levels[key] = level + 1
|
||
return levels
|
||
|
||
|
||
# ===========================================================================
|
||
# Links
|
||
# ===========================================================================
|
||
# pymupdf4llm übernimmt keine Link-Annotationen, und gedruckte URLs zerbrechen
|
||
# beim Zeilenumbruch ("deutsche-|sehen" -> "deutschesehen", ".../ youtube-...").
|
||
# Wir gleichen URLs im Text mit den Linkzielen der PDF ab, ersetzen sie durch
|
||
# das korrekte Ziel und schreiben sie als Autolink <https://…>, den pandoc im
|
||
# EPUB als echten Link ausgibt.
|
||
|
||
URL_START = re.compile(
|
||
r"(?<![<\]/A-Za-z0-9])(?<!\]\()(?:https?://|www\.)", re.IGNORECASE
|
||
) # nicht in <…>-Links und [Text](…)-Zielen
|
||
URL_TRAILING = ".,;:!?)]}»«“”„\"'*_\\" # "\\": harter Zeilenumbruch
|
||
MAX_URL_PARTS = 4 # eine URL kann über so viele Zeilen gebrochen sein
|
||
# Ohne Linkziel: Folgewort gehört zur URL, wenn diese an einer typischen
|
||
# Umbruchstelle endet und das Wort wie ein URL-Teil aussieht.
|
||
URL_BREAK_AFTER = "/_-.=&?#"
|
||
URL_CONTINUATION = re.compile(r"^(?:\S*/\S*|(?:html?|php|pdf|aspx?)\W*)$", re.IGNORECASE)
|
||
|
||
|
||
def url_key(text: str) -> str:
|
||
"""Vergleichsschlüssel: Bindestriche, Leerraum und "/" am Ende zählen nicht."""
|
||
return re.sub(r"[-\s]", "", text).rstrip("/").lower()
|
||
|
||
|
||
def uri_keys(uri: str) -> set[str]:
|
||
"""Schlüssel, unter denen ein Linkziel im gedruckten Text auftauchen kann.
|
||
|
||
Gedruckt fehlen oft Abfrageteil ("?utm_source=…") oder Endung (".html").
|
||
"""
|
||
base = uri.split("#")[0]
|
||
keys = set()
|
||
for variant in (uri, base, base.split("?")[0]):
|
||
keys.add(url_key(variant))
|
||
keys.add(url_key(re.sub(r"\.(?:html?|php|aspx?)$", "", variant.rstrip("/"))))
|
||
return keys
|
||
|
||
|
||
def clean_uri(uri: str) -> str:
|
||
"""Tracking-Parameter (utm_…) entfernen."""
|
||
parts = urllib.parse.urlsplit(uri)
|
||
if "utm_" not in parts.query.lower():
|
||
return uri
|
||
query = [
|
||
(k, v) for k, v in urllib.parse.parse_qsl(parts.query, keep_blank_values=True)
|
||
if not k.lower().startswith("utm_")
|
||
]
|
||
return urllib.parse.urlunsplit(parts._replace(query=urllib.parse.urlencode(query)))
|
||
|
||
|
||
def load_link_uris(doc: pymupdf.Document) -> list[str]:
|
||
"""Web-Linkziele der PDF (Link-Annotationen), ohne Dubletten."""
|
||
uris: dict[str, None] = {}
|
||
for page in doc:
|
||
for link in page.get_links():
|
||
uri = (link.get("uri") or "").strip()
|
||
if re.match(r"https?://", uri, re.IGNORECASE):
|
||
uris.setdefault(uri.replace(" ", "%20"))
|
||
return list(uris)
|
||
|
||
|
||
def _autolink(url: str) -> str:
|
||
if url.lower().startswith("www."):
|
||
return f"[{url}](https://{url})"
|
||
return f"<{url}>"
|
||
|
||
|
||
def _match_known(tokens: list[str], known: dict[str, str]) -> tuple[str, int] | None:
|
||
"""Längste Verbindung von Wörtern, die zu einem Linkziel der PDF passt."""
|
||
for count in range(min(MAX_URL_PARTS, len(tokens)), 0, -1):
|
||
parts = [t.rstrip() for t in tokens[:count]]
|
||
joined = "".join(parts)
|
||
consumed = sum(map(len, tokens[: count - 1])) + len(parts[-1])
|
||
for strip in range(len(joined)):
|
||
candidate = joined[: len(joined) - strip]
|
||
if uri := known.get(url_key(candidate)):
|
||
return uri, consumed - strip
|
||
if candidate[-1] not in URL_TRAILING:
|
||
break
|
||
return None
|
||
|
||
|
||
def _match_printed(tokens: list[str]) -> tuple[str, int]:
|
||
"""URL ohne (passendes) Linkziel: aus dem gedruckten Text zusammensetzen."""
|
||
parts = [tokens[0].rstrip()]
|
||
for token in tokens[1:MAX_URL_PARTS]:
|
||
if parts[-1][-1] in URL_BREAK_AFTER and URL_CONTINUATION.match(token.rstrip()):
|
||
parts.append(token.rstrip())
|
||
else:
|
||
break
|
||
joined = "".join(parts)
|
||
url = joined.rstrip(URL_TRAILING)
|
||
consumed = sum(map(len, tokens[: len(parts) - 1])) + len(parts[-1])
|
||
return url, consumed - (len(joined) - len(url))
|
||
|
||
|
||
def _unescape_map(text: str) -> tuple[str, list[int]]:
|
||
"""Text ohne Markdown-Maskierung und Position jedes Zeichens im Original."""
|
||
plain, index, i = [], [], 0
|
||
while i < len(text):
|
||
index.append(i) # Position samt Backslash, damit er beim Kopieren bleibt
|
||
if text[i] == "\\" and i + 1 < len(text) and text[i + 1] in MD_SPECIAL:
|
||
i += 1
|
||
plain.append(text[i])
|
||
i += 1
|
||
return "".join(plain), index + [len(text)]
|
||
|
||
|
||
def _link_line(line: str, known: dict[str, str]) -> str:
|
||
# URLs auf dem unmaskierten Text suchen ("Leitlinien\_" → "Leitlinien_"),
|
||
# den übrigen Text unverändert (maskiert) übernehmen.
|
||
plain, index = _unescape_map(line)
|
||
out = []
|
||
pos = 0
|
||
while match := URL_START.search(plain, pos):
|
||
out.append(line[index[pos]:index[match.start()]])
|
||
tokens = re.findall(r"\S+\s*", plain[match.start():])
|
||
url, consumed = _match_known(tokens, known) or _match_printed(tokens)
|
||
end = match.start() + consumed
|
||
out.append(_autolink(url) if "." in url else line[index[match.start()]:index[end]])
|
||
pos = end
|
||
out.append(line[index[pos]:])
|
||
return "".join(out)
|
||
|
||
|
||
def link_urls(markdown: str, uris: list[str] | None = None) -> str:
|
||
"""URLs im Fließtext anklickbar machen und mit PDF-Linkzielen reparieren."""
|
||
known: dict[str, str] = {}
|
||
for uri in uris or []:
|
||
for key in uri_keys(uri):
|
||
known.setdefault(key, clean_uri(uri))
|
||
lines = markdown.split("\n")
|
||
in_fence = False
|
||
for i, line in enumerate(lines):
|
||
if FENCE.match(line):
|
||
in_fence = not in_fence
|
||
elif not in_fence and not IMAGE.match(line):
|
||
lines[i] = _link_line(line, known)
|
||
return "\n".join(lines)
|
||
|
||
|
||
# Links mit eigenem Text ("hier", "Bericht"): Linktext und die Wörter davor aus
|
||
# der PDF lesen und die Stelle im Markdown als [Text](Ziel) markieren.
|
||
|
||
ANCHOR_CONTEXT_WORDS = 3
|
||
URL_LIKE_WORD = re.compile(r"://|^www\.|/|\w\.(?:[a-z]{2,4})\.?$", re.IGNORECASE)
|
||
|
||
|
||
def load_link_anchors(doc: pymupdf.Document) -> list[tuple[str, list[str], list[str]]]:
|
||
"""(Ziel, Wörter davor, Linkwörter) für Links, deren Text keine URL ist."""
|
||
anchors: list[tuple[str, list[str], list[str]]] = []
|
||
for page in doc:
|
||
words = page.get_text("words", sort=True)
|
||
links = [link for link in page.get_links() if (link.get("uri") or "").startswith("http")]
|
||
merged: list[tuple[str, list[int]]] = [] # (Ziel, Indizes der Wörter)
|
||
for link in links:
|
||
rect = link["from"]
|
||
hits = [
|
||
i for i, w in enumerate(words)
|
||
if rect.contains(pymupdf.Point((w[0] + w[2]) / 2, (w[1] + w[3]) / 2))
|
||
]
|
||
if not hits:
|
||
continue
|
||
if merged and merged[-1][0] == link["uri"] and hits[0] <= merged[-1][1][-1] + 1:
|
||
merged[-1][1].extend(i for i in hits if i not in merged[-1][1])
|
||
else:
|
||
merged.append((link["uri"], hits))
|
||
for uri, hits in merged:
|
||
text = [words[i][4] for i in sorted(hits)]
|
||
if any(URL_LIKE_WORD.search(word) for word in text):
|
||
continue # URL als Linktext: erledigt link_urls
|
||
first = min(hits)
|
||
before = [w[4] for w in words[max(0, first - ANCHOR_CONTEXT_WORDS):first]]
|
||
anchors.append((clean_uri(uri), before, text))
|
||
return anchors
|
||
|
||
|
||
def _md_literal(text: str) -> str:
|
||
"""Regex für Text, der im Markdown maskiert sein kann ("\\_", "\\*" …)."""
|
||
return "".join(
|
||
(r"\\?" if char in MD_SPECIAL else "") + re.escape(char) for char in text
|
||
)
|
||
|
||
|
||
def _words_pattern(words: list[str]) -> str:
|
||
"""Wörter mit beliebigem Leerraum, Hervorhebung oder Trennung dazwischen."""
|
||
parts = []
|
||
for word in words:
|
||
core = word.rstrip("-\u00ad") # Silbentrennung am Zeilenende
|
||
parts.append(_md_literal(core) + (r"-?" if core != word else ""))
|
||
return r"[\s*_]*".join(parts)
|
||
|
||
|
||
# Zwischen Kontextwörtern: Satzzeichen, Fußnotenziffern, Linkziele "](…)".
|
||
CONTEXT_GAP = r"(?:[\W_\d]|\]\([^)]*\))*"
|
||
|
||
|
||
def _context_pattern(words: list[str]) -> str:
|
||
cores = [core for word in words for core in re.findall(r"[^\W\d_]+", word)]
|
||
return CONTEXT_GAP.join(re.escape(core) for core in cores)
|
||
|
||
|
||
def link_anchors(markdown: str, anchors: list[tuple[str, list[str], list[str]]]) -> str:
|
||
"""Linktexte im Markdown als [Text](Ziel) setzen, in Lesereihenfolge."""
|
||
pos = 0
|
||
for uri, before, text in anchors:
|
||
anchor = r"(?P<anchor>" + _words_pattern(text) + r")"
|
||
patterns = []
|
||
if context := _context_pattern(before):
|
||
patterns.append(re.compile(context + CONTEXT_GAP + anchor))
|
||
if len(" ".join(text)) >= 6:
|
||
patterns.append(re.compile(r"(?<![\w\[])" + anchor)) # ohne Kontext
|
||
for pattern in patterns:
|
||
match = next(
|
||
(m for m in pattern.finditer(markdown, pos)
|
||
if not _inside_link(markdown, m.start("anchor"))),
|
||
None,
|
||
)
|
||
if match:
|
||
start, end = match.span("anchor")
|
||
markdown = f"{markdown[:start]}[{markdown[start:end]}]({uri}){markdown[end:]}"
|
||
pos = start
|
||
break
|
||
return markdown
|
||
|
||
|
||
def _inside_link(markdown: str, index: int) -> bool:
|
||
"""Liegt die Stelle schon in einem Link (<…>, [..](…))?"""
|
||
line_start = markdown.rfind("\n", 0, index) + 1
|
||
before = markdown[line_start:index]
|
||
|
||
def count(char: str) -> int: # maskierte Zeichen ("\\[") zählen nicht
|
||
return len(re.findall(r"(?<!\\)" + re.escape(char), before))
|
||
|
||
return count("<") > count(">") or count("[") > count("]") or (
|
||
before.rfind("](") > before.rfind(")")
|
||
)
|
||
|
||
|
||
# ===========================================================================
|
||
# Korrektur für pymupdf4llm: Trennstriche am Zeilenende, OCR-Ebenen
|
||
# ===========================================================================
|
||
# pymupdf4llm entfernt jeden Bindestrich am Zeilenende ("Militär-" + Zeile
|
||
# "Geheimdienst" -> "MilitärGeheimdienst"). Für Großbuchstaben und
|
||
# Konjunktionen ("Arbeits- und") markieren wir den Strich vorher, sodass er
|
||
# erhalten bleibt. Echte Silbentrennung (Kleinbuchstabe) bleibt unverändert.
|
||
# Weiche Trennstriche (U+00AD) und "¬" kennt pymupdf4llm nicht ("wach senden");
|
||
# sie trennen immer innerhalb eines Wortes und werden zu "-", das dann entfällt.
|
||
#
|
||
# Unsichtbarer Text (OCR-Ebene eines Scans) ist oft in Courier gesetzt;
|
||
# pymupdf4llm hielte ihn sonst für Programmcode und setzte jedes Wort in `…`.
|
||
|
||
KEEP_HYPHEN = "\ue000" # Strich behalten, ohne Leerzeichen anschließen
|
||
KEEP_HYPHEN_SPACE = "\ue001" # Strich und Leerzeichen behalten
|
||
|
||
|
||
SOFT_HYPHENS = ("\u00ad", "¬")
|
||
MONOSPACE_FLAG = 8
|
||
|
||
|
||
def _mark_line_end_hyphens(spans: list[dict]) -> list[dict]:
|
||
marked = []
|
||
for i, span in enumerate(spans):
|
||
nxt = spans[i + 1] if i + 1 < len(spans) else None
|
||
if not span.get("char_flags", VISIBLE_TEXT) & VISIBLE_TEXT:
|
||
span = {**span, "flags": span.get("flags", 0) & ~MONOSPACE_FLAG}
|
||
if not (span.get("flags", 0) & MONOSPACE_FLAG): # Code bleibt wörtlich
|
||
span = {**span, "text": md_escape(span["text"])}
|
||
text = span["text"].rstrip()
|
||
line_end = nxt is not None and (span.get("block"), span.get("line")) != (
|
||
nxt.get("block"), nxt.get("line")
|
||
)
|
||
if line_end and text.endswith(SOFT_HYPHENS):
|
||
span = {**span, "text": text[:-1] + "-"}
|
||
elif line_end and text.endswith("-"):
|
||
word = FIRST_WORD.match(nxt["text"].lstrip())
|
||
last = text.split()[-1]
|
||
if URL_START.search(last) or "/" in last:
|
||
span = {**span, "text": text + KEEP_HYPHEN} # URLs trennt niemand
|
||
elif word and word.group(0).lower() in CONJUNCTIONS:
|
||
span = {**span, "text": text + KEEP_HYPHEN_SPACE}
|
||
elif word and word.group(0)[0].isupper():
|
||
span = {**span, "text": text + KEEP_HYPHEN}
|
||
marked.append(span)
|
||
return marked
|
||
|
||
|
||
def patch_pymupdf4llm_hyphens() -> None:
|
||
original = getattr(document_layout, "get_styled_text", None)
|
||
if original is None or getattr(original, "pdf2md_patched", False):
|
||
return
|
||
|
||
def get_styled_text(spans):
|
||
output, suffix = original(_mark_line_end_hyphens(spans))
|
||
output = (
|
||
output.replace(KEEP_HYPHEN + " ", "")
|
||
.replace(KEEP_HYPHEN, "")
|
||
.replace(KEEP_HYPHEN_SPACE, "")
|
||
)
|
||
return output, suffix
|
||
|
||
get_styled_text.pdf2md_patched = True
|
||
document_layout.get_styled_text = get_styled_text
|
||
|
||
|
||
# Gedruckte Bücher trennen Absätze durch Einzug statt Leerraum. Auf Scanseiten
|
||
# fasst das Layout-Modell dann mehrere Absätze zu einem Block zusammen, und
|
||
# pymupdf4llm verbindet alle Zeilen. Wir trennen den Block vorher dort, wo
|
||
# eine Zeile mit Satzende schließt und die nächste eingerückt ist bzw. die
|
||
# vorige Zeile deutlich kürzer war.
|
||
|
||
INDENT_MIN_EM = 0.8 # Einzug ab ca. 1 Geviert
|
||
INDENT_MAX_EM = 8.0 # mehr ist eher zentrierter Text
|
||
SHORT_LINE_RATIO = 0.7 # Absatzende: Zeile kürzer als 70 % der längsten
|
||
|
||
|
||
def _line_text(line: dict) -> str:
|
||
return "".join(span["text"] for span in line["spans"]).strip()
|
||
|
||
|
||
def split_by_indent(textlines: list[dict], column_left: float | None = None) -> list[list[dict]]:
|
||
"""Zeilen eines Textblocks in Absätze aufteilen.
|
||
|
||
column_left: linker Rand der Textspalte, falls bekannt – sonst der Blockrand
|
||
(dann bleiben gleich eingerückte einzeilige Absätze unerkannt).
|
||
"""
|
||
if len(textlines) < 2:
|
||
return [textlines]
|
||
|
||
sizes = sorted(s["size"] for line in textlines for s in line["spans"])
|
||
em = sizes[len(sizes) // 2] if sizes else 10.0
|
||
lefts = sorted(line["bbox"][0] for line in textlines)
|
||
left = lefts[len(lefts) // 4] # Blockrand, robust gegen Einzüge
|
||
if column_left is not None:
|
||
left = min(left, column_left)
|
||
widths = [line["bbox"][2] - line["bbox"][0] for line in textlines]
|
||
longest = max(widths)
|
||
|
||
groups = [[textlines[0]]]
|
||
for prev, line, prev_width in zip(textlines, textlines[1:], widths):
|
||
indent = line["bbox"][0] - left
|
||
indented = INDENT_MIN_EM * em <= indent <= INDENT_MAX_EM * em
|
||
short = prev_width < SHORT_LINE_RATIO * longest
|
||
if (indented or short) and ends_sentence(_line_text(prev)):
|
||
groups.append([line])
|
||
else:
|
||
groups[-1].append(line)
|
||
return groups
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Zeilengetreue Abschnitte: Code, Register, Verse, Adressen …
|
||
# ---------------------------------------------------------------------------
|
||
# Ein Zeilenumbruch stammt vom Satzprogramm (weich), wenn das erste Wort der
|
||
# nächsten Zeile nicht mehr in die Zeile gepasst hätte. Sonst hat ihn der Autor
|
||
# gesetzt (hart); dann bleibt die Zeile samt Einrückung erhalten. Überwiegen in
|
||
# einem Block die harten Umbrüche, wird er zeilengetreu ausgegeben: als
|
||
# Codeblock (```) oder als Zeilen mit hartem Umbruch ("\" am Zeilenende).
|
||
|
||
HARD_BREAK = "\ue002" # Zeile endet mit hartem Umbruch (wird zu "\")
|
||
BLOCK_END = "\ue004" # Block endet ohne Satzende: "\ue004<freie Zeichen>[P]\ue004"
|
||
BLOCK_END_MARK = re.compile("\ue004(\\d+)(P?)\ue004$") # P: mehrzeiliger Absatz
|
||
PARAGRAPH_END = "\ue005" # Absatz endet hier, auch ohne Satzende
|
||
MARKERS = re.compile("\ue004\\d*P?\ue004|[\ue002-\ue005]") # alle internen Markierungen
|
||
STANZA_BREAK = "\ue003" # Leerzeile innerhalb eines zeilengetreuen Blocks
|
||
NBSP = "\u00a0"
|
||
TEXT_BOXES = ("text", "list-item", "footnote")
|
||
# Spielraum beim "Hätte das Wort noch gepasst?": Satzprogramme brechen nicht
|
||
# exakt um (Band 1, Flattersatz: bis 3 % der Spalte), Register und Verse
|
||
# lassen fast immer weit mehr frei.
|
||
FIT_TOLERANCE = 0.05 # Anteil der Spaltenbreite
|
||
FIT_TOLERANCE_EM = 0.3 # mindestens
|
||
HARD_BLOCK_RATIO = 0.5 # ab diesem Anteil harter Umbrüche: zeilengetreu
|
||
STANZA_GAP = 1.6 # Zeilenabstand (× üblichem) für Leerzeile
|
||
INDENT_STEP_EM = 0.8 # Einrückungsstufen: Abstand mindestens so groß
|
||
CODE_INDENT = " " # Leerzeichen pro Einrückungsstufe im Code
|
||
TEXT_INDENT = NBSP * 4 # geschützte Leerzeichen pro Stufe sonst
|
||
|
||
CODE_HINT = re.compile(
|
||
r"[{};]\s*$" # Zeile endet mit { } ;
|
||
r"|^\s*[}\])]" # beginnt mit } ] )
|
||
r"|::|=>|->|==|!=|\+=|&&|\|\|" # Operatoren
|
||
r"|^\s*(?:def|class|fn|let|mut|use|import|from|return|pub|impl|struct|"
|
||
r"enum|match|const|var|function|elif|lambda|async|await|#include)\b"
|
||
r"|^\s*(?:if|for|while|else|try|except|with)\b.*:\s*$" # Python-Block
|
||
r"|^\s*(?:#\[|@\w+|//|/\*|\$ )" # Attribut, Dekorator, Kommentar
|
||
r"|\w\([^()]*\)\s*$" # endet mit Aufruf f(x)
|
||
)
|
||
|
||
# Zeilen rund um Code: Konfiguration (zählt erst ab zwei Zeilen als Code) …
|
||
CONFIG_LINE = re.compile(
|
||
r"^\s*\[[\w.\-]+\]\s*$" # [dependencies]
|
||
r"|^\s*[\w.\-]+\s*=\s*\S" # name = "wert"
|
||
)
|
||
# … sowie Befehle und Programmausgaben (auch als einzelne Zeile Code).
|
||
COMMAND_LINE = re.compile(
|
||
r"^\s*(?:\$|>>>|C:\\>)\s" # Eingabeaufforderung
|
||
r"|^\s*(?:cargo|rustup|rustc|pip3?|python3?|pytest|npm|npx|git|sudo|"
|
||
r"apt(?:-get)?|docker|conda|javac)\s+\S" # Befehl
|
||
r"|^\s*(?:Compiling|Finished|Traceback|error(?:\[\w+\])?:|warning:)\s"
|
||
r"|^\s*running \d+ tests?\b|^\s*test result:|\.\.\. (?:ok|FAILED|ignored)\s*$"
|
||
)
|
||
|
||
_page_lines: list[tuple[float, float]] = [] # (x0, x1) der Textzeilen der Seite
|
||
|
||
|
||
def _remember_page_lines(boxes) -> None:
|
||
_page_lines.clear()
|
||
for box in boxes:
|
||
if getattr(box, "boxclass", None) in TEXT_BOXES:
|
||
for line in getattr(box, "textlines", None) or []:
|
||
_page_lines.append((line["bbox"][0], line["bbox"][2]))
|
||
|
||
|
||
def _visual_lines(textlines: list[dict]) -> list[dict]:
|
||
"""Textzeilen auf gleicher Höhe (z. B. Wechsel der Schrift) zusammenfassen."""
|
||
merged: list[dict] = []
|
||
for line in textlines:
|
||
x0, y0, x1, y1 = line["bbox"]
|
||
if merged:
|
||
last = merged[-1]
|
||
ly0, ly1 = last["bbox"][1], last["bbox"][3]
|
||
if (y0 + y1) / 2 < ly1 and (y0 + y1) / 2 > ly0 and x0 >= last["bbox"][0]:
|
||
bx0, by0, bx1, by1 = last["bbox"]
|
||
last["bbox"] = (bx0, min(by0, y0), max(bx1, x1), max(by1, y1))
|
||
last["spans"] = last["spans"] + line["spans"]
|
||
continue
|
||
merged.append({"bbox": tuple(line["bbox"]), "spans": list(line["spans"])})
|
||
return merged
|
||
|
||
|
||
def _raw_text(line: dict) -> str:
|
||
"""Text einer Zeile; Leerzeichen dort ergänzen, wo die PDF Wörter nur durch
|
||
Abstand trennt (sonst "Openthefile." in Codeblöcken)."""
|
||
text, prev = "", None
|
||
for span in line["spans"]:
|
||
if prev is not None and text and not text[-1].isspace() and not span["text"][:1].isspace():
|
||
if span["bbox"][0] - prev["bbox"][2] > 0.15 * span.get("size", 10):
|
||
text += " "
|
||
text += span["text"]
|
||
prev = span
|
||
return text
|
||
|
||
|
||
def _char_width(line: dict, em: float) -> float:
|
||
chars = sum(len(s["text"]) for s in line["spans"])
|
||
width = sum(s["bbox"][2] - s["bbox"][0] for s in line["spans"])
|
||
return width / chars if chars and width > 0 else 0.5 * em
|
||
|
||
|
||
def _column_right(lines: list[dict], em: float) -> float:
|
||
"""Rechter Rand der Textspalte, in der der Block steht."""
|
||
bx0 = min(line["bbox"][0] for line in lines)
|
||
bx1 = max(line["bbox"][2] for line in lines)
|
||
edges = [
|
||
x1 for x0, x1 in _page_lines
|
||
if x0 < bx1 and x1 > bx0 and x0 >= bx0 - 2 * em
|
||
]
|
||
return max(edges + [bx1])
|
||
|
||
|
||
def _fit_tolerance(lines: list[dict], right: float, em: float) -> float:
|
||
left = min(line["bbox"][0] for line in lines)
|
||
return max(FIT_TOLERANCE_EM * em, FIT_TOLERANCE * (right - left))
|
||
|
||
|
||
def _column_left(lines: list[dict], em: float) -> float:
|
||
"""Linker Rand der Textspalte (Einzüge einzeiliger Absätze erkennen)."""
|
||
bx0 = min(line["bbox"][0] for line in lines)
|
||
bx1 = max(line["bbox"][2] for line in lines)
|
||
edges = [x0 for x0, x1 in _page_lines if x0 < bx1 and x1 > bx0 and x0 >= bx0 - 10 * em]
|
||
return min(edges + [bx0])
|
||
|
||
|
||
def _is_soft_break(prev: dict, line: dict, right: float, em: float, tol: float) -> bool:
|
||
"""Hätte das erste Wort der Zeile nicht mehr in die vorige gepasst?"""
|
||
prev_text = _raw_text(prev).rstrip()
|
||
text = _raw_text(line).strip()
|
||
if not text:
|
||
return True
|
||
if prev_text.endswith(("-",) + SOFT_HYPHENS) and text[0].islower():
|
||
return True # Silbentrennung setzt nur das Satzprogramm
|
||
word = text.split()[0]
|
||
needed = (len(word) + 1) * _char_width(line, em)
|
||
return prev["bbox"][2] + needed > right - tol
|
||
|
||
|
||
def _indent_levels(lines: list[dict], em: float) -> list[int]:
|
||
"""Einrückung jeder Zeile als Stufe 0, 1, 2 … (geometrisch)."""
|
||
lefts = sorted({round(line["bbox"][0], 1) for line in lines})
|
||
stops: list[float] = []
|
||
for x in lefts:
|
||
if not stops or x - stops[-1] >= INDENT_STEP_EM * em:
|
||
stops.append(x)
|
||
levels = []
|
||
for line in lines:
|
||
x = line["bbox"][0]
|
||
levels.append(max(i for i, stop in enumerate(stops) if stop <= x + 0.5))
|
||
return levels
|
||
|
||
|
||
def _is_code(lines: list[dict], ignore_code: bool, short: bool = False) -> bool:
|
||
"""Ganzer Block Code? short: kurzer Block – dann zählen auch Konfiguration,
|
||
Befehle und Programmausgaben (sonst nur zeilenweise, siehe _mixed_block)."""
|
||
if ignore_code:
|
||
return False
|
||
texts = [_raw_text(line) for line in lines]
|
||
if all(
|
||
s["flags"] & 8 and s.get("char_flags", VISIBLE_TEXT) & VISIBLE_TEXT
|
||
for line in lines for s in line["spans"] if s["text"].strip()
|
||
):
|
||
return True # Schreibmaschinenschrift
|
||
hints = sum(
|
||
1 for t in texts
|
||
if not REGISTER_ENTRY.search(t) and CODE_HINT.search(t.replace(NBSP, " "))
|
||
or (short and (_code_like(t) or COMMAND_LINE.search(t)))
|
||
)
|
||
indented = sum(1 for t in texts if t[:1] in (" ", NBSP, "\t"))
|
||
return hints >= max(1, (len(texts) + 1) // 2) or (hints and indented)
|
||
|
||
|
||
# Registereintrag ("match(), 247", "Closure, 128, 130–132"): nie Code.
|
||
REGISTER_ENTRY = re.compile(r",\s*\d+(?:\s*[-–]\s*\d+)?(?:\s*,\s*\d+(?:\s*[-–]\s*\d+)?)*\s*$")
|
||
|
||
|
||
def _code_like(text: str) -> bool:
|
||
text = text.replace(NBSP, " ")
|
||
if REGISTER_ENTRY.search(text):
|
||
return False
|
||
return bool(CODE_HINT.search(text) or CONFIG_LINE.search(text))
|
||
|
||
|
||
def _mixed_block(
|
||
logical: list[list[dict]], levels: list[int], gaps: list[bool], end: str
|
||
) -> str | None:
|
||
"""Zeilengetreuer Block aus Text und Code (Anleitung, Konfiguration, Befehl,
|
||
Ausgabe): Code-Abschnitte als ```-Block, den Rest als Zeilen."""
|
||
texts = [" ".join(_raw_text(line).strip() for line in parts) for parts in logical]
|
||
strong = [bool(COMMAND_LINE.search(t)) and not REGISTER_ENTRY.search(t) for t in texts]
|
||
code = [s or _code_like(t) for s, t in zip(strong, texts)]
|
||
for i in range(1, len(code) - 1): # Zeile zwischen zwei Codezeilen gehört dazu,
|
||
prose = texts[i].endswith(":") or ends_sentence(texts[i]) # außer Anleitung
|
||
code[i] = code[i] or (code[i - 1] and code[i + 1] and not prose)
|
||
|
||
segments: list[list] = [] # [start, ende, ist_code]
|
||
for i, is_code in enumerate(code):
|
||
if segments and segments[-1][2] == is_code:
|
||
segments[-1][1] = i + 1
|
||
else:
|
||
segments.append([i, i + 1, is_code])
|
||
for segment in segments: # einzelne Treffer ("@classmethod, 190") sind kein Code
|
||
a, b, is_code = segment
|
||
if is_code and b - a < 2 and not any(strong[a:b]):
|
||
segment[2] = False
|
||
if not any(is_code for _, _, is_code in segments):
|
||
return None
|
||
|
||
out = []
|
||
for n, (a, b, is_code) in enumerate(segments):
|
||
last = n == len(segments) - 1
|
||
if is_code:
|
||
base = min(levels[a:b])
|
||
out.append(_code_block(logical[a:b], [lvl - base for lvl in levels[a:b]]))
|
||
else:
|
||
out.append(_verbatim_block(logical[a:b], levels[a:b], gaps[a:b - 1], end if last else ""))
|
||
return "".join(out)
|
||
|
||
|
||
def _code_block(logical: list[list[dict]], levels: list[int]) -> str:
|
||
rows = []
|
||
for parts, level in zip(logical, levels):
|
||
text = " ".join(_raw_text(line).replace(NBSP, " ").rstrip() for line in parts)
|
||
rows.append(CODE_INDENT * level + text)
|
||
fence = "~~~" if any("```" in row for row in rows) else "```"
|
||
return fence + "\n" + "\n".join(rows) + "\n" + fence + "\n\n"
|
||
|
||
|
||
def _verbatim_block(
|
||
logical: list[list[dict]], levels: list[int], gaps: list[bool], end: str
|
||
) -> str:
|
||
rows = []
|
||
for i, (parts, level) in enumerate(zip(logical, levels)):
|
||
spans = [span for line in parts for span in line["spans"]]
|
||
text, _ = document_layout.get_styled_text(spans)
|
||
lead = len(_raw_text(parts[0])) - len(_raw_text(parts[0]).lstrip())
|
||
row = TEXT_INDENT * level + NBSP * lead + text.strip()
|
||
last = i == len(logical) - 1
|
||
row += (end or PARAGRAPH_END) if last else HARD_BREAK
|
||
rows.append(row)
|
||
if not last and gaps[i]:
|
||
rows.append(STANZA_BREAK)
|
||
return "\n".join(rows) + "\n\n"
|
||
|
||
|
||
def _block_end(lines: list[dict], right: float, em: float, tol: float) -> str:
|
||
"""Markierung für ein Blockende ohne Satzende: freie Zeichen in der Zeile.
|
||
|
||
Ob der Satz im nächsten Block weitergeht, entscheidet normalize_markdown
|
||
mit dem ersten Wort dieses Blocks ("Hätte es noch gepasst?").
|
||
"""
|
||
last = lines[-1]
|
||
text = _raw_text(last).strip()
|
||
if not text or ends_sentence(text) or text.endswith(("-",) + SOFT_HYPHENS):
|
||
return ""
|
||
free = (right - tol - last["bbox"][2]) / _char_width(last, em)
|
||
paragraph = "P" if len(lines) > 1 else ""
|
||
return f"{BLOCK_END}{max(0, int(free))}{paragraph}{BLOCK_END}"
|
||
|
||
|
||
def lines_to_md(textlines: list[dict], original, ignore_code: bool = False) -> str:
|
||
"""Textblock ausgeben: Fließtext wie bisher, sonst zeilengetreu."""
|
||
lines = _visual_lines(textlines)
|
||
sizes = sorted(s["size"] for line in lines for s in line["spans"])
|
||
em = sizes[len(sizes) // 2] if sizes else 10.0
|
||
right = _column_right(lines, em)
|
||
tol = _fit_tolerance(lines, right, em)
|
||
end = _block_end(lines, right, em, tol)
|
||
|
||
# Harte Umbrüche zählen; ein Satzende mit eingerückter Folgezeile ist dagegen
|
||
# ein Absatzende (Roman), ohne Einrückung eine Verszeile (Gedicht).
|
||
soft = [_is_soft_break(a, b, right, em, tol) for a, b in zip(lines, lines[1:])]
|
||
left = _column_left(lines, em)
|
||
|
||
def new_paragraph(prev: dict, line: dict) -> bool:
|
||
indent = line["bbox"][0] - left
|
||
return ends_sentence(_raw_text(prev).strip()) and (
|
||
INDENT_MIN_EM * em <= indent <= INDENT_MAX_EM * em
|
||
)
|
||
|
||
hard = sum(
|
||
1 for is_soft, a, b in zip(soft, lines, lines[1:])
|
||
if not is_soft and not new_paragraph(a, b)
|
||
)
|
||
|
||
verbatim = bool(soft) and hard / len(soft) >= HARD_BLOCK_RATIO
|
||
if not verbatim:
|
||
if end and len(lines) <= 3 and _is_code(lines, ignore_code, short=True):
|
||
return _code_block([lines], [0]) # kurze (umbrochene) Codezeile
|
||
# Fließtext: Absätze am Einzug trennen, wie bisher ausgeben.
|
||
groups = split_by_indent(textlines, left)
|
||
md = "".join(original(group, ignore_code) for group in groups)
|
||
return md.rstrip("\n").rstrip() + end + "\n\n" if end else md
|
||
|
||
# Zeilengetreu: weich umbrochene Zeilen zu logischen Zeilen verbinden.
|
||
logical: list[list[dict]] = [[lines[0]]]
|
||
for is_soft, line in zip(soft, lines[1:]):
|
||
if is_soft:
|
||
logical[-1].append(line)
|
||
else:
|
||
logical.append([line])
|
||
|
||
firsts = [parts[0] for parts in logical]
|
||
levels = _indent_levels(firsts, em)
|
||
if _is_code(lines, ignore_code):
|
||
return _code_block(logical, levels)
|
||
|
||
pitches = [b["bbox"][1] - a["bbox"][1] for a, b in zip(firsts, firsts[1:])]
|
||
usual = sorted(pitches)[len(pitches) // 2] if pitches else 0
|
||
gaps = [usual > 0 and pitch > STANZA_GAP * usual for pitch in pitches]
|
||
if not ignore_code and (mixed := _mixed_block(logical, levels, gaps, end)):
|
||
return mixed
|
||
return _verbatim_block(logical, levels, gaps, end)
|
||
|
||
|
||
def patch_pymupdf4llm_paragraphs() -> None:
|
||
original = getattr(document_layout, "text_to_md", None)
|
||
if original is None or getattr(original, "pdf2md_patched", False):
|
||
return
|
||
|
||
def text_to_md(textlines, ignore_code: bool = False):
|
||
if not textlines or document_layout.is_superscripted(textlines[0]):
|
||
return original(textlines, ignore_code)
|
||
try:
|
||
return lines_to_md(textlines, original, ignore_code)
|
||
except (KeyError, IndexError, TypeError, ValueError, ZeroDivisionError):
|
||
return original(textlines, ignore_code) # unerwartete Struktur
|
||
|
||
text_to_md.pdf2md_patched = True
|
||
document_layout.text_to_md = text_to_md
|
||
|
||
levels = getattr(document_layout, "create_list_item_levels", None)
|
||
if levels is not None and not getattr(levels, "pdf2md_patched", False):
|
||
|
||
def create_list_item_levels(layout_info):
|
||
_remember_page_lines(layout_info) # einmal pro Seite aufgerufen
|
||
return levels(layout_info)
|
||
|
||
create_list_item_levels.pdf2md_patched = True
|
||
document_layout.create_list_item_levels = create_list_item_levels
|
||
|
||
|
||
class _DiscardMessages(io.StringIO):
|
||
"""Ersatz für pymupdf4llms Meldungspuffer ("OCR on page …")."""
|
||
|
||
def write(self, text: str) -> int:
|
||
return len(text)
|
||
|
||
|
||
# JPEG-Qualität: pymupdf4llm speichert Bilder mit Pixmap.save() ohne Angabe
|
||
# der Qualität (Standard 95); wir setzen sie über --jpeg-quality.
|
||
DEFAULT_JPEG_QUALITY = 85
|
||
_jpeg_quality = [DEFAULT_JPEG_QUALITY]
|
||
_pixmap_save = pymupdf.Pixmap.save
|
||
|
||
|
||
def _save_pixmap(self, filename, output=None, jpg_quality=None):
|
||
quality = _jpeg_quality[0] if jpg_quality is None else jpg_quality
|
||
return _pixmap_save(self, filename, output, quality)
|
||
|
||
|
||
pymupdf.Pixmap.save = _save_pixmap
|
||
|
||
patch_pymupdf4llm_hyphens()
|
||
patch_pymupdf4llm_paragraphs()
|
||
document_layout.INFO_MESSAGES = _DiscardMessages()
|
||
|
||
|
||
# ===========================================================================
|
||
# Seiten klassifizieren und OCR planen
|
||
# ===========================================================================
|
||
|
||
PAGE_TEXT = "text" # digitale Textseite
|
||
PAGE_OCR_LAYER = "ocr-layer" # Scan mit vorhandener (unsichtbarer) OCR-Ebene
|
||
PAGE_SCAN = "scan" # Bildseite ohne Text
|
||
PAGE_EMPTY = "empty"
|
||
|
||
MIN_TEXT_CHARS = 50
|
||
MAX_BAD_CHAR_RATIO = 0.1
|
||
MIN_IMAGE_COVER = 0.3
|
||
MIN_OCR_LAYER_CHARS = 20
|
||
|
||
PLAUSIBLE_MIN_CHARS = 40
|
||
PLAUSIBLE_MIN_WORDS = 20 # Cover/Titelbilder haben weniger Wörter
|
||
PLAUSIBLE_WORD_RATIO = 0.7
|
||
PLAUSIBLE_FUNCTION_RATIO = 0.1
|
||
WORDLIKE = re.compile(r"^\W*([^\W\d_]{2,}(?:[-'’][^\W\d_]+)*)\W*$")
|
||
|
||
# Häufige Funktionswörter: In echtem Fließtext machen sie gut 10 % aus,
|
||
# im "Text", den die OCR aus Fotos oder Rauschen liest, praktisch nie.
|
||
FUNCTION_WORDS = {
|
||
"deu": set(
|
||
"der die das den dem des ein eine einen einem einer und oder aber in im "
|
||
"an am auf aus bei mit von zu zum zur für über unter vor nach ist war "
|
||
"sind waren hat hatte wird wurde nicht er sie es ich wir ihr du man sich "
|
||
"so wie als auch noch nur schon dass daß wenn da dann doch ja".split()
|
||
),
|
||
"eng": set(
|
||
"the a an and or but in on at of to for with is was are were be been he "
|
||
"she it i we you they not that this as by from his her had have".split()
|
||
),
|
||
"fra": set(
|
||
"le la les un une et ou de du des en dans est sont pas que qui il elle "
|
||
"au aux ce se sur pour par avec".split()
|
||
),
|
||
"spa": set(
|
||
"el la los las un una y o de del en es son no que se al por con para "
|
||
"su lo como".split()
|
||
),
|
||
}
|
||
|
||
SKEW_TOLERANCE = 0.002 # sin(Winkel), ca. 0,1°
|
||
MAX_SKEWED_LINES = 0.2 # mehr schiefe Zeilen -> OCR-Ebene neu erkennen
|
||
|
||
VISIBLE_TEXT = pymupdf.mupdf.FZ_STEXT_FILLED | pymupdf.mupdf.FZ_STEXT_STROKED
|
||
|
||
# Verarbeitung pro Seite
|
||
RUN_PLAIN = "plain" # vorhandenen Text verwenden, keine OCR
|
||
RUN_OCR = "ocr" # Tesseract-OCR
|
||
RUN_REUSE = "reuse" # vorhandene OCR-Ebene übernehmen
|
||
|
||
CHUNK_SIZE = {RUN_PLAIN: 50, RUN_OCR: 10, RUN_REUSE: 25}
|
||
|
||
|
||
def classify_page(page: pymupdf.Page) -> str:
|
||
visible = invisible = bad = 0
|
||
for block in page.get_text("dict")["blocks"]:
|
||
for line in block.get("lines", []):
|
||
for span in line["spans"]:
|
||
text = span["text"].strip()
|
||
if not text:
|
||
continue
|
||
if span["char_flags"] & VISIBLE_TEXT and span.get("alpha", 255):
|
||
visible += len(text)
|
||
bad += text.count("\ufffd")
|
||
else:
|
||
invisible += len(text)
|
||
|
||
image_rect = pymupdf.EMPTY_RECT()
|
||
for kind, bbox in page.get_bboxlog():
|
||
if kind in ("fill-image", "fill-imgmask"):
|
||
image_rect |= pymupdf.Rect(bbox) & page.rect
|
||
cover = abs(image_rect) / abs(page.rect) if abs(page.rect) else 0.0
|
||
|
||
if visible >= MIN_TEXT_CHARS:
|
||
return PAGE_SCAN if bad / visible > MAX_BAD_CHAR_RATIO else PAGE_TEXT
|
||
if cover >= MIN_IMAGE_COVER:
|
||
return PAGE_OCR_LAYER if invisible >= MIN_OCR_LAYER_CHARS else PAGE_SCAN
|
||
return PAGE_TEXT if visible else PAGE_EMPTY
|
||
|
||
|
||
def is_wordlike(token: str) -> bool:
|
||
"""Buchstabenwort mit üblicher Schreibweise (nicht "erNee", "eeBR")."""
|
||
match = WORDLIKE.match(token)
|
||
if not match:
|
||
return False
|
||
core = re.sub(r"[-'’]", "", match.group(1))
|
||
return core.islower() or core.isupper() or core.istitle()
|
||
|
||
|
||
def skewed_line_ratio(page: pymupdf.Page) -> float:
|
||
"""Anteil schräg liegender Textzeilen (z. B. OCR-Ebene eines schiefen Scans)."""
|
||
total = skewed = 0
|
||
for block in page.get_text("dict")["blocks"]:
|
||
for line in block.get("lines", []):
|
||
total += 1
|
||
skewed += abs(line["dir"][1]) > SKEW_TOLERANCE
|
||
return skewed / total if total else 0.0
|
||
|
||
|
||
def looks_like_text(text: str, language: str = "deu") -> bool:
|
||
"""Plausibilitätsprüfung: Ergibt die OCR echten Text oder Kauderwelsch?"""
|
||
words = text.split()
|
||
if len(text.strip()) < PLAUSIBLE_MIN_CHARS or len(words) < PLAUSIBLE_MIN_WORDS:
|
||
return False
|
||
if sum(map(is_wordlike, words)) / len(words) < PLAUSIBLE_WORD_RATIO:
|
||
return False
|
||
|
||
function_words = set().union(
|
||
*(FUNCTION_WORDS.get(lang, set()) for lang in language.split("+"))
|
||
)
|
||
if not function_words:
|
||
return True # Sprache ohne Wortliste: nur die Wortform zählt
|
||
lowered = [w.strip(".,;:!?»«„“”\"'‚‘’()").lower() for w in words]
|
||
hits = sum(1 for w in lowered if w in function_words)
|
||
return hits / len(words) >= PLAUSIBLE_FUNCTION_RATIO
|
||
|
||
|
||
def check_ocr_language(language: str) -> None:
|
||
try:
|
||
tessdata = pymupdf.get_tessdata()
|
||
except Exception:
|
||
tessdata = None
|
||
if not tessdata:
|
||
raise RuntimeError(
|
||
"Tesseract nicht gefunden – installieren mit: "
|
||
"sudo apt install tesseract-ocr tesseract-ocr-deu"
|
||
)
|
||
missing = [
|
||
lang for lang in language.split("+")
|
||
if not (Path(tessdata) / f"{lang}.traineddata").is_file()
|
||
]
|
||
if missing:
|
||
packages = " ".join(f"tesseract-ocr-{lang}" for lang in missing)
|
||
raise RuntimeError(
|
||
f"Tesseract-Sprachdaten fehlen: {', '.join(missing)} – "
|
||
f"installieren mit: sudo apt install {packages}"
|
||
)
|
||
|
||
|
||
def probe_ocr(page: pymupdf.Page, language: str, dpi: int) -> bool:
|
||
textpage = page.get_textpage_ocr(language=language, dpi=dpi, full=True)
|
||
return looks_like_text(page.get_text(textpage=textpage), language)
|
||
|
||
|
||
def plan_pages(
|
||
doc: pymupdf.Document, mode: str, language: str, dpi: int
|
||
) -> tuple[list[str], list[str]]:
|
||
"""Seitentypen ermitteln und pro Seite die Verarbeitung festlegen."""
|
||
kinds = [classify_page(page) for page in doc]
|
||
counts = Counter(kinds)
|
||
|
||
if mode == "never":
|
||
return kinds, [RUN_REUSE if k == PAGE_OCR_LAYER else RUN_PLAIN for k in kinds]
|
||
|
||
if mode == "always":
|
||
check_ocr_language(language)
|
||
return kinds, [RUN_PLAIN if k == PAGE_EMPTY else RUN_OCR for k in kinds]
|
||
|
||
# auto
|
||
if counts[PAGE_SCAN] or counts[PAGE_OCR_LAYER]:
|
||
check_ocr_language(language)
|
||
scanned_doc = counts[PAGE_SCAN] + counts[PAGE_OCR_LAYER] > counts[PAGE_TEXT]
|
||
|
||
plan = []
|
||
for page, kind in zip(doc, kinds):
|
||
if kind == PAGE_OCR_LAYER:
|
||
# Gute OCR-Ebenen übernehmen; schiefe (Scan nicht entzerrt)
|
||
# liefern eine falsche Lesereihenfolge -> neu erkennen.
|
||
skewed = skewed_line_ratio(page) > MAX_SKEWED_LINES
|
||
plan.append(RUN_OCR if skewed else RUN_REUSE)
|
||
elif kind == PAGE_SCAN and (scanned_doc or probe_ocr(page, language, dpi)):
|
||
# In digitalen PDFs sind reine Bildseiten meist Fotos/Cover;
|
||
# die werden nur übernommen, wenn die Probe-OCR echten Text liefert.
|
||
plan.append(RUN_OCR)
|
||
else:
|
||
plan.append(RUN_PLAIN)
|
||
return kinds, plan
|
||
|
||
|
||
def make_tasks(plan: list[str]) -> list[tuple[str, list[int]]]:
|
||
"""Aufeinanderfolgende Seiten gleicher Verarbeitung zu Paketen bündeln."""
|
||
tasks: list[tuple[str, list[int]]] = []
|
||
for pno, run in enumerate(plan):
|
||
if (
|
||
tasks
|
||
and tasks[-1][0] == run
|
||
and tasks[-1][1][-1] == pno - 1
|
||
and len(tasks[-1][1]) < CHUNK_SIZE[run]
|
||
):
|
||
tasks[-1][1].append(pno)
|
||
else:
|
||
tasks.append((run, [pno]))
|
||
return tasks
|
||
|
||
|
||
# ===========================================================================
|
||
# Konvertierung mit pymupdf4llm
|
||
# ===========================================================================
|
||
|
||
def keep_existing_ocr(page, **kwargs) -> None:
|
||
"""OCR-Funktion, die nichts tut: Die vorhandene OCR-Ebene wird genutzt.
|
||
|
||
pymupdf4llm verwendet unsichtbaren Text nur auf Seiten, die als
|
||
OCR-Seiten gelten; mit dieser Funktion entfällt eine zweite OCR.
|
||
"""
|
||
|
||
|
||
def convert_pages(source: str, run: str, pages: list[int], options: dict) -> str:
|
||
_jpeg_quality[0] = options.get("jpeg_quality", DEFAULT_JPEG_QUALITY) # auch im Worker
|
||
kwargs = dict(
|
||
pages=pages,
|
||
header=options["keep_headers"],
|
||
footer=options["keep_footers"],
|
||
write_images=options["images"],
|
||
image_path=options["image_path"],
|
||
image_format=options["image_format"],
|
||
)
|
||
if run == RUN_OCR:
|
||
kwargs.update(
|
||
force_ocr=True,
|
||
ocr_function=tesseract_api.exec_ocr,
|
||
ocr_language=options["ocr_language"],
|
||
ocr_dpi=options["ocr_dpi"],
|
||
)
|
||
elif run == RUN_REUSE:
|
||
kwargs.update(force_ocr=True, ocr_function=keep_existing_ocr)
|
||
else:
|
||
kwargs.update(use_ocr=False)
|
||
|
||
# pymupdf4llm schreibt Statusmeldungen auf stdout; wir melden selbst.
|
||
with contextlib.redirect_stdout(io.StringIO()):
|
||
return pymupdf4llm.to_markdown(source, **kwargs)
|
||
|
||
|
||
def _init_worker() -> None:
|
||
# Ein Thread pro Prozess, falls Tesseract mit OpenMP gebaut ist;
|
||
# die Parallelität kommt von den Prozessen.
|
||
os.environ["OMP_THREAD_LIMIT"] = "1"
|
||
|
||
|
||
def extract_markdown(source: Path, plan: list[str], options: dict, jobs: int) -> str:
|
||
tasks = make_tasks(plan)
|
||
total = len(tasks)
|
||
labels = {RUN_PLAIN: "Text", RUN_OCR: "OCR", RUN_REUSE: "OCR-Ebene"}
|
||
|
||
def report(i: int, run: str, pages: list[int]) -> None:
|
||
print(
|
||
f" [{i}/{total}] Seiten {pages[0] + 1}–{pages[-1] + 1} "
|
||
f"({labels[run]}) fertig",
|
||
flush=True,
|
||
)
|
||
|
||
parts = []
|
||
# Nur OCR profitiert von mehreren Prozessen; das Layout-Modell nutzt
|
||
# ohnehin alle Kerne.
|
||
if jobs <= 1 or total == 1 or RUN_OCR not in plan:
|
||
for i, (run, pages) in enumerate(tasks, 1):
|
||
parts.append(convert_pages(str(source), run, pages, options))
|
||
report(i, run, pages)
|
||
else:
|
||
context = multiprocessing.get_context("spawn")
|
||
with ProcessPoolExecutor(
|
||
max_workers=min(jobs, total),
|
||
mp_context=context,
|
||
initializer=_init_worker,
|
||
) as pool:
|
||
futures = [
|
||
pool.submit(convert_pages, str(source), run, pages, options)
|
||
for run, pages in tasks
|
||
]
|
||
for i, (future, (run, pages)) in enumerate(zip(futures, tasks), 1):
|
||
parts.append(future.result())
|
||
report(i, run, pages)
|
||
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
IMAGE_LINK = re.compile(r"(!\[[^\]]*\]\()([^)\s]+)\)")
|
||
|
||
|
||
def make_image_links_relative(markdown: str, image_dir: Path) -> str:
|
||
"""Bildpfade relativ zur Markdown-Datei machen ("<name>_images/datei.jpg").
|
||
|
||
pymupdf4llm schreibt den Pfad relativ zum Arbeitsverzeichnis, wenn der
|
||
Bilderordner darunter liegt (z. B. "Output/x_images/…"), sonst absolut.
|
||
"""
|
||
image_dir = image_dir.resolve()
|
||
|
||
def fix(match: re.Match) -> str:
|
||
path = Path(match.group(2))
|
||
full = path if path.is_absolute() else Path.cwd() / path
|
||
if full.resolve().parent != image_dir:
|
||
return match.group(0)
|
||
return f"{match.group(1)}{image_dir.name}/{path.name})"
|
||
|
||
return IMAGE_LINK.sub(fix, markdown)
|
||
|
||
|
||
# ===========================================================================
|
||
# EPUB mit pandoc
|
||
# ===========================================================================
|
||
|
||
def heading_depth(markdown: str) -> int:
|
||
"""Tiefste verwendete Überschriftenebene (1–3) für die Kapitelaufteilung."""
|
||
levels = [
|
||
len(m.group(1))
|
||
for m in re.finditer(r"^(#{1,6}) ", markdown, re.MULTILINE)
|
||
]
|
||
return min(max(levels, default=1), 3)
|
||
|
||
|
||
# Tesseract-Sprachcode -> Sprachcode im EPUB (ISO 639-1)
|
||
EPUB_LANGUAGES = {
|
||
"deu": "de", "deu_latf": "de", "frk": "de", "eng": "en", "fra": "fr",
|
||
"spa": "es", "ita": "it", "nld": "nl", "por": "pt", "pol": "pl",
|
||
"ces": "cs", "dan": "da", "swe": "sv", "nor": "no", "fin": "fi",
|
||
"rus": "ru", "ukr": "uk", "lat": "la", "ell": "el", "tur": "tr",
|
||
}
|
||
|
||
|
||
def epub_language(ocr_language: str) -> str:
|
||
"""EPUB-Sprache aus der (ersten) OCR-Sprache: "eng+deu" -> "en"."""
|
||
first = ocr_language.split("+")[0].strip().lower()
|
||
return EPUB_LANGUAGES.get(first, "de")
|
||
|
||
|
||
def default_jobs() -> int:
|
||
"""Halbe Kernzahl, höchstens 8: Jeder Prozess lädt das Layout-Modell."""
|
||
return max(1, min(8, (os.cpu_count() or 2) // 2))
|
||
|
||
|
||
def title_author_from_filename(stem: str) -> tuple[str, str]:
|
||
"""Titel und Autor aus einem Dateinamen nach dem Muster "Autor_-_Titel".
|
||
|
||
"Katja_Adler_-_Rolle_rückwärts" -> ("Rolle rückwärts", "Katja Adler");
|
||
ohne " - " ist der ganze Name der Titel.
|
||
"""
|
||
name = re.sub(r"\s+", " ", stem.replace("_", " ")).strip()
|
||
author, sep, title = name.partition(" - ")
|
||
if sep and author.strip() and title.strip():
|
||
return title.strip(), author.strip()
|
||
return name, ""
|
||
|
||
|
||
# Ergänzung zum Standard-Stylesheet von pandoc: Code auf dem kleinen
|
||
# Bildschirm umbrechen statt abschneiden, Einrückung bleibt erhalten.
|
||
EPUB_CSS = """
|
||
pre, pre code { white-space: pre-wrap; overflow-wrap: anywhere; }
|
||
pre { font-size: 0.85em; line-height: 1.3; }
|
||
"""
|
||
|
||
|
||
def epub_stylesheet(pandoc: str) -> str:
|
||
done = subprocess.run(
|
||
[pandoc, "--print-default-data-file", "epub.css"],
|
||
capture_output=True, text=True,
|
||
)
|
||
return (done.stdout if done.returncode == 0 else "") + EPUB_CSS
|
||
|
||
|
||
def write_epub(
|
||
markdown_path: Path,
|
||
epub_path: Path,
|
||
*,
|
||
title: str,
|
||
author: str,
|
||
language: str,
|
||
cover: Path | None,
|
||
split_level: int,
|
||
) -> None:
|
||
pandoc = shutil.which("pandoc")
|
||
if not pandoc:
|
||
raise RuntimeError("pandoc nicht gefunden – installieren mit: sudo apt install pandoc")
|
||
|
||
cmd = [
|
||
pandoc,
|
||
str(markdown_path),
|
||
"--from", "markdown-yaml_metadata_block+autolink_bare_uris",
|
||
"--to", "epub3",
|
||
"--output", str(epub_path),
|
||
"--resource-path", str(markdown_path.parent),
|
||
f"--split-level={split_level}",
|
||
f"--toc-depth={split_level}",
|
||
"--metadata", f"title={title}",
|
||
"--metadata", f"lang={language}",
|
||
]
|
||
if author:
|
||
cmd += ["--metadata", f"author={author}"]
|
||
if cover:
|
||
cmd += ["--epub-cover-image", str(cover)]
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
css = Path(tmp) / "pdf2md.css"
|
||
css.write_text(epub_stylesheet(pandoc), encoding="utf-8")
|
||
done = subprocess.run(cmd + ["--css", str(css)], capture_output=True, text=True)
|
||
if done.returncode != 0:
|
||
raise RuntimeError(f"pandoc fehlgeschlagen: {done.stderr.strip()}")
|
||
|
||
|
||
# ===========================================================================
|
||
# Metadaten-Kopf, Aufteilen in Einzeldateien, EPUB aus Markdown
|
||
# ===========================================================================
|
||
# Jede erzeugte Markdown-Datei beginnt mit einem YAML-Kopf (title, author, lang,
|
||
# cover …). Wir schreiben und lesen ihn selbst; pandoc bekommt nur den Text,
|
||
# weil "---"-Zeilen im Buchtext sonst als YAML missverstanden werden könnten.
|
||
|
||
FRONTMATTER = re.compile(r"\A---\n(.*?)\n---\n+", re.DOTALL)
|
||
INDEX_LINK = re.compile(r"\]\(([^)\s]+\.md)\)")
|
||
UMLAUTS = str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue", "ß": "ss"})
|
||
|
||
|
||
def make_frontmatter(meta: dict) -> str:
|
||
lines = ["---"]
|
||
for key, value in meta.items():
|
||
if value not in (None, ""):
|
||
lines.append(f"{key}: {json.dumps(value, ensure_ascii=False)}")
|
||
return "\n".join(lines + ["---"]) + "\n\n"
|
||
|
||
|
||
def split_frontmatter(text: str) -> tuple[dict, str]:
|
||
"""YAML-Kopf (nur "schlüssel: wert") abtrennen und auswerten."""
|
||
match = FRONTMATTER.match(text)
|
||
if not match:
|
||
return {}, text
|
||
meta = {}
|
||
for line in match.group(1).splitlines():
|
||
key, sep, value = line.partition(":")
|
||
if sep:
|
||
value = value.strip()
|
||
try:
|
||
meta[key.strip()] = json.loads(value)
|
||
except ValueError:
|
||
meta[key.strip()] = value.strip("'\"")
|
||
return meta, text[match.end():]
|
||
|
||
|
||
def slugify(text: str, limit: int = 50) -> str:
|
||
text = clean_inline(text).lower().translate(UMLAUTS)
|
||
text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode()
|
||
slug = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
|
||
return slug[:limit].rstrip("-") or "abschnitt"
|
||
|
||
|
||
def iter_headings(lines: list[str]):
|
||
"""(Zeilennummer, Ebene, Text) aller Überschriften außerhalb von Codeblöcken."""
|
||
fence = None
|
||
for index, line in enumerate(lines):
|
||
marker = FENCE.match(line)
|
||
if fence:
|
||
if marker and marker.group(1)[0] == fence:
|
||
fence = None
|
||
continue
|
||
if marker:
|
||
fence = marker.group(1)[0]
|
||
elif match := re.match(r"^(#{1,6}) (.+)$", line):
|
||
yield index, len(match.group(1)), match.group(2).strip()
|
||
|
||
|
||
SPLIT_CHARS = 30_000 # ohne Gliederung: Abschnitte von etwa dieser Länge
|
||
|
||
|
||
def split_by_length(markdown: str, size: int = SPLIT_CHARS) -> list[dict]:
|
||
"""Ohne verwertbare Überschriften: an Absatzgrenzen in etwa gleich lange Stücke."""
|
||
blocks, fence = [], False
|
||
for part in re.split(r"(\n\n+)", markdown):
|
||
if blocks and fence:
|
||
blocks[-1] += part # Codeblöcke nicht zerteilen
|
||
elif part.strip() or not blocks:
|
||
blocks.append(part)
|
||
else:
|
||
blocks[-1] += part
|
||
fence ^= sum(1 for line in part.split("\n") if FENCE.match(line)) % 2 == 1
|
||
sections, current = [], ""
|
||
for block in blocks:
|
||
if current and len(current) + len(block) > size:
|
||
sections.append(current)
|
||
current = ""
|
||
current += block
|
||
if current.strip():
|
||
sections.append(current)
|
||
return [
|
||
{"title": f"Abschnitt {n}", "body": body.strip() + "\n"}
|
||
for n, body in enumerate(sections, start=1)
|
||
]
|
||
|
||
|
||
def split_markdown(markdown: str, mode: str) -> list[dict]:
|
||
"""Buch in Abschnitte teilen: "teile" (oberste Ebene) oder "kapitel".
|
||
|
||
Titelei und Inhaltsverzeichnis bilden den Vorspann.
|
||
"""
|
||
lines = markdown.split("\n")
|
||
headings = list(iter_headings(lines))
|
||
counts = Counter(level for _, level, _ in headings)
|
||
levels = sorted(level for level, n in counts.items() if n >= 2 and level <= 3)
|
||
if not levels:
|
||
return split_by_length(markdown)
|
||
part_level, chapter_level = levels[0], levels[-1]
|
||
cut_levels = {part_level} if mode == "teile" else {part_level, chapter_level}
|
||
|
||
# Bis einschließlich Inhaltsverzeichnis (falls am Anfang) alles Vorspann.
|
||
start = 0
|
||
for index, _level, text in headings:
|
||
if CONTENTS_TITLE.match(clean_inline(text)) and index < len(lines) // 5:
|
||
start = index + 1
|
||
break
|
||
cuts = [(i, lvl, t) for i, lvl, t in headings if lvl in cut_levels and i >= start]
|
||
|
||
sections = []
|
||
first = cuts[0][0] if cuts else len(lines)
|
||
front = "\n".join(lines[:first]).strip()
|
||
if front:
|
||
sections.append({"title": "Vorspann", "body": front + "\n"})
|
||
part = ""
|
||
for n, (index, level, text) in enumerate(cuts):
|
||
end = cuts[n + 1][0] if n + 1 < len(cuts) else len(lines)
|
||
title = md_unescape(clean_inline(text))
|
||
if level == part_level:
|
||
part = title
|
||
number = re.match(r"(\d+)\b", title) if level == chapter_level != part_level else None
|
||
sections.append({
|
||
"title": title,
|
||
"part": part if part != title else "",
|
||
"chapter": int(number.group(1)) if number else None,
|
||
"body": "\n".join(lines[index:end]).strip() + "\n",
|
||
})
|
||
return sections
|
||
|
||
|
||
def _page_key(text: str) -> str:
|
||
return re.sub(r"[\W_]+", "", clean_inline(text).lower())
|
||
|
||
|
||
def add_source_pages(doc: pymupdf.Document, sections: list[dict]) -> None:
|
||
"""Seiten der PDF (1-basiert) je Abschnitt.
|
||
|
||
Startseite aus der PDF-Gliederung, sonst die erste Seite (nach der vorigen),
|
||
auf der die Überschrift steht – Verzeichnisseiten ausgenommen.
|
||
"""
|
||
outline = {norm_title(title): page - 1 for _, title, page in doc.get_toc() if page > 0}
|
||
pages = [_page_key(page.get_text()) for page in doc]
|
||
keys = [_page_key(section["title"])[:40] for section in sections]
|
||
listing = { # Verzeichnisse: viele Abschnittstitel auf einer Seite
|
||
i for i, text in enumerate(pages) if sum(1 for k in keys if k and k in text) >= 3
|
||
}
|
||
starts, page = [], 0
|
||
for section, key in zip(sections, keys):
|
||
found = outline.get(norm_title(section["title"]))
|
||
if found is None and section["title"] != "Vorspann" and key:
|
||
found = next(
|
||
(i for i in range(page, len(pages)) if i not in listing and key in pages[i]),
|
||
None,
|
||
)
|
||
page = max(page, found) if found is not None else page
|
||
starts.append(page)
|
||
for n, section in enumerate(sections):
|
||
end = starts[n + 1] if n + 1 < len(sections) else len(pages) - 1
|
||
# Beginnt der nächste Abschnitt oben auf einer Seite, endet dieser davor.
|
||
if n + 1 < len(sections) and end > starts[n] and _starts_page(doc[end], keys[n + 1]):
|
||
end -= 1
|
||
section["source_pages"] = [starts[n] + 1, max(starts[n], end) + 1]
|
||
|
||
|
||
def _starts_page(page: pymupdf.Page, key: str) -> bool:
|
||
"""Steht über der Überschrift nur Kolumnentitel/Seitenzahl (≤ 3 Wörter je Block)?"""
|
||
blocks = [b for b in page.get_text("blocks") if b[6] == 0 and b[4].strip()]
|
||
heading = next((b for b in blocks if key and key in _page_key(b[4])), None)
|
||
if heading is None or not _page_key(heading[4]).startswith(key):
|
||
return False # nicht gefunden oder Text davor im selben Block
|
||
above = [b for b in blocks if b[3] <= heading[1] + 1 and b is not heading]
|
||
return all(len(b[4].split()) <= 3 for b in above)
|
||
|
||
|
||
def write_split(folder: Path, sections: list[dict], meta: dict) -> list[Path]:
|
||
"""Abschnitte als nummerierte Dateien plus index.md (Reihenfolge) schreiben."""
|
||
folder.mkdir(parents=True, exist_ok=True)
|
||
width = max(2, len(str(len(sections))))
|
||
files = []
|
||
for order, section in enumerate(sections):
|
||
name = f"{order:0{width}d}_{slugify(section['title'])}.md"
|
||
head = {
|
||
"title": section["title"],
|
||
"part": section.get("part"),
|
||
"chapter": section.get("chapter"),
|
||
"order": order,
|
||
"source_pages": section.get("source_pages"),
|
||
"book": meta.get("title"),
|
||
"author": meta.get("author"),
|
||
}
|
||
(folder / name).write_text(make_frontmatter(head) + section["body"], encoding="utf-8")
|
||
files.append(folder / name)
|
||
|
||
entries = "\n".join(
|
||
f"- [{md_escape(s['title'])}]({f.name})" for s, f in zip(sections, files)
|
||
)
|
||
index = (
|
||
make_frontmatter(meta)
|
||
+ "<!-- Reihenfolge der Dateien im Buch; zum Umsortieren Zeilen verschieben. -->\n\n"
|
||
+ entries + "\n"
|
||
)
|
||
(folder / "index.md").write_text(index, encoding="utf-8")
|
||
return files
|
||
|
||
|
||
def read_markdown_source(path: Path) -> tuple[dict, str, Path]:
|
||
"""Metadaten, Text und Basisordner aus einer .md-Datei oder einem Split-Ordner."""
|
||
if path.is_dir():
|
||
index = path / "index.md"
|
||
if not index.is_file():
|
||
raise RuntimeError(f"index.md fehlt in {path}")
|
||
meta, listing = split_frontmatter(index.read_text(encoding="utf-8"))
|
||
parts = []
|
||
for name in INDEX_LINK.findall(listing):
|
||
file = path / name
|
||
if not file.is_file():
|
||
raise RuntimeError(f"In index.md genannt, aber nicht vorhanden: {file}")
|
||
parts.append(split_frontmatter(file.read_text(encoding="utf-8"))[1].strip())
|
||
if not parts:
|
||
raise RuntimeError(f"index.md nennt keine Dateien: {index}")
|
||
return meta, "\n\n".join(parts) + "\n", path
|
||
meta, body = split_frontmatter(path.read_text(encoding="utf-8"))
|
||
return meta, body, path.parent
|
||
|
||
|
||
def markdown_to_epub(
|
||
source: Path,
|
||
epub_path: Path,
|
||
*,
|
||
title: str | None = None,
|
||
author: str | None = None,
|
||
language: str | None = None,
|
||
) -> None:
|
||
"""EPUB aus Markdown-Datei oder Split-Ordner; Angaben aus dem YAML-Kopf."""
|
||
meta, body, base = read_markdown_source(source)
|
||
cover = base / meta["cover"] if meta.get("cover") else None
|
||
with tempfile.NamedTemporaryFile(
|
||
"w", encoding="utf-8", suffix=".md", dir=base, delete=False
|
||
) as tmp:
|
||
tmp.write(body)
|
||
try:
|
||
write_epub(
|
||
Path(tmp.name),
|
||
epub_path,
|
||
title=title or meta.get("title") or source.stem,
|
||
author=author or meta.get("author") or "",
|
||
language=language or meta.get("lang") or "de",
|
||
cover=cover if cover and cover.is_file() else None,
|
||
split_level=heading_depth(body),
|
||
)
|
||
finally:
|
||
Path(tmp.name).unlink(missing_ok=True)
|
||
|
||
|
||
# ===========================================================================
|
||
# Kommandozeile
|
||
# ===========================================================================
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(
|
||
description=(
|
||
"PDF (digital oder gescannt) in Markdown umwandeln; Fließtext nach "
|
||
"Satzenden gliedern, optional Grafiken speichern und EPUB erzeugen."
|
||
)
|
||
)
|
||
parser.add_argument("pdf", type=Path, nargs="?", help="Eingabe-PDF")
|
||
parser.add_argument(
|
||
"markdown",
|
||
type=Path,
|
||
nargs="?",
|
||
help="Ausgabe-Datei; Standard: gleicher Name mit .md",
|
||
)
|
||
parser.add_argument(
|
||
"--ocr",
|
||
choices=["auto", "always", "never"],
|
||
default="auto",
|
||
help=(
|
||
"auto: nur Scanseiten per OCR lesen, gerade vorhandene OCR-Ebenen "
|
||
"übernehmen, schiefe neu erkennen (Standard); always: alle Seiten "
|
||
"neu erkennen; never: keine OCR"
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--no-ocr",
|
||
dest="ocr",
|
||
action="store_const",
|
||
const="never",
|
||
help="Kurzform für --ocr never",
|
||
)
|
||
parser.add_argument(
|
||
"--ocr-language",
|
||
default="deu",
|
||
help="Tesseract-Sprachcode bei OCR, z. B. deu+eng (Standard: deu)",
|
||
)
|
||
parser.add_argument(
|
||
"--ocr-dpi",
|
||
type=int,
|
||
default=300,
|
||
help="Auflösung für die OCR (Standard: 300)",
|
||
)
|
||
parser.add_argument(
|
||
"--jobs",
|
||
type=int,
|
||
default=default_jobs(),
|
||
help="Anzahl paralleler OCR-Prozesse (Standard: halbe Kernzahl, "
|
||
"höchstens 8; hier %(default)s)",
|
||
)
|
||
parser.add_argument(
|
||
"--images",
|
||
action="store_true",
|
||
help="Grafiken speichern und in Markdown verlinken",
|
||
)
|
||
parser.add_argument(
|
||
"--image-format",
|
||
choices=["jpg", "png"],
|
||
default="jpg",
|
||
help="Bildformat: jpg (klein, Standard) oder png (verlustfrei, groß)",
|
||
)
|
||
parser.add_argument(
|
||
"--epub",
|
||
action="store_true",
|
||
help="zusätzlich EPUB mit pandoc erzeugen (schaltet --images ein)",
|
||
)
|
||
parser.add_argument(
|
||
"--epub-file",
|
||
type=Path,
|
||
metavar="PFAD",
|
||
help="Pfad des EPUB; Standard: gleicher Name wie Markdown mit .epub",
|
||
)
|
||
parser.add_argument(
|
||
"--jpeg-quality",
|
||
type=int,
|
||
default=DEFAULT_JPEG_QUALITY,
|
||
metavar="1-100",
|
||
help="JPEG-Qualität der Bilder (Standard: %(default)s; 95 = fast verlustfrei, größer)",
|
||
)
|
||
parser.add_argument(
|
||
"--split",
|
||
choices=["teile", "kapitel"],
|
||
help="Teile bzw. Kapitel als einzelne Markdown-Dateien in einem Ordner "
|
||
"(mit index.md für die Reihenfolge) statt einer Datei",
|
||
)
|
||
parser.add_argument(
|
||
"--from-md",
|
||
type=Path,
|
||
metavar="PFAD",
|
||
help="nur EPUB erzeugen: aus einer (bearbeiteten) Markdown-Datei oder einem "
|
||
"--split-Ordner; Titel, Autor, Cover aus dem YAML-Kopf",
|
||
)
|
||
parser.add_argument("--title", help="Titel im EPUB (statt PDF-Metadaten/Dateiname)")
|
||
parser.add_argument("--author", help="Autor im EPUB (statt PDF-Metadaten/Dateiname)")
|
||
parser.add_argument(
|
||
"--lang",
|
||
help="Sprache im EPUB, z. B. en (Standard: aus --ocr-language, "
|
||
"deu -> de, eng -> en)",
|
||
)
|
||
parser.add_argument(
|
||
"--keep-headers",
|
||
action="store_true",
|
||
help="Seitenköpfe/Kolumnentitel übernehmen",
|
||
)
|
||
parser.add_argument(
|
||
"--keep-footers",
|
||
action="store_true",
|
||
help="Seitenfüße/Seitenzahlen übernehmen",
|
||
)
|
||
args = parser.parse_args()
|
||
if not 1 <= args.jpeg_quality <= 100:
|
||
parser.error("--jpeg-quality muss zwischen 1 und 100 liegen.")
|
||
|
||
if args.from_md:
|
||
return epub_from_markdown(args, parser)
|
||
if args.pdf is None:
|
||
parser.error("Eingabe-PDF fehlt (oder --from-md PFAD angeben).")
|
||
args.lang = args.lang or epub_language(args.ocr_language)
|
||
|
||
source = args.pdf.expanduser()
|
||
target = (args.markdown or source.with_suffix(".md")).expanduser()
|
||
|
||
if not source.is_file():
|
||
parser.error(f"Eingabedatei nicht gefunden: {source}")
|
||
|
||
if source.resolve() == target.resolve():
|
||
parser.error("Eingabe- und Ausgabedatei müssen verschieden sein.")
|
||
|
||
epub_path = None
|
||
if args.epub or args.epub_file:
|
||
args.images = True # EPUB ohne Bilder wäre unvollständig
|
||
epub_path = (args.epub_file or target.with_suffix(".epub")).expanduser()
|
||
|
||
# --split: Ordner "<name>/" mit Einzeldateien und "images/" statt einer Datei
|
||
folder = target.with_suffix("") if args.split else None
|
||
image_dir = folder / "images" if folder else target.parent / f"{target.stem}_images"
|
||
|
||
try:
|
||
with pymupdf.open(source) as doc:
|
||
kinds, plan = plan_pages(doc, args.ocr, args.ocr_language, args.ocr_dpi)
|
||
outline = load_outline(doc)
|
||
uris = load_link_uris(doc)
|
||
anchors = load_link_anchors(doc)
|
||
page_count = doc.page_count
|
||
file_title, file_author = title_author_from_filename(source.stem)
|
||
title = (
|
||
args.title or ((doc.metadata or {}).get("title") or "").strip() or file_title
|
||
)
|
||
author = (
|
||
args.author or ((doc.metadata or {}).get("author") or "").strip() or file_author
|
||
)
|
||
|
||
counts = Counter(kinds)
|
||
print(
|
||
f"{page_count} Seiten: {counts[PAGE_TEXT]} Text, "
|
||
f"{counts[PAGE_SCAN]} Scan/Bild, {counts[PAGE_OCR_LAYER]} mit "
|
||
f"OCR-Ebene, {counts[PAGE_EMPTY]} leer → "
|
||
f"{plan.count(RUN_OCR)} Seiten werden per OCR gelesen"
|
||
)
|
||
unread = sum(
|
||
1 for k, run in zip(kinds, plan) if k == PAGE_SCAN and run != RUN_OCR
|
||
)
|
||
if unread:
|
||
print(
|
||
f"Hinweis: {unread} Bildseite(n) ohne Text bleiben ohne OCR "
|
||
"(Fotos/Cover oder --ocr never)."
|
||
)
|
||
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
cover = None
|
||
if args.images:
|
||
# Cover immer mit ablegen, damit --from-md es später findet.
|
||
image_dir.mkdir(parents=True, exist_ok=True)
|
||
cover = image_dir / f"cover.{args.image_format}"
|
||
_jpeg_quality[0] = args.jpeg_quality
|
||
doc[0].get_pixmap(dpi=150).save(cover)
|
||
|
||
options = dict(
|
||
keep_headers=args.keep_headers,
|
||
keep_footers=args.keep_footers,
|
||
images=args.images,
|
||
image_path=str(image_dir.resolve()) if args.images else "",
|
||
image_format=args.image_format,
|
||
ocr_language=args.ocr_language,
|
||
ocr_dpi=args.ocr_dpi,
|
||
jpeg_quality=args.jpeg_quality,
|
||
)
|
||
markdown = extract_markdown(source, plan, options, args.jobs)
|
||
|
||
if args.images:
|
||
markdown = make_image_links_relative(markdown, image_dir)
|
||
|
||
markdown = normalize_markdown(
|
||
markdown,
|
||
outline,
|
||
drop_page_numbers=not args.keep_footers,
|
||
drop_running_heads=not args.keep_headers,
|
||
running_head_min=max(3, page_count // 10),
|
||
)
|
||
markdown = link_urls(markdown, uris)
|
||
markdown = link_anchors(markdown, anchors)
|
||
|
||
base = folder or target.parent
|
||
meta = {
|
||
"title": title,
|
||
"author": author,
|
||
"lang": args.lang,
|
||
"cover": cover.relative_to(base).as_posix() if cover else None,
|
||
}
|
||
if folder:
|
||
sections = split_markdown(markdown, args.split)
|
||
if sections and sections[0]["title"] == "Abschnitt 1":
|
||
print("Hinweis: keine Gliederung erkannt – nach Länge geteilt.")
|
||
with pymupdf.open(source) as doc:
|
||
add_source_pages(doc, sections)
|
||
files = write_split(folder, sections, meta)
|
||
else:
|
||
target.write_text(make_frontmatter(meta) + markdown, encoding="utf-8")
|
||
|
||
if epub_path:
|
||
markdown_to_epub(folder or target, epub_path)
|
||
|
||
except Exception as exc:
|
||
print(f"Fehler bei der Konvertierung: {exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
if folder:
|
||
print(f"Markdown geschrieben: {len(files)} Dateien in {folder} (Reihenfolge: index.md)")
|
||
else:
|
||
print(f"Markdown geschrieben: {target}")
|
||
if args.images:
|
||
print(f"Bilder gespeichert in: {image_dir}")
|
||
if epub_path:
|
||
print(f"EPUB geschrieben: {epub_path}")
|
||
|
||
return 0
|
||
|
||
|
||
def epub_from_markdown(args, parser) -> int:
|
||
"""--from-md: EPUB aus bearbeitetem Markdown bzw. Split-Ordner erzeugen."""
|
||
source = args.from_md.expanduser()
|
||
if not source.exists():
|
||
parser.error(f"Nicht gefunden: {source}")
|
||
base = source if source.is_dir() else source.with_suffix("")
|
||
epub_path = (args.epub_file or base.with_suffix(".epub")).expanduser()
|
||
try:
|
||
markdown_to_epub(
|
||
source, epub_path, title=args.title, author=args.author, language=args.lang
|
||
)
|
||
except Exception as exc:
|
||
print(f"Fehler beim EPUB-Erzeugen: {exc}", file=sys.stderr)
|
||
return 1
|
||
print(f"EPUB geschrieben: {epub_path}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|