Pipeline: extract (pdftotext) -> scan (OCR-defect report + fix template) -> clean (noise/header removal, hyphenation, exact-string/regex OCR repairs, chapter-heading detection) -> voices (sample + comparison) -> chunk (sentence-aware) -> synth (resumable edge-tts) -> stitch (ffmpeg concat + RMS verify) -> chapters (ID3v2.4 CHAP). Config-driven per book; validated end-to-end on a clean digital PDF and on a 546-page scanned book (byte-identical clean output to the reference run, 95 auto-detected chapters).
202 lines
7.9 KiB
Python
202 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
"""pdf2audiobook - turn a PDF into a narrated MP3 audiobook.
|
|
|
|
Subcommands (run in order for a full book):
|
|
|
|
extract PDF -> raw.txt (pdftotext + quality report)
|
|
scan raw.txt -> suspect token report (+ optional fix template)
|
|
clean raw.txt -> clean.txt + clean.chapters.json (+ .chunk_offsets.json)
|
|
voices <voice names> -> samples/*.mp3 + compare_all_voices.mp3
|
|
chunk clean.txt -> chunks/*.txt + clean.chunk_offsets.json
|
|
synth <voice> -> audio/*.mp3 (resumable; --limit N for checks)
|
|
stitch audio/ -> <book>.mp3 (concat + RMS verification)
|
|
chapters clean.txt -> <book>_chapters.mp3 (ID3v2.4 CHAP markers)
|
|
full PDF -> runs the whole pipeline with --voice/--config
|
|
|
|
All files land in the --workdir (default: <pdf>_work/).
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from pdf2audiobook import chunk as chunk_mod
|
|
from pdf2audiobook import clean as clean_mod
|
|
from pdf2audiobook import chapters as chapters_mod
|
|
from pdf2audiobook import extract as extract_mod
|
|
from pdf2audiobook import scan as scan_mod
|
|
from pdf2audiobook import stitch as stitch_mod
|
|
from pdf2audiobook import synth as synth_mod
|
|
from pdf2audiobook import voices as voices_mod
|
|
|
|
|
|
def workdir_for(args) -> str:
|
|
wd = args.workdir
|
|
if wd is None:
|
|
base = os.path.basename(args.pdf or "book").rsplit(".", 1)[0]
|
|
wd = f"{base}_work"
|
|
os.makedirs(wd, exist_ok=True)
|
|
return wd
|
|
|
|
|
|
def cmd_extract(a):
|
|
wd = workdir_for(a)
|
|
info = extract_mod.extract(a.pdf, os.path.join(wd, "raw.txt"))
|
|
print(json.dumps(info, indent=1))
|
|
print(f"next: pdf2audiobook --workdir {wd} scan (if the report shows OCR damage)")
|
|
|
|
|
|
def cmd_scan(a):
|
|
scan_mod.cmd_scan(os.path.join(a.workdir, "raw.txt"), top=a.top,
|
|
template=a.template)
|
|
|
|
|
|
def cmd_clean(a):
|
|
info = clean_mod.clean(os.path.join(a.workdir, "raw.txt"),
|
|
cfg_path=a.config,
|
|
out_path=os.path.join(a.workdir, "clean.txt"))
|
|
print(json.dumps(info, indent=1))
|
|
if info["fffd_left"]:
|
|
print(f"NOTE: {info['fffd_left']} U+FFFD glyphs remain in clean.txt; "
|
|
f"review them with `scan clean.txt` if they matter")
|
|
|
|
|
|
def _sample_text(path, chars=1500):
|
|
text = open(path, encoding="utf-8").read()
|
|
# sample from ~1/3 in, skip title pages/front matter
|
|
start = len(text) // 3
|
|
return text[start:start + chars].strip()
|
|
|
|
|
|
def cmd_voices(a):
|
|
import asyncio
|
|
|
|
path = a.text or os.path.join(a.workdir, "clean.txt")
|
|
if not os.path.exists(path):
|
|
raise SystemExit(f"sample text file not found: {path} "
|
|
f"(run extract/clean first, or pass --text)")
|
|
text = _sample_text(path, a.chars)
|
|
outdir = os.path.join(a.workdir, "samples")
|
|
print(f"sample text ({len(text)} chars): {text[:120]}...")
|
|
asyncio.run(voices_mod.sample(text, a.voices, outdir, compare=not a.no_compare))
|
|
|
|
|
|
def cmd_chunk(a):
|
|
n = chunk_mod.chunk_text(
|
|
open(os.path.join(a.workdir, "clean.txt"), encoding="utf-8").read(),
|
|
os.path.join(a.workdir, "chunks"),
|
|
chunk_size=a.size,
|
|
offsets_path=os.path.join(a.workdir, "clean.chunk_offsets.json"),
|
|
)
|
|
print(f"{n} chunks written to chunks/ (avg {a.size} chars)")
|
|
|
|
|
|
def cmd_synth(a):
|
|
synth_mod.main(os.path.join(a.workdir, "chunks"),
|
|
os.path.join(a.workdir, "audio"),
|
|
a.voice, limit=a.limit)
|
|
|
|
|
|
def cmd_stitch(a):
|
|
out = a.out or os.path.join(a.workdir, "audiobook.mp3")
|
|
info = stitch_mod.stitch(os.path.join(a.workdir, "audio"), out)
|
|
print(json.dumps(info, indent=1))
|
|
if info.get("silent_sections"):
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_chapters(a):
|
|
out = a.out or os.path.join(a.workdir, "audiobook_chapters.mp3")
|
|
chapters_mod.main(os.path.join(a.workdir, "clean.txt"),
|
|
os.path.join(a.workdir, "audio"),
|
|
os.path.join(a.workdir, "audiobook.mp3"), out)
|
|
|
|
|
|
def cmd_full(a):
|
|
wd = workdir_for(a)
|
|
print(f"== extract ==")
|
|
extract_mod.extract(a.pdf, os.path.join(wd, "raw.txt"))
|
|
print(f"== clean ==")
|
|
clean_mod.clean(os.path.join(wd, "raw.txt"), cfg_path=a.config,
|
|
out_path=os.path.join(wd, "clean.txt"))
|
|
print(f"== chunk ==")
|
|
chunk_mod.chunk_text(open(os.path.join(wd, "clean.txt"), encoding="utf-8").read(),
|
|
os.path.join(wd, "chunks"), chunk_size=a.size,
|
|
offsets_path=os.path.join(wd, "clean.chunk_offsets.json"))
|
|
print(f"== synth ({a.voice}) ==")
|
|
synth_mod.main(os.path.join(wd, "chunks"), os.path.join(wd, "audio"),
|
|
a.voice, limit=a.limit)
|
|
print(f"== stitch ==")
|
|
out = os.path.join(wd, "audiobook.mp3")
|
|
stitch_mod.stitch(os.path.join(wd, "audio"), out)
|
|
print(f"== chapters ==")
|
|
try:
|
|
chapters_mod.main(os.path.join(wd, "clean.txt"), os.path.join(wd, "audio"),
|
|
out, os.path.join(wd, "audiobook_chapters.mp3"))
|
|
except SystemExit as e:
|
|
print(f"chapters skipped: {e} (audiobook.mp3 is still complete)")
|
|
print(f"\nDONE: {wd}/audiobook.mp3 (and audiobook_chapters.mp3 if tagged)")
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(
|
|
prog="pdf2audiobook",
|
|
description="Turn a PDF into a narrated MP3 audiobook (edge-tts).")
|
|
p.add_argument("--workdir", default=None,
|
|
help="work directory (default: <pdf-name>_work/)")
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
s = sub.add_parser("extract", help="extract text layer with pdftotext")
|
|
s.add_argument("pdf")
|
|
|
|
s = sub.add_parser("scan", help="report suspect/OCR-damaged tokens")
|
|
s.add_argument("--top", type=int, default=200)
|
|
s.add_argument("--template", default=None,
|
|
help="write a fix-config JSON template with the tokens as empty values")
|
|
|
|
s = sub.add_parser("clean", help="apply cleaning passes + optional fix config")
|
|
s.add_argument("--config", default=None, help="fix config JSON (see configs/example.json)")
|
|
|
|
s = sub.add_parser("voices", help="render a sample in several voices for comparison")
|
|
s.add_argument("voices", nargs="+", help="voice names, e.g. en-GB-LibbyNeural en-US-AriaNeural")
|
|
s.add_argument("--text", default=None, help="sample text file (default: clean.txt)")
|
|
s.add_argument("--chars", type=int, default=1500)
|
|
s.add_argument("--no-compare", action="store_true", help="skip the combined comparison MP3")
|
|
|
|
s = sub.add_parser("chunk", help="split clean.txt into synthesis chunks")
|
|
s.add_argument("--size", type=int, default=2600)
|
|
|
|
s = sub.add_parser("synth", help="synthesize chunks (resumable)")
|
|
s.add_argument("voice")
|
|
s.add_argument("--limit", type=int, default=None, help="only the first N missing chunks")
|
|
|
|
s = sub.add_parser("stitch", help="concat chunks into one MP3 + verify")
|
|
s.add_argument("--out", default=None)
|
|
|
|
s = sub.add_parser("chapters", help="tag the MP3 with ID3v2.4 CHAP chapters")
|
|
s.add_argument("--out", default=None)
|
|
|
|
s = sub.add_parser("full", help="run extract..chapters end to end")
|
|
s.add_argument("pdf")
|
|
s.add_argument("--voice", default="en-US-AriaNeural")
|
|
s.add_argument("--config", default=None)
|
|
s.add_argument("--size", type=int, default=2600)
|
|
s.add_argument("--limit", type=int, default=None, help="limit chunks for a quick test render")
|
|
|
|
a = p.parse_args()
|
|
if a.cmd in ("scan", "clean", "voices", "chunk", "synth", "stitch", "chapters") \
|
|
and not a.workdir:
|
|
raise SystemExit("these steps need --workdir (or use `full <pdf>` for the whole pipeline)")
|
|
|
|
{
|
|
"extract": cmd_extract, "scan": cmd_scan, "clean": cmd_clean,
|
|
"voices": cmd_voices, "chunk": cmd_chunk, "synth": cmd_synth,
|
|
"stitch": cmd_stitch, "chapters": cmd_chapters, "full": cmd_full,
|
|
}[a.cmd](a)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|