"""Orchestration of the complete EPUB -> Markdown conversion.""" from __future__ import annotations import os import posixpath import re import shutil import sys import tempfile from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any from . import TOOL_NAME, __version__ from .chapters import ChapterBuilder from .diagnostics import Diagnostics from .epub_reader import EpubReader, parse_xml from .errors import ContainerError, InputError, StructureError from .footnotes import collect_notes from .html_converter import RenderOptions, Renderer from .linkcheck import check_links from .media import MediaRegistry from .models import Chapter from .normalize import Doc, book_uses_title_tables, is_html_toc_doc, make_doc, normalize_doc, remove_keep_tail from .references import LinkResolver from .utils import is_inside, iter_elements, sha256_file, slugify from . import writer MARKER_FILES = ("conversion-report.json", "manifest.json") LARGE_CHAPTER_WORDS = 30000 # above this a deeper --chapter-level is suggested (RAG/readability) @dataclass class Options: format: str = "gfm" extract_media: bool = True copy_original: bool = False chapter_level: int = 1 toc: bool = True frontmatter: bool = True footnotes: str = "end-of-chapter" math: str = "auto" images: str = "relative" slug_language: str = "de" keep_html: bool = False code_language: str = "" keep_wrap_markers: bool = False strict: bool = False verbose: bool = False dry_run: bool = False force: bool = False report: str | None = None command: list[str] = field(default_factory=list) def public(self) -> dict[str, Any]: data = asdict(self) data.pop("command") return data @dataclass class Result: report: dict[str, Any] chapters: list[Chapter] files: dict[str, str] output_dir: Path def check_paths(input_path: Path, output_dir: Path, opts: Options) -> None: if not input_path.is_file(): raise InputError(f"Eingabedatei nicht gefunden oder keine Datei: {input_path}") if not os.access(input_path, os.R_OK): raise InputError(f"Eingabedatei nicht lesbar: {input_path}") if not 1 <= opts.chapter_level <= 6: raise InputError("--chapter-level muss zwischen 1 und 6 liegen.") if opts.dry_run: return if output_dir.exists() and not output_dir.is_dir(): raise InputError(f"Ausgabepfad existiert und ist kein Verzeichnis: {output_dir}") if output_dir.exists() and any(output_dir.iterdir()): if not opts.force: raise InputError( f"Ausgabeverzeichnis existiert bereits und ist nicht leer: {output_dir}\n" "Mit --force wird es ersetzt (nur wenn es eine frühere epub2md-Ausgabe ist)." ) resolved = output_dir.resolve() protected = {Path("/").resolve(), Path.home().resolve(), Path.cwd().resolve()} if resolved in protected or is_inside(Path.cwd(), resolved): raise InputError(f"--force verweigert: {output_dir} ist ein geschütztes Verzeichnis.") if is_inside(input_path, output_dir): raise InputError("--force verweigert: die Eingabedatei liegt im Ausgabeverzeichnis.") if not any((output_dir / m).is_file() for m in MARKER_FILES): raise InputError( f"--force verweigert: {output_dir} enthält keine frühere epub2md-Ausgabe " "(conversion-report.json/manifest.json fehlen). Bitte manuell prüfen." ) def convert(input_path: Path, output_dir: Path, opts: Options) -> Result: diag = Diagnostics(verbose=opts.verbose, strict=opts.strict) check_paths(input_path, output_dir, opts) generated_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat() sha = sha256_file(input_path) diag.step(f"Öffne {input_path}") reader = EpubReader(input_path, diag) try: return _convert(reader, input_path, output_dir, opts, diag, generated_at, sha) finally: reader.close() def _load_docs(reader: EpubReader, package: Any, diag: Diagnostics) -> list[Doc]: docs: list[Doc] = [] for item in package.spine: try: root = parse_xml(reader.read(item.path), item.path, diag, html_fallback=True) except StructureError as exc: diag.error("xhtml-unreadable", str(exc), item.path, critical=True) continue except KeyError: diag.error("spine-file-missing", f"Spine-Dokument fehlt: {item.path}", item.path, critical=True) continue docs.append(make_doc(item.path, root, item.linear)) if not docs: raise StructureError("Keines der Spine-Dokumente konnte gelesen werden.") return docs def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Options, diag: Diagnostics, generated_at: str, sha: str) -> Result: package = reader.load_package() md = package.metadata diag.step(f"EPUB {package.version}: {md.title!r}, {len(package.spine)} Spine-Dokumente") docs = _load_docs(reader, package, diag) # --- normalization and HTML-TOC handling title_tables = book_uses_title_tables(docs) for doc in docs: normalize_doc(doc, diag, title_tables) landmark_toc_docs = {lm.path for lm in package.landmarks if lm.type == "toc" and lm.path and not lm.fragment} for lm in package.landmarks: if lm.type == "toc" and lm.fragment: toc_doc = next((d for d in docs if d.path == lm.path), None) el = next((e for e in iter_elements(toc_doc.body) if e.get("id") == lm.fragment), None) if toc_doc else None if toc_doc is not None and el is not None and el is not toc_doc.body and _mostly_links(el): toc_doc.toc_ids.update(e.get("id") for e in iter_elements(el) if e.get("id")) remove_keep_tail(el) if package.toc: for doc in docs: if is_html_toc_doc(doc, package.nav_path, landmark_toc_docs): doc.skipped = "html-toc" diag.info("html-toc-skipped", "HTML-Inhaltsverzeichnis übersprungen (ersetzt durch README.md/NAV).", doc.path) if all(d.skipped for d in docs): for d in docs: d.skipped = "" # --- notes, chapters, links notes = collect_notes(docs, opts.footnotes, diag) diag.step(f"{len(notes.notes)} Anmerkungen erkannt") chapters = ChapterBuilder(opts.chapter_level, package.toc, diag).build(docs) _assign_filenames(chapters, opts.slug_language) readme_path = "README.md" if opts.toc else None resolver = LinkResolver(docs, chapters, notes, readme_path) nav_targets: list[tuple[str, str]] = [] def walk(entries: list[Any]) -> None: for e in entries: if e.path and e.fragment: nav_targets.append((e.path, e.fragment)) walk(e.children) walk(package.toc) resolver.add_targets(nav_targets) diag.step(f"{len(chapters)} Kapitel gebildet") # --- rendering media = MediaRegistry(package, opts.images, opts.slug_language) cover_out = "" if md.cover_path and opts.extract_media and reader.exists(md.cover_path): cover_out = media.register(md.cover_path, role="cover", hint="cover").output language = (md.language or "de").lower() renderer = Renderer( RenderOptions(format=opts.format, footnotes=opts.footnotes, math=opts.math, extract_media=opts.extract_media, keep_html=opts.keep_html, language=language, code_language=opts.code_language, keep_wrap_markers=opts.keep_wrap_markers), resolver, media, notes, diag, set(reader.names), ) width = max(3, len(str(len(chapters) + 1))) notes_title = "Anmerkungen" if language.startswith("de") else "Notes" endnotes_number = 999 if len(chapters) < 999 else len(chapters) + 1 renderer.endnotes_file = f"chapters/{endnotes_number:0{width}d}-{slugify(notes_title, opts.slug_language)}.md" for ch in chapters: diag.step(f"Rendere {ch.filename}") ch.markdown = renderer.render_chapter(ch) ch.stats.word_count = _word_count(ch.markdown) _check_chapter_size(ch, opts.chapter_level, diag) wraps: dict[str, int] = {} for block in renderer.code_blocks: if block["wrap_lines_joined"]: wraps[block["chapter"]] = wraps.get(block["chapter"], 0) + block["wrap_lines_joined"] for chapter_file, joined in wraps.items(): diag.info("wrap-markers-joined", f"{joined} Druck-Zeilenumbrüche (↪) in Codeblöcken zusammengeführt.", chapter_file, chapter_file) endnotes: Chapter | None = None endnotes_md = "" if opts.footnotes == "end-of-book" and renderer.book_notes: endnotes_md = renderer.render_endnotes_file(notes_title) endnotes = Chapter(number=len(chapters) + 1, title=notes_title, items=[], docs=[], types=["endnotes"]) endnotes.filename = renderer.endnotes_file endnotes.stats.footnotes = len(renderer.book_notes) endnotes.stats.word_count = _word_count(endnotes_md) # --- remaining media (fonts, css, audio, unreferenced images) skipped_formula_images = 0 if opts.extract_media: for item in package.manifest.values(): if item.path in media.files or not reader.exists(item.path): continue if item.media_type in ("application/xhtml+xml", "text/html", "application/x-dtbncx+xml", "application/oebps-package+xml") or item.path == package.opf_path: continue if item.path in renderer.consumed_formula_images: skipped_formula_images += 1 continue mf = media.register(item.path, role="unreferenced") mf.references = 0 for mf in media.files.values(): if mf.data is None and reader.exists(mf.source): mf.size = reader.names[mf.source].file_size # --- assemble output files in memory unresolved_notes = [n for n in notes.notes.values() if n.ref_count and not n.label and notes.extracted] for n in unresolved_notes: diag.warn("note-not-rendered", "Anmerkung wurde referenziert, aber nicht ausgegeben.", n.key) opf_dir = posixpath.dirname(package.opf_path) files: dict[str, str] = {} for ch in chapters: warn_count = sum(1 for d in diag.items if d.chapter == ch.filename and d.level != "info") ch.warnings = warn_count front = writer.chapter_frontmatter(ch, md, input_path.name, opf_dir, warn_count) if opts.frontmatter else "" files[ch.filename] = front + ch.markdown if endnotes is not None: front = writer.chapter_frontmatter(endnotes, md, input_path.name, opf_dir, 0) if opts.frontmatter else "" files[endnotes.filename] = front + endnotes_md all_chapters = chapters + ([endnotes] if endnotes else []) stats = _book_stats(all_chapters, media, renderer, diag) if opts.toc: files["README.md"] = writer.readme(md, chapters, endnotes, package.toc, resolver, stats, cover_out, package.version, input_path.name, opts.command) existing = {mf.output for mf in media.files.values()} if opts.extract_media else set() if opts.copy_original: existing.add(f"source/{input_path.name}") existing.update({"metadata.yaml", "manifest.json", "conversion-report.json", "diagnostics/warnings.md", "diagnostics/unsupported-elements.md"}) broken = check_links(files, existing) for b in broken: diag.warn("broken-output-link", f"Link {b['link']} in der Ausgabe ist ungültig ({b['reason']}).", b["file"], b["file"]) stats["warnings"] = diag.count("warning") report = _report(input_path, sha, generated_at, package, opts, all_chapters, media, renderer, notes, diag, docs, broken, skipped_formula_images) files["metadata.yaml"] = writer.metadata_yaml(md, input_path.name, sha, package.version, cover_out, opts.public(), generated_at) files["manifest.json"] = writer.to_json(_manifest(md, all_chapters, media, diag)) files["conversion-report.json"] = writer.to_json(report) files["diagnostics/warnings.md"] = writer.warnings_md(diag) files["diagnostics/unsupported-elements.md"] = writer.unsupported_md(diag) if opts.dry_run: return Result(report=report, chapters=all_chapters, files=files, output_dir=output_dir) _write_output(reader, input_path, output_dir, opts, files, media) if opts.report: report_path = Path(opts.report) report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(writer.to_json(report), encoding="utf-8") return Result(report=report, chapters=all_chapters, files=files, output_dir=output_dir) def _mostly_links(el: Any) -> bool: from .utils import local, norm_ws, text_content total = len(norm_ws(text_content(el))) links = sum(len(norm_ws(text_content(a))) for a in iter_elements(el) if local(a.tag) == "a") return total > 0 and links / total >= 0.5 def _assign_filenames(chapters: list[Chapter], slug_language: str) -> None: width = max(3, len(str(len(chapters) + 1))) used: set[str] = set() for ch in chapters: base = slugify(ch.title, slug_language, fallback=(ch.types[0] if ch.types else "kapitel")) name = f"{ch.number:0{width}d}-{base}" candidate, n = name, 2 while candidate in used: candidate = f"{name}-{n}" n += 1 used.add(candidate) ch.filename = f"chapters/{candidate}.md" def _check_chapter_size(ch: Chapter, chapter_level: int, diag: Diagnostics) -> None: """Suggest a deeper --chapter-level for very large chapters with enough sub-headings.""" if ch.stats.word_count <= LARGE_CHAPTER_WORDS or chapter_level >= 6: return sections = len(re.findall(r"^## ", ch.markdown, re.M)) if sections >= 2: diag.warn( "chapter-large", f"Kapitel „{ch.title}“ hat {ch.stats.word_count} Wörter und {sections} Unterabschnitte; " f"--chapter-level {chapter_level + 1} würde sie als eigene Dateien ausgeben.", ch.docs[0] if ch.docs else "", ch.filename, ) def _word_count(markdown: str) -> int: text = re.sub(r"^(`{3,}).*?^\1\s*$", " ", markdown, flags=re.M | re.S) text = re.sub(r"<[^>]+>|\]\([^)]*\)", " ", text) return len(re.findall(r"\w+", text)) def _book_stats(chapters: list[Chapter], media: MediaRegistry, renderer: Renderer, diag: Diagnostics) -> dict[str, int]: def total(attr: str) -> int: return sum(getattr(c.stats, attr) for c in chapters) return { "chapters": len(chapters), "images": total("images"), "svgs": total("svgs"), "media_files": len(media.files), "code_blocks": total("code_blocks"), "code_blocks_unlabeled": total("code_blocks_unlabeled"), "formulas": len(renderer.formulas), "formulas_as_image": total("math_as_image"), "formulas_as_mathml": total("math_as_mathml"), "footnotes": len(renderer.book_notes), "tables": total("tables"), "tables_html": total("tables_html"), "warnings": diag.count("warning"), } def _manifest(md: Any, chapters: list[Chapter], media: MediaRegistry, diag: Diagnostics) -> dict[str, Any]: return { "book": {"title": md.title, "subtitle": md.subtitle, "authors": md.authors, "language": md.language, "identifier": md.identifier, "isbn": md.isbn, "publisher": md.publisher}, "chapters": [ { "number": c.number, "title": c.title, "file": c.filename, "source_href": c.docs[0] if c.docs else None, "source_hrefs": c.docs, "types": c.types, "anchors": c.anchors, "word_count": c.stats.word_count, "code_blocks": c.stats.code_blocks, "images": c.stats.images + c.stats.svgs, "math_blocks": c.stats.math_block, "math_inline": c.stats.math_inline, "footnotes": c.stats.footnotes, "tables": c.stats.tables, "warnings": c.warnings, } for c in chapters ], "media": [m.as_dict() for m in media.files.values()], "warnings": [d.as_dict() for d in diag.items if d.level != "info"], } def _report(input_path: Path, sha: str, generated_at: str, package: Any, opts: Options, chapters: list[Chapter], media: MediaRegistry, renderer: Renderer, notes: Any, diag: Diagnostics, docs: list[Doc], broken: list[dict[str, str]], skipped_formula_images: int) -> dict[str, Any]: from collections import Counter reps = Counter(f["representation"] for f in renderer.formulas) origins = Counter(f["origin"] for f in renderer.formulas) langs = Counter(c["language"] or "(keine)" for c in renderer.code_blocks) special = [{"type": t, "title": c.title, "file": c.filename} for c in chapters for t in c.types if t in ("glossary", "bibliography", "index", "index-persons", "endnotes", "appendix")] errors = [d.as_dict() for d in diag.items if d.level == "error"] warnings = [d.as_dict() for d in diag.items if d.level == "warning"] status = "success" if not errors and not warnings else ("success-with-errors" if errors else "success-with-warnings") return { "tool": TOOL_NAME, "version": __version__, "generated_at": generated_at, "status": status, "lossless": False if (warnings or errors) else None, "input": {"file": str(input_path), "name": input_path.name, "sha256": sha, "size": input_path.stat().st_size}, "epub": {"version": package.version, "opf": package.opf_path, "nav": package.nav_path, "ncx": package.ncx_path, "spine_items": len(package.spine), "manifest_items": len(package.manifest)}, "options": opts.public(), "metadata": asdict(package.metadata), "chapters": [ {"number": c.number, "title": c.title, "file": c.filename, "source_hrefs": c.docs, "types": c.types, "stats": c.stats.as_dict(), "warnings": c.warnings} for c in chapters ], "skipped_documents": [{"href": d.path, "reason": d.skipped} for d in docs if d.skipped], "media": {"count": len(media.files), "files": [m.as_dict() for m in media.files.values()], "formula_images_replaced_by_tex": skipped_formula_images}, "code_blocks": {"count": len(renderer.code_blocks), "languages": dict(langs), "items": renderer.code_blocks}, "formulas": {"count": len(renderer.formulas), "representations": dict(reps), "origins": dict(origins), "items": renderer.formulas}, "footnotes": {"mode": opts.footnotes, "detected": len(notes.notes), "referenced": sum(1 for n in notes.notes.values() if n.ref_count), "rendered": len(renderer.book_notes)}, "special_sections": special, "warnings": warnings, "errors": errors, "infos": [d.as_dict() for d in diag.items if d.level == "info"], "unresolved_links": diag.unresolved_links, "unsupported_elements": dict(diag.unsupported), "output_link_check": {"broken": broken, "broken_count": len(broken)}, } def _write_output(reader: EpubReader, input_path: Path, output_dir: Path, opts: Options, files: dict[str, str], media: MediaRegistry) -> None: parent = output_dir.resolve().parent try: parent.mkdir(parents=True, exist_ok=True) tmp = Path(tempfile.mkdtemp(prefix=".epub2md-", dir=parent)) except OSError as exc: raise InputError(f"Ausgabeverzeichnis nicht beschreibbar: {parent} ({exc.strerror or exc})") from exc try: for rel, content in files.items(): _write_file(tmp, rel, content.encode("utf-8")) if opts.extract_media: for mf in media.files.values(): data = mf.data if mf.data is not None else reader.read(mf.source) _write_file(tmp, mf.output, data) if opts.copy_original: dest = tmp / "source" / input_path.name dest.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(input_path, dest) if output_dir.exists(): if any(output_dir.iterdir()): shutil.rmtree(output_dir) # safety checks were done in check_paths() else: output_dir.rmdir() os.replace(tmp, output_dir) except OSError as exc: shutil.rmtree(tmp, ignore_errors=True) raise InputError(f"Schreibfehler im Ziel {output_dir}: {exc.strerror or exc}") from exc except BaseException: shutil.rmtree(tmp, ignore_errors=True) raise def _write_file(root: Path, rel: str, data: bytes) -> None: target = (root / rel).resolve() if not is_inside(target, root): raise ContainerError(f"Ausgabepfad außerhalb des Zielverzeichnisses verweigert: {rel}") target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(data) def print_summary(result: Result, opts: Options) -> None: r = result.report n_warn = len(r["warnings"]) n_err = len(r["errors"]) action = "Analyse (--dry-run, nichts geschrieben)" if opts.dry_run else f"Ausgabe: {result.output_dir}" print(f"{TOOL_NAME}: {len(r['chapters'])} Kapitel, {r['media']['count']} Mediendateien, " f"{r['code_blocks']['count']} Codeblöcke, {r['formulas']['count']} Formeln, " f"{r['footnotes']['rendered']} Anmerkungen, {n_warn} Warnungen, {n_err} Fehler") print(action) if opts.dry_run: for c in r["chapters"]: print(f" {c['file']} ← {', '.join(c['source_hrefs'][:2])}{' …' if len(c['source_hrefs']) > 2 else ''}") if opts.report: print("Hinweis: --report wird bei --dry-run nicht geschrieben.", file=sys.stderr)