3ac32f4d34
Die Formelerkennung von pix2text (MFD) findet jede Formel auf der Seite, eingebettet oder abgesetzt. Sie wird zeichengenau aus dem Text geschnitten (gemischte Textstücke werden geteilt, Summen mit Grenzen an der richtigen Zeile verankert), als 300-dpi-Bild gespeichert und in <bilder>/formulas-*.json mit Seite, Rechteck, PDF-Text und "flach" vermerkt. Ergänzung über Mathematikschriften für übersehene Symbole. "Formel"-Boxen mit Text werden zu Text, solche mit Kurven zu Grafiken. Wörter neben hohen Formeln verlieren die falsche Hochstellung, Fußnotenzeichen bleiben. Für formula_img2latex, das die Formeln in LaTeX übersetzt. pix2text ist optional. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3273 lines
131 KiB
Python
3273 lines
131 KiB
Python
#!/usr/bin/env python3
|
|
"""PDF (digital oder gescannt) nach Markdown und optional EPUB konvertieren.
|
|
|
|
Ablauf:
|
|
1. Jede Seite wird klassifiziert: Textseite, Scan ohne Text, Scan mit
|
|
vorhandener OCR-Ebene oder leer.
|
|
2. Seitengruppen gehen mit passender OCR-Einstellung an pymupdf4llm
|
|
(optional parallel).
|
|
3. normalize_markdown() gliedert Fließtext nach Satzenden, setzt
|
|
Überschriften anhand der PDF-Gliederung, fügt Silbentrennungen
|
|
zusammen und entfernt Seitenzahlen und Kolumnentitel.
|
|
4. Optional erzeugt pandoc daraus ein EPUB.
|
|
"""
|
|
|
|
import argparse
|
|
import 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
|
|
|
|
|
|
# ===========================================================================
|
|
# Formeln seitenweise erkennen (--formulas)
|
|
# ===========================================================================
|
|
# pymupdf4llm schneidet nur Blöcke der Layout-Klasse "formula" als Bild aus. Darunter
|
|
# sind auch Merkkästen mit Text und Grafiken, und Formeln im Fließtext bleiben (oft
|
|
# verstümmelter) Text ("x4n" statt x^{4n}). Mit --formulas sucht die Formelerkennung
|
|
# von pix2text (MFD) jede Formel auf der Seite, eingebettet oder abgesetzt, auch in
|
|
# Kästen. Jede wird als Bild ausgeschnitten und in <bilder>/formulas-*.json mit Seite
|
|
# und Rechteck vermerkt; formula_img2latex übersetzt sie dann in LaTeX.
|
|
|
|
FORMULA_DPI = 150 # Seitenbild für die Formelerkennung
|
|
FORMULA_CROP_DPI = 300 # gespeicherter Ausschnitt (bleibt im Buch, solange keine Übersetzung bestätigt ist)
|
|
FORMULA_MIN_SCORE = 0.3 # schwächere Treffer bleiben Text
|
|
FORMULA_BOX_WORDS = 3 # so viele Wörter in Textschrift außerhalb der Treffer: "Formel"-Box ist ein Kasten
|
|
FORMULA_FIGURE_PATHS = 3 # so viele gekrümmte/schräge Linien: "Formel"-Box ist eine Grafik
|
|
FORMULA_FIGURE_WORDS = 40 # ... sofern sie nicht vor allem Text enthält
|
|
# Mathematikschriften (TeX und Office); ihre Buchstaben ("lim", "sin") sind keine Wörter
|
|
MATH_FONT = re.compile(r"CMMI|CMSY|CMEX|CMBSY|MSBM|MSAM|EUFM|EURM|EUSM|RSFS|Math|math|Symbol|SY\d|rsfs|"
|
|
r"txsy|txex|pxsy|pxex|txmi|pxmi|NewTXMI|NewPXMI|STIX.*Math|MTMI|MTSY|MTEX|LMMath")
|
|
|
|
FOOTNOTE_MARK = re.compile(r"[\d*\u2020\u2021\u00a7]+") # Fußnotenzeichen: Ziffern, *, Kreuze, Paragraph
|
|
|
|
_formulas: dict = {"on": False, "detector": None, "docs": {}, "entries": {}, "tokens": {}}
|
|
TEXT_LIKE_BOXES = ("text", "list-item", "footnote", "title", "section-header", "caption")
|
|
|
|
|
|
def _formula_detector():
|
|
if _formulas["detector"] is None:
|
|
from pix2text import MathFormulaDetector # optional und schwer: nur mit --formulas laden
|
|
|
|
_formulas["detector"] = MathFormulaDetector()
|
|
return _formulas["detector"]
|
|
|
|
|
|
def detect_formulas(page: pymupdf.Page) -> list[dict]:
|
|
"""Formeln auf der Seite: [{"rect": Rect in pt, "type": "embedding"|"isolated", "score": float}]."""
|
|
from PIL import Image
|
|
|
|
pix = page.get_pixmap(dpi=FORMULA_DPI)
|
|
img = Image.frombytes("RGB", (pix.width, pix.height), pix.samples)
|
|
f = 72.0 / FORMULA_DPI
|
|
hits = []
|
|
for hit in _formula_detector().detect(img):
|
|
if hit["score"] < FORMULA_MIN_SCORE:
|
|
continue
|
|
xs = [p[0] for p in hit["box"]]
|
|
ys = [p[1] for p in hit["box"]]
|
|
hits.append({"rect": pymupdf.Rect(min(xs) * f, min(ys) * f, max(xs) * f, max(ys) * f),
|
|
"type": hit["type"], "score": float(hit["score"])})
|
|
return hits
|
|
|
|
|
|
def _hit_for(bbox, hits: list[dict]) -> int | None:
|
|
"""Index des Treffers, in dem die Mitte von bbox liegt."""
|
|
x, y = (bbox[0] + bbox[2]) / 2, (bbox[1] + bbox[3]) / 2
|
|
for k, hit in enumerate(hits):
|
|
r = hit["rect"]
|
|
if r.x0 - 1 <= x <= r.x1 + 1 and r.y0 - 1 <= y <= r.y1 + 1:
|
|
return k
|
|
return None
|
|
|
|
|
|
def _text_words(textlines: list[dict], hits: list[dict]) -> int:
|
|
"""Wörter in Textschrift außerhalb der Formeltreffer."""
|
|
return sum(len(REAL_WORD.findall(span["text"]))
|
|
for line in textlines for span in line["spans"]
|
|
if not MATH_FONT.search(span.get("font", "")) and _hit_for(span["bbox"], hits) is None)
|
|
|
|
|
|
def _figure_paths(page: pymupdf.Page, rect: pymupdf.Rect) -> int:
|
|
"""Gekrümmte oder schräge Linien in rect: Kurven, Achsen mit Pfeilen. Bruch- und Wurzelstriche
|
|
sind waagerecht, Rahmen und Hinterlegungen Rechtecke."""
|
|
n = 0
|
|
for path in page.get_drawings():
|
|
if not rect.contains(path["rect"]):
|
|
continue
|
|
for item in path["items"]:
|
|
if item[0] == "c" or (item[0] == "l" and abs(item[1].y - item[2].y) > 0.5 and abs(item[1].x - item[2].x) > 0.5):
|
|
n += 1
|
|
break
|
|
return n
|
|
|
|
|
|
def _formula_token(page: pymupdf.Page, hits: list[dict], k: int, image_path: str, filename: str,
|
|
chars: list | None = None) -> str:
|
|
"""Platzhalter für Treffer k; schneidet ihn beim ersten Mal aus und vermerkt ihn. Nur Buchstaben und
|
|
Ziffern: übersteht jede Textausgabe und Maskierung, finish_formula_links macht das Bild daraus."""
|
|
token = f"FORMELBILD{page.number + 1:04d}N{k:03d}"
|
|
if token not in _formulas["tokens"]:
|
|
hit = hits[k]
|
|
rect = hit["rect"] + (-1, -1, 1, 1)
|
|
name = f"{filename}-{page.number + 1:04d}-f{k:03d}.png"
|
|
md_name, save_name = document_layout.utils.md_path(image_path, name)
|
|
page.get_pixmap(dpi=FORMULA_CROP_DPI, clip=rect).save(save_name)
|
|
text, flat = _formula_text(hit["rect"], chars or [])
|
|
_formulas["entries"][Path(save_name).name] = {"page": page.number + 1, "rect": [round(v, 2) for v in rect],
|
|
"type": hit["type"], "score": round(hit["score"], 3), "text": text, "flat": flat}
|
|
_formulas["tokens"][token] = (md_name, hit["type"] == "isolated")
|
|
return token
|
|
|
|
|
|
def _page_chars(page: pymupdf.Page) -> list[tuple[pymupdf.Rect, str, float, float]]:
|
|
"""Alle Zeichen der Seite mit Position, Grundlinie und Größe (die Textstücke der PDF mischen Text und
|
|
Formel: "if 0")."""
|
|
chars = []
|
|
for block in page.get_text("rawdict")["blocks"]:
|
|
for line in block.get("lines", []):
|
|
for span in line["spans"]:
|
|
chars.extend((pymupdf.Rect(c["bbox"]), c["c"], c["origin"][1], span["size"]) for c in span["chars"])
|
|
return chars
|
|
|
|
|
|
def _formula_text(rect: pymupdf.Rect, chars: list) -> tuple[str, bool]:
|
|
"""Text der PDF-Textebene im Formelrechteck und ob er "flach" ist: alle Zeichen auf einer Grundlinie in
|
|
gleicher Größe, also ohne Hoch-/Tiefstellung, Bruch oder Grenzen, die die Textebene verlieren würde."""
|
|
inside = [ch for ch in chars if rect.contains(pymupdf.Point((ch[0].x0 + ch[0].x1) / 2, (ch[0].y0 + ch[0].y1) / 2))]
|
|
inside.sort(key=lambda ch: (round(ch[2]), ch[0].x0))
|
|
text = "".join(ch[1] for ch in inside).strip()
|
|
flat = bool(inside) and max(ch[2] for ch in inside) - min(ch[2] for ch in inside) <= 0.6 \
|
|
and max(ch[3] for ch in inside) - min(ch[3] for ch in inside) <= 0.2
|
|
return text, flat
|
|
|
|
|
|
def _span_chars(span: dict, chars: list[tuple[pymupdf.Rect, str]]) -> list[tuple[pymupdf.Rect, str]]:
|
|
"""Die Zeichen eines Textstücks mit Position; sonst gleichmäßig über seine Breite verteilt."""
|
|
box = pymupdf.Rect(span["bbox"]) + (-0.5, -0.5, 0.5, 0.5)
|
|
inside = [(ch[0], ch[1]) for ch in chars if box.contains(pymupdf.Point((ch[0].x0 + ch[0].x1) / 2,
|
|
(ch[0].y0 + ch[0].y1) / 2))]
|
|
inside.sort(key=lambda rc: rc[0].x0)
|
|
if "".join(c for _, c in inside) == span["text"]:
|
|
return inside
|
|
x0, y0, x1, y1 = span["bbox"]
|
|
n = max(1, len(span["text"]))
|
|
w = (x1 - x0) / n
|
|
return [(pymupdf.Rect(x0 + i * w, y0, x0 + (i + 1) * w, y1), c) for i, c in enumerate(span["text"])]
|
|
|
|
|
|
def _replace_formula_spans(page, textlines: list[dict], hits: list[dict], chars, image_path: str, filename: str
|
|
) -> list[dict]:
|
|
"""Zeichen innerhalb eines Treffers durch einen Platzhalter ersetzen (einen je Treffer und Seite);
|
|
gemischte Textstücke werden dabei geteilt. Eine Formel über mehrere Textzeilen (Summe mit Grenzen)
|
|
steht in der Zeile, deren Mitte ihrer Mitte am nächsten liegt."""
|
|
split: list[list[tuple[dict, list]]] = [] # je Zeile: (Textstück, [[treffer oder None, zeichen]])
|
|
anchor: dict[int, tuple[float, int]] = {} # treffer -> (abstand, zeile)
|
|
for n, line in enumerate(textlines):
|
|
parts = []
|
|
for span in line["spans"]:
|
|
if not any(pymupdf.Rect(span["bbox"]).intersects(h["rect"]) for h in hits):
|
|
parts.append((span, [[None, None]]))
|
|
continue
|
|
groups: list[list] = []
|
|
for rect, c in _span_chars(span, chars):
|
|
k = _hit_for(tuple(rect), hits)
|
|
if groups and groups[-1][0] == k:
|
|
groups[-1][1].append((rect, c))
|
|
else:
|
|
groups.append([k, [(rect, c)]])
|
|
for k, _ in groups:
|
|
if k is not None:
|
|
mid = (line["bbox"][1] + line["bbox"][3]) / 2
|
|
dist = abs(mid - (hits[k]["rect"].y0 + hits[k]["rect"].y1) / 2)
|
|
if k not in anchor or dist < anchor[k][0]:
|
|
anchor[k] = (dist, n)
|
|
parts.append((span, groups))
|
|
split.append(parts)
|
|
out = []
|
|
for n, (line, parts) in enumerate(zip(textlines, split)):
|
|
spans = []
|
|
has_formula = any(k is not None for _, groups in parts for k, _ in groups)
|
|
for span, groups in parts:
|
|
if groups == [[None, None]]:
|
|
if has_formula and span["flags"] & pymupdf.TEXT_FONT_SUPERSCRIPT and not FOOTNOTE_MARK.fullmatch(span["text"].strip()):
|
|
span = dict(span, flags=span["flags"] & ~pymupdf.TEXT_FONT_SUPERSCRIPT) # Wort neben hoher Formel
|
|
spans.append(span)
|
|
continue
|
|
for k, part in groups:
|
|
if k is None:
|
|
text = "".join(c for _, c in part)
|
|
flags = span["flags"]
|
|
if not FOOTNOTE_MARK.fullmatch(text.strip()): # neben hoher Formel nicht hochgestellt (Fußnote: "22")
|
|
flags &= ~pymupdf.TEXT_FONT_SUPERSCRIPT
|
|
spans.append(dict(span, text=text, flags=flags, bbox=tuple(pymupdf.Rect(part[0][0]) | part[-1][0])))
|
|
elif anchor[k][1] == n and k not in _formulas["page_done"]:
|
|
_formulas["page_done"].add(k)
|
|
spans.append(dict(span, text=_formula_token(page, hits, k, image_path, filename, chars), flags=0,
|
|
char_flags=0, font="Formula", bbox=tuple(hits[k]["rect"])))
|
|
if any(sp["text"].strip() for sp in spans):
|
|
out.append(dict(line, spans=spans))
|
|
return out
|
|
|
|
|
|
def _font_hits(textlines: list[dict], hits: list[dict]) -> list[dict]:
|
|
"""Formeln in Mathematikschrift, die die Formelerkennung übersehen hat (einzelnes θ, π, "n ≥ m"):
|
|
Textstücke in Mathematikschrift samt angrenzender kurzer kursiver Buchstaben und Ziffern."""
|
|
found = []
|
|
for line in textlines:
|
|
spans = line["spans"]
|
|
math = [bool(MATH_FONT.search(sp.get("font", ""))) and sp["text"].strip() != ""
|
|
and _hit_for(sp["bbox"], hits) is None for sp in spans]
|
|
i = 0
|
|
while i < len(spans):
|
|
if not math[i]:
|
|
i += 1
|
|
continue
|
|
a = b = i
|
|
while a > 0 and _short_math_neighbor(spans[a - 1], hits):
|
|
a -= 1
|
|
while b + 1 < len(spans) and (math[b + 1] or _short_math_neighbor(spans[b + 1], hits)):
|
|
b += 1
|
|
rect = pymupdf.Rect(spans[a]["bbox"])
|
|
for sp in spans[a:b + 1]:
|
|
rect |= pymupdf.Rect(sp["bbox"])
|
|
found.append({"rect": rect, "type": "embedding", "score": 0.0})
|
|
i = b + 1
|
|
return found
|
|
|
|
|
|
def _short_math_neighbor(span: dict, hits: list[dict]) -> bool:
|
|
"""Kursiver Einzelbuchstabe, Ziffern oder Operator neben Mathematikschrift ("n" in "n ≥ m")."""
|
|
text = span["text"].strip()
|
|
return (_hit_for(span["bbox"], hits) is None and 0 < len(text) <= 3 and not REAL_WORD.search(text)
|
|
and (bool(span["flags"] & pymupdf.TEXT_FONT_ITALIC) or bool(re.fullmatch(r"[\d.,+\-=<>()]+", text))))
|
|
|
|
|
|
def _prepare_formula_page(page: pymupdf.Page, layout, image_path: str, filename: str) -> None:
|
|
"""Formeln der Seite finden; Boxen der Klasse "formula" neu einordnen (Kasten mit Text -> Text,
|
|
Grafik -> Bild, sonst abgesetzte Formel) und in allen Textboxen die Formeln durch Platzhalter ersetzen."""
|
|
hits = detect_formulas(page)
|
|
_formulas["page_done"] = set()
|
|
chars = _page_chars(page) if hits else []
|
|
for box in layout.boxes:
|
|
if box.boxclass == "formula":
|
|
rect = pymupdf.Rect(box.x0, box.y0, box.x1, box.y1)
|
|
lines = [{"bbox": ln[0], "spans": ln[1]}
|
|
for ln in document_layout.get_raw_lines(textpage=None, blocks=layout.fulltext, clip=rect,
|
|
ignore_invisible=False)]
|
|
words = _text_words(lines, hits)
|
|
if _figure_paths(page, rect) >= FORMULA_FIGURE_PATHS and words < FORMULA_FIGURE_WORDS:
|
|
box.boxclass = "picture"
|
|
elif words >= FORMULA_BOX_WORDS:
|
|
if isinstance(box.image, str): # pymupdf4llm hat den ganzen Kasten schon als Bild gespeichert
|
|
Path(box.image).unlink(missing_ok=True)
|
|
box.boxclass, box.image, box.textlines = "text", None, lines
|
|
if box.boxclass in TEXT_LIKE_BOXES and box.textlines:
|
|
extra = _font_hits(box.textlines, hits)
|
|
if extra and not chars:
|
|
chars = _page_chars(page)
|
|
hits.extend(extra)
|
|
if hits:
|
|
box.textlines = _replace_formula_spans(page, box.textlines, hits, chars, image_path, filename)
|
|
|
|
|
|
def patch_pymupdf4llm_formulas() -> None:
|
|
parse = document_layout.parse_document
|
|
if getattr(parse, "pdf2md_formulas", False):
|
|
return
|
|
|
|
def parse_document(doc, *args, **kwargs):
|
|
parsed = parse(doc, *args, **kwargs)
|
|
if _formulas["on"]:
|
|
if not isinstance(doc, pymupdf.Document):
|
|
doc = _formulas["docs"].setdefault(str(doc), pymupdf.open(doc))
|
|
for layout in parsed.pages:
|
|
_prepare_formula_page(doc[layout.page_number - 1], layout, parsed.image_path, parsed.filename)
|
|
return parsed
|
|
|
|
parse_document.pdf2md_formulas = True
|
|
document_layout.parse_document = parse_document
|
|
|
|
|
|
FORMULA_TOKEN = re.compile(r"[ \t]*FORMELBILD\d{4}N\d{3}[ \t]*")
|
|
|
|
|
|
def finish_formula_links(markdown: str) -> str:
|
|
"""Platzhalter -> Bildverweis; abgesetzte Formeln aus Textblöcken (Merkkästen) in einen eigenen Absatz."""
|
|
def link(m: re.Match) -> str:
|
|
token = m.group(0).strip()
|
|
md_name, isolated = _formulas["tokens"].get(token, (None, False))
|
|
if md_name is None:
|
|
return m.group(0)
|
|
return f"\n\n\n\n" if isolated else m.group(0).replace(token, f"")
|
|
return FORMULA_TOKEN.sub(link, markdown)
|
|
|
|
|
|
def write_formula_list(image_path: str, first_page: int) -> None:
|
|
"""Die Formeln dieses Arbeitsprozesses als <bilder>/formulas-<seite>.json."""
|
|
if _formulas["entries"]:
|
|
out = Path(image_path) / f"formulas-{first_page + 1:04d}.json"
|
|
out.write_text(json.dumps(_formulas["entries"], indent=1), encoding="utf-8")
|
|
_formulas["entries"] = {}
|
|
|
|
|
|
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()
|
|
patch_pymupdf4llm_formulas()
|
|
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)
|
|
_formulas["on"] = bool(options.get("formulas"))
|
|
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)
|
|
|
|
if _formulas["on"]:
|
|
texts = _convert_pages(source, run, pages, options, kwargs)
|
|
write_formula_list(options["image_path"], pages[0])
|
|
return [finish_formula_links(t) for t in texts]
|
|
return _convert_pages(source, run, pages, options, kwargs)
|
|
|
|
|
|
def _convert_pages(source: str, run: str, pages: list[int], options: dict, kwargs: dict) -> list[str]:
|
|
# 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",
|
|
"--mathml", # $$...$$ (e.g. from formula_img2latex) as MathML, not as TeX text
|
|
"--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(
|
|
"--formulas",
|
|
action="store_true",
|
|
help="Formeln seitenweise erkennen (pix2text), auch im Fließtext und in Kästen, und als "
|
|
"Bilder ausschneiden - für formula_img2latex (schaltet --images ein)",
|
|
)
|
|
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 or args.formulas:
|
|
args.images = True # EPUB ohne Bilder wäre unvollständig; Formeln sind Bilder
|
|
|
|
# --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,
|
|
formulas=args.formulas,
|
|
)
|
|
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())
|