dschlueter/pdf2md
Watch
1
0
Fork
You've already forked pdf2md
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity
pdf2md/pdf2md.py
Dieter Schlüter 5cd5e5d69b --split seiten, --epub-per-file, Nachspann, --from-md --split, --split-chars 
- --split seiten mit --pages-per-file N (Standard 20, Ausrichtung an
  Überschriften bis eine Seite neben der Grenze) oder --page-ranges
  (streng; Seiten außerhalb werden eigene Dateien)
- Schnitt nur im fertigen Gesamt-Markdown (lückenlos): vor dem ersten
  Absatz, der auf der Grenzseite beginnt; Überschriften vor dem Schnitt
  wandern in die neue Datei; source_pages aus Anfang/Ende jedes Absatzes
- Seitenzuordnung über die Rohtexte je Seite (pymupdf4llm page_chunks=True,
  aneinandergehängt byte-gleich; Band 1/2 unverändert), auch für OCR-Seiten
- --epub-per-file: ein EPUB je Einzeldatei in <name>/epub/
- Nachspann: Anhang/Autor/Impressum nach dem letzten Kapitel als eigene Datei
- --from-md DATEI --split teile|kapitel: bearbeitetes Markdown nachträglich
  teilen, Bild- und Cover-Pfade angepasst
- --split-chars N; Längen-Abschnitte tragen die Überschrift am Anfang als Titel
- Tests (tests/test_split_pages.py), README, Bedienungsanleitung, CLAUDE.md, TODO

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-03 01:04:50 +02:00

2963 lines
115 KiB
Python
Raw Blame History

#!/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 bisect
import contextlib
import difflib
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 OCRMode, tesseract_api
# ===========================================================================
# Markdown-Muster
# ===========================================================================
HEADING = re.compile(r"^\s{0,3}(#{1,6})\s*(\S.*?)\s*#*\s*$")
# Schließende Anführungszeichen (deutsch U+201E...U+201C, U+201A...U+2018, U+00BB...U+00AB;
# englisch; Schweiz U+00AB...U+00BB) - als Escapes, damit keine verwechselbaren Zeichen im Quelltext stehen.
# Klammern und Markdown-Hervorhebungen.
CLOSERS = r"""\u201d\u201c\u2019\u2018"'\u00bb\u00ab\u203a\u2039)\]*_"""
# Satzzeichen, optional gefolgt von schließenden Zeichen (z. B. "...da?" + U+00AB).
SENTENCE_END = re.compile(rf"""[.!?\u2026]+[{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"""(?:[.\u2026][{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( # Format bis pymupdf4llm 1.27 bzw. ab 1.28
r"(?:\*\*----- Start of picture text -----\*\*<br>|<!-- Start of picture text -->)"
r".*?"
r"(?:\*\*----- End of picture text -----\*\*<br>|<!-- End of picture text -->)\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"^(?:[-\u2013\u2014=]\s*)?(?:(?:seite|page|s\.)\s*)?\d{1,4}(?:\s*[-\u2013\u2014=])?$"
r"|^[-\u2013\u2014=]\s*[ivxlc]{1,7}\s*[-\u2013\u2014=]$", # "- VIII -" (OCR: "= 58 -")
re.IGNORECASE,
)
QUOTE_CHARS = set("\u00ab\u00bb\u201e\u201c\u201d\"\u201a\u2018\u2019\u203a\u2039")
DASHES = str.maketrans({"\u2013": "-", "\u2014": "-", "\u2010": "-", "\u2011": "-"})
# Silbentrennung am Zeilen-/Blockende (Bindestrich, weicher Trennstrich U+00AD, U+00AC).
HYPHEN_END = re.compile(r"[^\W\d_]{2,}[-\u00ad\u00ac]$")
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 "\u2013\u2014-\u2026,;" 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(" .-\u2013\u2014|\u00b7").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) and drop_page_numbers and PAGE_NUMBER.match(
clean_inline(line.removesuffix(PARAGRAPH_END))
):
continue # Seitenzahl am Strophen-/Seitenende
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 not HAS_ALNUM.search(text):
continue # Überschrift nur aus Zeichen ("# ||", OCR-Rauschen)
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 = ".,;:!?)]}\u00bb\u00ab\u201c\u201d\u201e\"'*_\\" # "\\": 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)
# "und/oder", "Ein/Aus": nur Buchstaben und ein "/" - ein Wortpaar, kein URL-Pfad.
WORD_PAIR = re.compile(r"^[^\W\d_]+/[^\W\d_]+$")
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]:
word = token.strip().rstrip(URL_TRAILING)
extension_only = "/" not in word # "html": nur nach "artikel."
if (
parts[-1][-1] in URL_BREAK_AFTER
and URL_CONTINUATION.match(token.rstrip())
and not WORD_PAIR.match(word)
and not (extension_only and not parts[-1].endswith("."))
):
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)
LIST_MARKER = re.compile(r"^\s*(?:[-*+]|\d+[.)])\s+")
URL_PATH = re.compile(r"^[\w\-./%?=&#~+:]+$")
def _url_tail(line: str) -> str | None:
"""URL am Zeilenende (unmaskiert), sonst None."""
plain = _unescape_map(line)[0].rstrip()
tokens = plain.split()
if not tokens:
return None
for n in range(len(tokens) - 1, -1, -1):
if URL_START.search(tokens[n]):
if all(URL_CONTINUATION.match(t) for t in tokens[n + 1:]):
return "".join(tokens[n:])
return None
return None
def _join_split_urls(lines: list[str], known: dict[str, str]) -> list[str]:
"""URL, die in einem neuen Block weitergeht (oft als Listenpunkt "- ..."),
mit ihrer Fortsetzung verbinden."""
i = 0
while i < len(lines):
tail = _url_tail(lines[i])
j = i + 1
while j < len(lines) and not lines[j].strip() and j - i <= 2:
j += 1
if tail is None or j >= len(lines):
i += 1
continue
nxt = lines[j]
if HEADING.match(nxt) or FENCE.match(nxt) or IMAGE.match(nxt) or nxt.lstrip().startswith(("|", ">")):
i += 1
continue
rest = LIST_MARKER.sub("", nxt, count=1).lstrip(" \t" + NBSP)
first = _unescape_map(rest)[0].split()
if not first:
i += 1
continue
piece = first[0].rstrip(URL_TRAILING)
joined = url_key(tail.rstrip(URL_TRAILING) + piece)
if known:
fits = url_key(tail.rstrip(URL_TRAILING)) not in known and any(
key.startswith(joined) for key in known
)
else:
fits = (
tail.endswith(("-", "/", "_"))
and URL_PATH.match(piece) is not None
and (piece[:1].islower() or piece[:1].isdigit())
and any(c in piece for c in "-/.")
and not WORD_PAIR.match(piece)
)
if not fits:
i += 1
continue
lines[i] = lines[i].rstrip() + rest # ohne Leerzeichen: dieselbe URL
del lines[i + 1:j + 1]
return lines
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 = _join_split_urls(markdown.split("\n"), known)
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")
break
else:
span = _fuzzy_anchor(markdown, pos, before, text)
if span is None:
continue
start, end = span
markdown = f"{markdown[:start]}[{markdown[start:end]}]({uri}){markdown[end:]}"
pos = start
return markdown
def _fuzzy_anchor(
markdown: str, pos: int, before: list[str], text: list[str]
) -> tuple[int, int] | None:
"""Linktext mit Abweichungen (verlorene Ligatur: "Verifcation") finden:
Anfang über Kontext und erste Wörter, Ende über das letzte Wort."""
if len(text) < 3 or not (context := _context_pattern(before)):
return None
head = re.compile(context + CONTEXT_GAP + r"(?P<anchor>" + _words_pattern(text[:2]) + r")")
match = next(
(m for m in head.finditer(markdown, pos) if not _inside_link(markdown, m.start("anchor"))),
None,
)
if match is None:
return None
start = match.start("anchor")
window = markdown[start:start + int(len(" ".join(text)) * 1.2) + 10]
if "\n\n" in window:
window = window[:window.index("\n\n")] # nicht über den Absatz hinaus
last = None
for last in re.finditer(_words_pattern(text[-1:]), window):
pass
return (start, start + last.end()) if last else None
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 U+00AC kennt pymupdf4llm nicht ("wach<U+00AD> 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", "\u00ac")
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("char_flags", VISIBLE_TEXT) & VISIBLE_TEXT: # OCR-Text
span = {**span, "text": OCR_NOISE.sub("", span["text"])} # Scanrand-"|"
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 (x ü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
_page_text: list[tuple[float, float] | None] = [None] # Textbereich aus echten Zeilen
_keep_lines = [False] # --keep-lines: jede PDF-Zeile bleibt eine Zeile
REAL_WORD = re.compile(r"[^\W\d_]{3,}")
def _dedupe_box_lines(boxes) -> None:
"""Überlappende Layout-Blöcke enthalten dieselben Zeilen doppelt
(Gedichtseiten in Scans) - jede Zeile nur im ersten Block behalten."""
seen: set = set()
for box in boxes:
lines = getattr(box, "textlines", None)
if not lines or getattr(box, "boxclass", None) not in TEXT_BOXES:
continue
kept = []
for line in lines:
key = (tuple(round(v) for v in line["bbox"]),
"".join(span["text"] for span in line["spans"]))
if key not in seen:
seen.add(key)
kept.append(line)
box.textlines = kept
def _remember_page_lines(boxes) -> None:
_dedupe_box_lines(boxes)
_page_lines.clear()
real: list[tuple[float, float]] = []
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]))
text = "".join(span["text"] for span in line["spans"])
if len(REAL_WORD.findall(text)) >= 2: # keine Seitenzahl, kein Scan-Rauschen
real.append((line["bbox"][0], line["bbox"][2]))
_page_text[0] = (min(x0 for x0, _ in real), max(x1 for _, x1 in real)) if real else None
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_monospace(lines: list[dict]) -> bool:
return 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()
)
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 _is_monospace(lines):
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*[-\u2013]\s*\d+)?(?:\s*,\s*\d+(?:\s*[-\u2013]\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}"
OCR_NOISE = re.compile(r"[|\u00a6]") # Buchkante/Seitenrand als "|" gelesen
HAS_ALNUM = re.compile(r"[^\W_]")
def _clean_ocr_lines(textlines: list[dict]) -> list[dict]:
"""OCR-Seiten: Zeilen mit Wörtern auf verschiedenen Grundlinien trennen
(schiefer Scan: "UndDon tauchteweißen") und Rauschen entfernen."""
result = []
for line in textlines:
groups: list[list[dict]] = []
for span in sorted(line["spans"], key=lambda s: s["origin"][1]):
text = OCR_NOISE.sub("", span["text"])
if not text.strip():
continue
span = {**span, "text": text}
if groups and abs(span["origin"][1] - groups[-1][0]["origin"][1]) <= 0.4 * span["size"]:
groups[-1].append(span)
else:
groups.append([span])
for spans in groups:
spans.sort(key=lambda s: s["bbox"][0])
if not HAS_ALNUM.search("".join(s["text"] for s in spans)):
continue # nur Rauschen (U+201E + "+", "%*")
bbox = (
min(s["bbox"][0] for s in spans), min(s["bbox"][1] for s in spans),
max(s["bbox"][2] for s in spans), max(s["bbox"][3] for s in spans),
)
result.append({**line, "bbox": bbox, "spans": spans})
result.sort(key=lambda line: (line["bbox"][1], line["bbox"][0]))
return result
def _is_ocr_text(textlines: list[dict]) -> bool:
"""OCR-Text ist unsichtbar (Render-Modus 3) - wie utils.is_ocr_text in pymupdf4llm."""
spans = [s for line in textlines for s in line["spans"] if s["text"].strip()]
return bool(spans) and not any(s.get("char_flags", VISIBLE_TEXT) & VISIBLE_TEXT for s in spans)
def _outside_text_area(lines: list[dict], em: float) -> bool:
"""Block ganz links/rechts neben dem Textbereich (OCR-Rauschen vom Scanrand)."""
if _page_text[0] is None:
return False
left, right = _page_text[0]
bx0 = min(line["bbox"][0] for line in lines)
bx1 = max(line["bbox"][2] for line in lines)
return bx1 < left - 0.5 * em or bx0 > right + 0.5 * em
def lines_to_md(textlines: list[dict], original, ignore_code: bool = False) -> str:
"""Textblock ausgeben: Fließtext wie bisher, sonst zeilengetreu."""
ocr = _is_ocr_text(textlines)
if ocr:
textlines = _clean_ocr_lines(textlines)
if not textlines:
return ""
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
if ocr and _outside_text_area(lines, em):
return "" # OCR-Seite: Rauschen am Rand verwerfen
if _keep_lines[0]:
return _keep_lines_md(lines, em, ignore_code)
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 _keep_lines_md(lines: list[dict], em: float, ignore_code: bool) -> str:
"""--keep-lines: jede Zeile bleibt eine Zeile, nichts wird zusammengefügt.
Strophen-/Absatzabstände werden Leerzeilen, Einrückungen bleiben. Ein Block
mit mehreren Zeilen endet mit Absatzende (Strophe), eine Einzelzeile mit
hartem Umbruch (sie kann zur nächsten Zeile gehören).
"""
logical = [[line] for line in lines]
levels = _indent_levels(lines, em)
if len(lines) > 1 and not ignore_code and _is_monospace(lines):
return _code_block(logical, levels)
pitches = [b["bbox"][1] - a["bbox"][1] for a, b in zip(lines, lines[1:])]
usual = sorted(pitches)[len(pitches) // 2] if pitches else 0
gaps = [usual > 0 and pitch > STANZA_GAP * usual for pitch in pitches]
end = PARAGRAPH_END if len(lines) > 1 else HARD_BREAK
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,}(?:[-'\u2019][^\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 Grad
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"[-'\u2019]", "", 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(".,;:!?\u00bb\u00ab\u201e\u201c\u201d\"'\u201a\u2018\u2019()").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) -> list[str]:
"""Markdown je Seite (in der Reihenfolge von pages)."""
_jpeg_quality[0] = options.get("jpeg_quality", DEFAULT_JPEG_QUALITY) # auch im Worker
_keep_lines[0] = options.get("keep_lines", False)
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"],
# Text je Seite; aneinandergehängt identisch mit page_chunks=False
page_chunks=True,
)
if run == RUN_OCR:
# FORCE_DROP_OLD: vorhandene (schlechte) OCR-Ebene verwerfen und neu
# erkennen; force_ocr=True hieße FORCE_KEEP_OLD und behielte den alten Text.
kwargs.update(
use_ocr=OCRMode.FORCE_DROP_OLD,
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()):
if run != RUN_OCR:
return [chunk["text"] for chunk in pymupdf4llm.to_markdown(source, **kwargs)]
if options.get("keep_lines"):
with pymupdf.open(source) as doc:
return [
ocr_lines_md(
doc[pno], options["ocr_language"], options["ocr_dpi"],
options.get("titles", {}).get(pno, []),
)
for pno in pages
]
# Fremde OCR-Ebenen entfernt pymupdf4llm nicht (nur eigene); sonst stünde
# alter und neuer Text doppelt da. Nur die geöffnete Kopie wird geändert.
with pymupdf.open(source) as doc:
for pno in pages:
drop_invisible_text(doc[pno])
return [chunk["text"] for chunk in pymupdf4llm.to_markdown(doc, **kwargs)]
# --keep-lines auf Scanseiten: Zeilen direkt aus Tesseract, ohne Layout-Modell.
# Dieses legt bei Gedicht-Scans eng stehende Verse zusammen ("UndDon tauchte...")
# und gibt Zeilen sich überlappender Blöcke doppelt aus.
OCR_TITLE_LINES = 8 # Titel suchen wir unter den ersten Zeilen der Seite
OCR_STANZA_GAP = 1.4 # Grundlinienabstand (x üblichem) für eine Strophengrenze
PAGE_LABEL = re.compile(r"^seite\s+\d+$", re.IGNORECASE) # Gliederungseintrag "Seite 4"
def _ocr_segments(page: pymupdf.Page, language: str, dpi: int) -> list[list]:
"""Textzeilen-Stücke [x0, y0, x1, y1, text] aus Tesseract, ohne Rauschen."""
pix = page.get_pixmap(dpi=dpi) # RGB: MuPDFs OCR liest Graustufen nicht
segments = []
with pymupdf.open("pdf", pix.pdfocr_tobytes(language=language)) as ocr:
for block in ocr[0].get_text("dict")["blocks"]:
for line in block.get("lines", []):
text = OCR_NOISE.sub("", "".join(s["text"] for s in line["spans"]))
text = re.sub(r"\s+", " ", text).strip()
if HAS_ALNUM.search(text): # Grundlinie für Zeilen- und Strophenabstand
segments.append([*line["bbox"], text, line["spans"][0]["origin"][1]])
real = [seg for seg in segments if len(REAL_WORD.findall(seg[4])) >= 2]
if real: # Spalte aus echten Zeilen; Fetzen am Scanrand liegen außerhalb
left = min(seg[0] for seg in real)
right = max(seg[2] for seg in real)
height = sorted(seg[3] - seg[1] for seg in real)[len(real) // 2]
segments = [
seg for seg in segments
if seg[2] > left - height and seg[0] < right + height
and (len(REAL_WORD.findall(seg[4])) >= 1 or PAGE_NUMBER.match(seg[4]))
]
return segments
def _ocr_rows(segments: list[list]) -> list[list]:
"""Stücke auf gleicher Höhe zu Zeilen verbinden: [x0, y0, x1, y1, text].
Verglichen wird mit dem Mittelpunkt des ersten Stücks einer Zeile - ein
wachsender Bereich würde sonst die Nachbarzeilen mit einsammeln."""
if not segments:
return []
heights = sorted(seg[3] - seg[1] for seg in segments)
height = heights[len(heights) // 2]
rows: list[list] = [] # [x0, y0, x1, y1, stücke, mittelpunkt]
for seg in sorted(segments, key=lambda s: (s[1] + s[3]) / 2):
center = (seg[1] + seg[3]) / 2
row = next((r for r in rows if abs(r[5] - center) < 0.5 * height
and all(seg[0] >= o[2] or seg[2] <= o[0] for o in r[4])), None)
if row is None:
rows.append([*seg[:4], [seg], center])
else:
row[0], row[1] = min(row[0], seg[0]), min(row[1], seg[1])
row[2], row[3] = max(row[2], seg[2]), max(row[3], seg[3])
row[4].append(seg)
rows.sort(key=lambda r: r[5])
return [
[*r[:4], " ".join(seg[4] for seg in sorted(r[4], key=lambda s: s[0])),
max(seg[5] for seg in r[4])]
for r in rows
]
def _title_key(text: str) -> str:
return re.sub(r"[\W\d_]+", "", text.lower())
def ocr_lines_md(page: pymupdf.Page, language: str, dpi: int, titles: list) -> str:
"""Scanseite zeilengetreu: Verse, Strophen (Abstand), Einrückung, Titel aus
der PDF-Gliederung (richtige Schreibweise statt OCR-Lesart)."""
rows = _ocr_rows(_ocr_segments(page, language, dpi))
if not rows:
return ""
heights = sorted(r[3] - r[1] for r in rows)
height = heights[len(heights) // 2]
left = min(r[0] for r in rows if not PAGE_NUMBER.match(r[4]))
pitches = sorted(b[5] - a[5] for a, b in zip(rows, rows[1:])) or [height]
pitch = pitches[len(pitches) // 2] # Abstand der Grundlinien
headings: dict[int, str] = {}
for level, title in titles: # Gliederungstitel dieser Seite wiederfinden
key = _title_key(title)
scored = [
(difflib.SequenceMatcher(None, key, _title_key(r[4])).ratio(), i)
for i, r in enumerate(rows[:OCR_TITLE_LINES]) if i not in headings
]
if scored and (best := max(scored))[0] >= 0.6:
headings[best[1]] = "#" * min(level, 6) + " " + md_escape(title)
out: list[str] = []
for i, row in enumerate(rows):
if i in headings:
out += ["", headings[i], ""]
continue
indent = int(max(0.0, row[0] - left) / (1.2 * height))
text = TEXT_INDENT * indent + md_escape(row[4])
gap = i + 1 < len(rows) and rows[i + 1][5] - row[5] > OCR_STANZA_GAP * pitch
last = i + 1 == len(rows) or (i + 1) in headings
out.append(text + (PARAGRAPH_END if gap or (last and i + 1 < len(rows)) else HARD_BREAK))
if gap:
out.append("")
return "\n".join(out).strip("\n") + "\n\n"
def drop_invisible_text(page: pymupdf.Page) -> None:
"""Unsichtbaren Text (vorhandene OCR-Ebene) entfernen; Bild und Grafik bleiben."""
page.add_redact_annot(page.rect, fill=False)
page.apply_redactions(
images=pymupdf.PDF_REDACT_IMAGE_NONE,
graphics=pymupdf.PDF_REDACT_LINE_ART_NONE,
text=pymupdf.PDF_REDACT_TEXT_REMOVE_INVISIBLE,
)
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
) -> tuple[str, dict[int, str]]:
"""Gesamt-Markdown und Rohtext je Seite (für --split seiten)."""
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 = []
page_texts: dict[int, str] = {}
def collect(pages: list[int], texts: list[str]) -> None:
page_texts.update(zip(pages, texts))
parts.append("".join(texts))
# 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):
collect(pages, 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):
collect(pages, future.result())
report(i, run, pages)
return "\n\n".join(parts), page_texts
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
PAGES_PER_FILE = 20 # --split seiten: Standard für --pages-per-file
HEADING_BLOCK = re.compile(r"^(#{1,6}) (.+)$")
# Nachspann: erste dieser Überschriften nach dem letzten Kapitel/Teil.
BACKMATTER_TITLE = re.compile(
r"^(?:nachwort|nachbemerkung|anhang|anhänge|der autor|die autorin|"
r"über den autor|über die autorin|zum autor|zur autorin|impressum|danksagung|dank|"
r"quellen|quellenverzeichnis|quellennachweis|literatur|literaturverzeichnis|"
r"bibliographie|bibliografie|register|personenregister|sachregister|"
r"stichwortverzeichnis|glossar|anmerkungen|endnoten|bildnachweis|"
r"about the author|acknowledgements|acknowledgments|appendix|bibliography|"
r"glossary|index|notes)\b",
re.IGNORECASE,
)
def md_blocks(markdown: str) -> list[str]:
"""Absätze samt folgendem Leerraum; Codeblöcke bleiben ganz.
Aneinandergehängt ergeben die Blöcke wieder genau den Text.
"""
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
return blocks
def block_heading(block: str) -> str | None:
"""Text der Überschrift, wenn der Block nur aus einer Überschrift besteht."""
lines = block.strip().split("\n")
match = HEADING_BLOCK.match(lines[0]) if len(lines) == 1 else None
return md_unescape(clean_inline(match.group(2))) if match else None
def split_by_length(markdown: str, size: int = SPLIT_CHARS) -> list[dict]:
"""Ohne verwertbare Überschriften: an Absatzgrenzen in etwa gleich lange Stücke."""
sections, current = [], []
for block in md_blocks(markdown):
if current and sum(map(len, current)) + len(block) > size:
sections.append(current)
current = []
current.append(block)
if "".join(current).strip():
sections.append(current)
return [
{
"title": block_heading(blocks[0]) or f"Abschnitt {n}",
"body": "".join(blocks).strip() + "\n",
}
for n, blocks in enumerate(sections, start=1)
]
def split_markdown(markdown: str, mode: str, size: int = SPLIT_CHARS) -> list[dict]:
"""Buch in Abschnitte teilen: "teile" (oberste Ebene) oder "kapitel".
Titelei und Inhaltsverzeichnis bilden den Vorspann, Anhang/Autor/Impressum
nach dem letzten Kapitel den Nachspann. Ohne Gliederung: nach Länge (size).
"""
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, size)
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]
# Nachspann (Ebene 0): Anhang & Co. unterhalb der Schnittebenen, sonst hinge
# er am letzten Kapitel.
after = cuts[-1][0] if cuts else start - 1
back = next(
(
(i, 0, t) for i, lvl, t in headings
if i > after and lvl <= 3 and lvl not in cut_levels
and BACKMATTER_TITLE.match(md_unescape(clean_inline(t)))
),
None,
)
if back:
cuts.append(back)
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)
body = "\n".join(lines[index:end]).strip() + "\n"
if level == 0:
sections.append({"title": "Nachspann", "body": body})
continue
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": body,
})
return sections
def parse_page_ranges(spec: str, page_count: int) -> list[int]:
"""--page-ranges "1-12,13-40,41-" -> Startseiten (0-basiert), immer mit 0.
Seiten vor dem ersten, zwischen und nach den Bereichen werden eigene
Abschnitte; so geht kein Text verloren.
"""
ranges = []
for item in spec.replace(" ", "").split(","):
match = re.fullmatch(r"(\d+)(?:-(\d*))?", item)
if not match:
raise ValueError(f"Seitenbereich nicht lesbar: {item!r} (Beispiel: 1-12,13-40,41-)")
first = int(match.group(1))
last = page_count if match.group(2) == "" else int(match.group(2) or first)
if not 1 <= first <= last <= page_count:
raise ValueError(f"Seitenbereich {item!r} passt nicht zu {page_count} Seiten")
ranges.append((first - 1, last - 1))
ranges.sort()
for (_, last), (first, _) in zip(ranges, ranges[1:]):
if first <= last:
raise ValueError(f"Seitenbereiche überlappen sich: Seite {first + 1}")
starts = {0}
for first, last in ranges:
starts.add(first)
if last + 1 < page_count:
starts.add(last + 1)
return sorted(starts)
PAGE_KEY_MIN = 6 # kürzere Absätze ("Ja.") erben die Seite des Vorgängers
PAGE_SEARCH = 50_000 # so weit (Zeichen ohne Leerraum) wird vorausgesucht
def block_pages(blocks: list[str], page_texts: dict[int, str]) -> list[tuple[int, int]]:
"""Start- und Endseite (0-basiert) jedes Blocks.
Die Rohtexte aller Seiten bilden einen fortlaufenden Vergleichstext; Anfang und
Ende jedes Blocks werden darin nur vorwärts gesucht. Nicht gefundene Blöcke
(Bilder, stark veränderter Text) erben die Seite des Vorgängers.
"""
stream, offsets = [], []
length = 0
for pno in range(max(page_texts, default=-1) + 1):
offsets.append(length)
key = _page_key(page_texts.get(pno, ""))
stream.append(key)
length += len(key)
text = "".join(stream)
def page_of(position: int) -> int:
return bisect.bisect_right(offsets, position) - 1
result, pos, page = [], 0, 0
for block in blocks:
key = _page_key(block)
end = page
if len(key) >= PAGE_KEY_MIN:
found = text.find(key[:30], pos, pos + PAGE_SEARCH + 30)
if found >= 0:
pos, page = found, page_of(found)
tail = text.find(key[-30:], pos, pos + 2 * len(key) + 1000)
end = page_of(tail + len(key[-30:]) - 1) if tail >= 0 else page
result.append((page, max(page, end)))
return result
def split_by_pages(
markdown: str,
starts: list[int],
page_texts: dict[int, str],
align: bool = True,
) -> list[dict]:
"""Nach Seiten der PDF teilen; starts = Startseiten (0-basiert).
Geschnitten wird vor dem ersten Block, der auf der Startseite beginnt; ein
Absatz über die Seitengrenze bleibt ganz in der Datei davor. Überschriften
direkt vor dem Schnitt gehören zur neuen Datei. Mit align wird an einer
Überschrift höchstens eine Seite vor oder nach der Grenze geschnitten.
"""
blocks = md_blocks(markdown)
if not blocks:
return []
spans = block_pages(blocks, page_texts)
pages = [first for first, _ in spans]
heading = [block_heading(b) is not None for b in blocks]
cuts: list[int] = []
for start in starts:
if start == 0:
continue
cut = next((i for i, page in enumerate(pages) if page >= start), None)
if cut is None:
break
if align:
near = [
i for i in range(1, len(blocks))
if heading[i] and not heading[i - 1] and abs(pages[i] - start) <= 1
and (not cuts or i > cuts[-1])
]
if near:
cut = min(near, key=lambda i: (abs(pages[i] - start), abs(i - cut)))
while cut > 0 and heading[cut - 1]:
cut -= 1
if cut > (cuts[-1] if cuts else 0):
cuts.append(cut)
last_page = max(page_texts, default=0)
bounds = [0, *cuts, len(blocks)]
sections = []
for a, b in zip(bounds, bounds[1:]):
first = pages[a] if a else 0 # Titelei ab Seite 1, auch ohne Text (Cover)
# Ende: bis vor die nächste Startseite, mindestens bis zum Ende des letzten
# Absatzes (der über die Seitengrenze reichen kann).
end = max(
max(last for _, last in spans[a:b]),
pages[b] - 1 if b < len(blocks) else last_page,
)
name = f"Seite {first + 1}" if end == first else f"Seiten {first + 1}-{end + 1}"
sections.append({
"title": block_heading(blocks[a]) or name,
"body": "".join(blocks[a:b]).strip() + "\n",
"source_pages": [first + 1, end + 1],
})
return [s for s in sections if s["body"].strip()]
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,
cover: Path | None = None,
) -> None:
"""EPUB aus Markdown-Datei oder Split-Ordner; Angaben aus dem YAML-Kopf."""
meta, body, base = read_markdown_source(source)
if cover is None and meta.get("cover"):
cover = base / meta["cover"]
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)
def epubs_per_file(folder: Path, files: list[Path], meta: dict) -> Path:
"""--epub-per-file: ein EPUB je Einzeldatei im Unterordner "epub/"."""
target = folder / "epub"
target.mkdir(exist_ok=True)
cover = folder / meta["cover"] if meta.get("cover") else None
for file in files:
head, _ = split_frontmatter(file.read_text(encoding="utf-8"))
book = meta.get("title") or folder.name
markdown_to_epub(
file,
target / f"{file.stem}.epub",
title=f"{book} - {head.get('title') or file.stem}",
author=meta.get("author"),
language=meta.get("lang"),
cover=cover if cover and cover.is_file() else None,
)
return target
def rebase_image_links(markdown: str, old_base: Path, new_base: Path) -> str:
"""Relative Bildpfade von old_base auf new_base umrechnen."""
def fix(match: re.Match) -> str:
link = match.group(2)
if re.match(r"^[a-z][a-z0-9+.-]*:", link, re.IGNORECASE) or Path(link).is_absolute():
return match.group(0) # URL, data: oder absoluter Pfad
new = os.path.relpath(old_base / link, new_base)
return f"{match.group(1)}{Path(new).as_posix()})"
return IMAGE_LINK.sub(fix, markdown)
# ===========================================================================
# 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", "seiten"],
help="Teile, Kapitel oder Seitenbereiche der PDF als einzelne Markdown-Dateien "
"in einem Ordner (mit index.md für die Reihenfolge) statt einer Datei",
)
parser.add_argument(
"--pages-per-file",
type=int,
metavar="N",
help=f"mit --split seiten: etwa N PDF-Seiten je Datei, Schnitt an Absatzgrenzen, "
f"bevorzugt an einer Überschrift (Standard: {PAGES_PER_FILE})",
)
parser.add_argument(
"--page-ranges",
metavar="BEREICHE",
help="mit --split seiten: feste Bereiche der PDF-Seiten, z. B. 1-12,13-40,41- "
"(Seiten außerhalb werden eigene Dateien)",
)
parser.add_argument(
"--split-chars",
type=int,
default=SPLIT_CHARS,
metavar="N",
help="--split teile|kapitel ohne Gliederung: Abschnitte von etwa N Zeichen "
"(Standard: %(default)s)",
)
parser.add_argument(
"--epub-per-file",
action="store_true",
help="mit --split: zusätzlich ein EPUB je Einzeldatei im Unterordner epub/",
)
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-lines",
action="store_true",
help="jede Zeile der PDF bleibt eine Zeile (Gedichte, Liedtexte); kein "
"Zusammenfügen, auch keine Silbentrennung",
)
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.pages_per_file is not None or args.page_ranges) and args.split != "seiten":
parser.error("--pages-per-file und --page-ranges nur mit --split seiten.")
if args.pages_per_file is not None and args.page_ranges:
parser.error("--pages-per-file und --page-ranges schließen sich aus.")
if args.pages_per_file is not None and args.pages_per_file < 1:
parser.error("--pages-per-file muss mindestens 1 sein.")
if args.split_chars < 1000:
parser.error("--split-chars muss mindestens 1000 sein.")
if args.epub_per_file and not args.split:
parser.error("--epub-per-file nur mit --split.")
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:
epub_path = (args.epub_file or target.with_suffix(".epub")).expanduser()
if epub_path or args.epub_per_file:
args.images = True # EPUB ohne Bilder wäre unvollständig
# --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)
titles: dict[int, list[tuple[int, str]]] = {} # Seite -> Gliederungstitel
for level, title, page_no in doc.get_toc():
if page_no > 0 and title.strip() and not PAGE_LABEL.match(title.strip()):
titles.setdefault(page_no - 1, []).append((level, title.strip()))
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,
keep_lines=args.keep_lines,
titles=titles,
)
markdown, page_texts = 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 args.split == "seiten":
if args.page_ranges:
starts = parse_page_ranges(args.page_ranges, page_count)
else:
starts = list(range(0, page_count, args.pages_per_file or PAGES_PER_FILE))
sections = split_by_pages(
markdown, starts, page_texts, align=not args.page_ranges
)
files = write_split(folder, sections, meta)
elif folder:
sections = split_markdown(markdown, args.split, args.split_chars)
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)
if args.epub_per_file:
epub_dir = epubs_per_file(folder, files, meta)
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}")
if args.epub_per_file:
print(f"Einzel-EPUBs geschrieben: {len(files)} in {epub_dir}")
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}")
if args.split:
return split_from_markdown(args, parser, 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
def split_from_markdown(args, parser, source: Path) -> int:
"""--from-md DATEI --split teile|kapitel: Gesamt-Markdown in einen Ordner teilen."""
if args.split == "seiten":
parser.error("--split seiten braucht die PDF (Seitenzuordnung), nicht --from-md.")
if not source.is_file():
parser.error(f"--split mit --from-md braucht eine Markdown-Datei: {source}")
folder = source.with_suffix("")
try:
meta, body = split_frontmatter(source.read_text(encoding="utf-8"))
body = rebase_image_links(body, source.parent, folder)
if meta.get("cover"):
meta["cover"] = Path(os.path.relpath(source.parent / meta["cover"], folder)).as_posix()
meta["title"] = args.title or meta.get("title") or source.stem
meta["author"] = args.author or meta.get("author")
meta["lang"] = args.lang or meta.get("lang")
sections = split_markdown(body, args.split, args.split_chars)
if sections and sections[0]["title"] == "Abschnitt 1":
print("Hinweis: keine Gliederung erkannt - nach Länge geteilt.")
files = write_split(folder, sections, meta)
print(f"Markdown geschrieben: {len(files)} Dateien in {folder} (Reihenfolge: index.md)")
if args.epub or args.epub_file:
epub_path = (args.epub_file or folder.with_suffix(".epub")).expanduser()
markdown_to_epub(folder, epub_path)
print(f"EPUB geschrieben: {epub_path}")
if args.epub_per_file:
epub_dir = epubs_per_file(folder, files, meta)
print(f"Einzel-EPUBs geschrieben: {len(files)} in {epub_dir}")
except Exception as exc:
print(f"Fehler beim Aufteilen: {exc}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in a new issue View git blame Copy permalink