dschlueter/epub2md
Watch
1
0
Fork
You've already forked epub2md
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

refactor: pipeline._convert in Teilschritte zerlegt (QA R5)

Browse source 
_convert (zuletzt 153 Zeilen) orchestriert jetzt nur noch die Teilschritte
_prepare_docs, _build_structure, _render (mit _report_wrap_markers und
_collect_remaining_media) und _assemble. Längste Funktion im Modul: 48 Zeilen.
Keine Verhaltensänderung: Die Ausgabe aller vier Beispielbücher (Standard
und end-of-book) ist bis auf Zeitstempel und Pfade byteidentisch.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-10-05 20:50:19 +02:00
commit 84821a2fcd
1 changed files with 73 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

104
epub2md/pipeline.py
View file

@ -131,11 +131,36 @@ def _load_docs(reader: EpubReader, package: Any, diag: Diagnostics) -> list[Doc]
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")
diag.step(f"EPUB {package.version}: {package.metadata.title!r}, {len(package.spine)} Spine-Dokumente")
docs = _load_docs(reader, package, diag)
_prepare_docs(docs, package, diag)
notes, chapters, resolver = _build_structure(docs, package, opts, diag)
rendered = _render(reader, package, chapters, resolver, notes, opts, diag)
files, all_chapters, report = _assemble(input_path, output_dir, opts, diag, generated_at, sha, package, docs,
notes, chapters, resolver, rendered)
if opts.dry_run:
return Result(report=report, chapters=all_chapters, files=files, output_dir=output_dir)
# --- normalization and HTML-TOC handling
_write_output(reader, input_path, output_dir, opts, files, rendered.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)
@dataclass
class _Rendered:
renderer: Renderer
media: MediaRegistry
cover_out: str
endnotes: Chapter | None
endnotes_md: str
skipped_formula_images: int
def _prepare_docs(docs: list[Doc], package: Any, diag: Diagnostics) -> None:
"""Normalize publisher markup and drop HTML tables of contents that NAV/NCX already cover."""
title_tables = book_uses_title_tables(docs)
for doc in docs:
normalize_doc(doc, diag, title_tables)
@ -156,13 +181,16 @@ def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Optio
for d in docs:
d.skipped = ""
# --- notes, chapters, links
def _build_structure(docs: list[Doc], package: Any, opts: Options,
diag: Diagnostics) -> tuple[Any, list[Chapter], LinkResolver]:
"""Notes, chapter units with file names, and the link resolver (incl. NAV/NCX targets)."""
md = package.metadata
notes = collect_notes(docs, opts.footnotes, diag)
diag.step(f"{len(notes.notes)} Anmerkungen erkannt")
chapters = ChapterBuilder(opts.chapter_level, package.toc, diag, md.title, md.authors).build(docs)
_assign_filenames(chapters, opts.slug_language)
readme_path = "README.md" if opts.toc else None
resolver = LinkResolver(docs, chapters, notes, readme_path)
resolver = LinkResolver(docs, chapters, notes, "README.md" if opts.toc else None)
nav_targets: list[tuple[str, str]] = []
def walk(entries: list[Any]) -> None:
@ -174,8 +202,13 @@ def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Optio
walk(package.toc)
resolver.add_targets(nav_targets)
diag.step(f"{len(chapters)} Kapitel gebildet")
return notes, chapters, resolver
# --- rendering
def _render(reader: EpubReader, package: Any, chapters: list[Chapter], resolver: LinkResolver, notes: Any,
opts: Options, diag: Diagnostics) -> _Rendered:
"""Render all chapters (and the end-of-book notes file) to Markdown and register media."""
md = package.metadata
media = MediaRegistry(package, opts.images, opts.slug_language)
cover_out = ""
if md.cover_path and opts.extract_media and reader.exists(md.cover_path):
@ -196,13 +229,7 @@ def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Optio
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)
_report_wrap_markers(renderer, diag)
endnotes: Chapter | None = None
endnotes_md = ""
if opts.footnotes == "end-of-book" and renderer.book_notes:
@ -211,8 +238,23 @@ def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Optio
endnotes.filename = renderer.endnotes_file
endnotes.stats.footnotes = len(renderer.book_notes)
endnotes.stats.word_count = _word_count(endnotes_md)
skipped = _collect_remaining_media(reader, package, media, renderer, opts)
return _Rendered(renderer, media, cover_out, endnotes, endnotes_md, skipped)
# --- remaining media (fonts, css, audio, unreferenced images)
def _report_wrap_markers(renderer: Renderer, diag: Diagnostics) -> None:
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)
def _collect_remaining_media(reader: EpubReader, package: Any, media: MediaRegistry, renderer: Renderer,
opts: Options) -> int:
"""Register fonts, CSS, audio and unreferenced images; returns the number of formula images replaced by TeX."""
skipped_formula_images = 0
if opts.extract_media:
for item in package.manifest.values():
@ -229,10 +271,17 @@ def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Optio
for mf in media.files.values():
if mf.data is None and reader.exists(mf.source):
mf.size = reader.names[mf.source].file_size
return skipped_formula_images
# --- 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:
def _assemble(input_path: Path, output_dir: Path, opts: Options, diag: Diagnostics, generated_at: str, sha: str,
package: Any, docs: list[Doc], notes: Any, chapters: list[Chapter], resolver: LinkResolver,
rendered: _Rendered) -> tuple[dict[str, str], list[Chapter], dict[str, Any]]:
"""Build all output files in memory (chapters, README, metadata, manifest, report, diagnostics)."""
md = package.metadata
renderer, media, endnotes = rendered.renderer, rendered.media, rendered.endnotes
for n in notes.notes.values():
if n.ref_count and not n.label and notes.extracted:
diag.warn("note-not-rendered", "Anmerkung wurde referenziert, aber nicht ausgegeben.", n.key)
opf_dir = posixpath.dirname(package.opf_path)
files: dict[str, str] = {}
@ -243,12 +292,12 @@ def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Optio
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
files[endnotes.filename] = front + rendered.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,
files["README.md"] = writer.readme(md, chapters, endnotes, package.toc, resolver, stats, rendered.cover_out,
package.version, input_path.name, opts.command, str(output_dir))
existing = {mf.output for mf in media.files.values()} if opts.extract_media else set()
if opts.copy_original:
@ -261,23 +310,14 @@ def _convert(reader: EpubReader, input_path: Path, output_dir: Path, opts: Optio
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,
docs, broken, rendered.skipped_formula_images)
files["metadata.yaml"] = writer.metadata_yaml(md, input_path.name, sha, package.version, rendered.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)
return files, all_chapters, report
def _mostly_links(el: Any) -> bool: