bitscuit/pdf-to-audiobook
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pdf-to-audiobook: PDF -> narrated MP3 via edge-tts

Browse Source 
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).
This commit is contained in:
bitscuit 2026-08-25 14:15:50 +02:00
commit a1796e82aa
14 changed files with 1206 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.gitignore vendored Normal file
View File

@ -0,0 +1,8 @@
__pycache__/
*.pyc
*_work/
*.mp3
*.raw.mp3
samples/
audio/
chunks/

167
README.md Normal file
View File

@ -0,0 +1,167 @@
# pdf-to-audiobook
Turn a PDF into a narrated MP3 audiobook: text extraction, OCR-defect
reporting and repair, voice sampling, resumable free edge-tts synthesis,
ffmpeg stitching, and ID3 chapter markers.
Proven on a 546-page scanned book: 546 pages in, a 14h MP3 out, 100+
chapters, zero cost.
## How it works
```
+----------+ +--------+ +-------+ +-------+ +--------+ +----------+
PDF -----+> | extract | ---> | scan | ---> | clean | ---> | chunk | ---> | synth | ---> | stitch |
+----------+ +--------+ +-------+ +-------+ +--------+ +----------+
raw.txt report + clean.txt chunks/ audio/*.mp3 audiobook.mp3
(pdftotext) template (+ chapters) (~2600 chars) (edge-tts) + chapters
```
- **extract** - `pdftotext` + a text-layer quality report (word count,
unreadable-glyph count, verdict).
- **scan** - enumerates every distinct OCR-suspect token with its count and
surrounding context, and can emit a fix-config template. This is the work
horse for scanned PDFs.
- **clean** - noise/page-header removal, hyphenation joining, and applies
your fix config (exact-string maps, regexes, keep/drop windows). Also
records chapter heading positions.
- **voices** - renders one identical sample in N voices, plus one combined
comparison MP3, so you can pick a voice in one playthrough.
- **chunk / synth / stitch** - sentence-aware chunking, resumable parallel
edge-tts synthesis (re-run to pick up stragglers), lossless concat with
RMS spot-checks.
- **chapters** - maps heading positions onto audio time and tags the MP3
with standard ID3v2.4 CHAP chapters (play in players with chapter
support).
## Install
```
pip install -r requirements.txt # edge-tts, mutagen
# plus system tools:
# ffmpeg/ffprobe (any recent build)
# pdftotext (poppler-utils / poppler)
```
## Quick start (clean digital PDF, no OCR damage)
```
python -m pdf2audiobook full book.pdf --voice en-GB-LibbyNeural
```
Everything lands in `book_work/`:
`raw.txt`, `clean.txt`, `chunks/`, `audio/`, `audiobook.mp3`,
`audiobook_chapters.mp3`.
A 250-page book at ~150 wpm is roughly 10h of audio; synthesis of ~100
chunks takes about 15-20 min on the free edge-tts endpoint.
## Step-by-step (scanned PDF, OCR damage, voice selection)
```
# 1. extract + quality report
python -m pdf2audiobook extract book.pdf # workdir: book_work/
# 2. see what the OCR broke
python -m pdf2audiobook --workdir book_work scan --top 200
python -m pdf2audiobook --workdir book_work scan --top 300 \
--template configs/book.json
# 3. fill in configs/book.json (see below), then clean
python -m pdf2audiobook --workdir book_work clean --config configs/book.json
# 4. pick a voice: render a sample in several candidates
python -m pdf2audiobook --workdir book_work voices \
en-GB-LibbyNeural en-GB-SoniaNeural en-US-AriaNeural en-US-ChristopherNeural
# -> book_work/samples/compare_all_voices.mp3 (one playthrough, all voices)
# listen, choose one
# 5. chunk + synthesize (resumable; safe to Ctrl-C and re-run)
python -m pdf2audiobook --workdir book_work chunk
python -m pdf2audiobook --workdir book_work synth en-GB-LibbyNeural
# quick check first: ... synth en-GB-LibbyNeural --limit 5
# 6. stitch + verify (RMS spot-checks for silent sections)
python -m pdf2audiobook --workdir book_work stitch
# 7. chapter markers
python -m pdf2audiobook --workdir book_work chapters
# -> book_work/audiobook_chapters.mp3
```
## Fixing OCR defects
Scanned PDFs break text in systematic ways. The OCR engine of a given scan
misreads the SAME glyph the SAME way (e.g. `;` for `s`, `J` for `th`, `9`
for `d`, and U+FFFD replacement characters for diacritics). So the reliable
tool is not a clever regex but an **exact-string map built from the actual
output**:
1. `scan --template configs/book.json` prints the distinct suspect tokens
with counts and context, and writes a config with every token as an empty
value.
2. Fill in the correct reading for the tokens that matter. Prioritize by
count: the high-frequency terms (names, technical vocabulary) are worth
mapping; hundreds of one-offs in an index or reference table usually
aren't. Leave the rest empty (empty values are ignored).
3. For a handful of pattern-like misreads, `regex_fixes` works, but only
with anchored patterns you have verified against the raw context. A loose
short-pattern "fix" will corrupt real words (this bit the reference
project: a catch-all rule corrupted `measure` six times).
4. `clean --config ...` applies everything and reports remaining U+FFFD
glyphs. Re-run `scan clean.txt` to eyeball what's left.
The config keys:
| key | meaning |
|---|---|
| `include.start` / `include.end` | regexes marking the first/last body lines (cuts front/back matter; `end_fallback` is a second try for `end`, `min_line_*` guard against early false hits) |
| `drop_blocks` | e.g. remove the table of contents: `start` + `end` line regexes, optional `end_followed_by` to require the block's end to be followed by specific lines |
| `page_header_patterns` | running headers to drop as whole lines (match at line start) |
| `drop_line_patterns` | other whole lines to drop |
| `strip_patterns` | inline removals applied to the joined text (plate/figure references, stray inline headers) |
| `raw_fixes` | exact-string fixes applied to the raw text before any structure is touched (typically U+FFFD glyph repairs that must happen before line filtering) |
| `regex_fixes` | `[pattern, replacement]` pairs, applied in order to the joined text |
| `fixes` | exact-string map applied longest-first (the main OCR repair table) |
| `word_fixes` | anchored word-boundary `[pattern, replacement]` pairs, applied last |
| `heading_pattern` | regex matching chapter heading lines (used for chapter markers) |
| `heading_exclude_patterns` | lines that must never count as headings (e.g. numbered instruction sentences) |
| `heading_next_pattern` / `heading_next_skip` | require a heading to be followed by this pattern within N lines (filters running headers that share a heading's shape) |
| `heading_count_max` | safety cap on chapter count |
A heavily-scanned book (foreign-language terms, diacritics, drop caps)
typically needs a few hundred exact-string `fixes` plus a smaller set of
`raw_fixes` (for U+FFFD glyphs) and a handful of anchored `regex_fixes`.
Start from a `scan --template` and fill in the high-frequency tokens first.
## Voices
`python -m edge_tts --list-voices` lists the full catalogue. Useful filters:
`en-GB-`, `en-US-`, `en-AU-`, `en-CA-`, `en-IN-`, `en-IE-`. Known-good as of
2026-08: en-US-{Aria,Jenny,Ava,Christopher,AndrewMultilingual,EmmaMultilingual},
en-GB-{Sonia,Libby,Ryan,Thomas}, en-AU-Natasha, en-CA-{Clara,Liam},
en-IN-{Neerja,Prabhat}. Some names 404 (Davis, Sinara, IE-Emma, ZA-James);
the sampler reports which ones failed.
edge-tts is free and unauthenticated. Keep concurrency at 2 (the default)
and let the built-in retries/backoff handle hiccups. A 546-page book is
~250 chunks in ~30-35 min.
## Output
- `audiobook.mp3` - one continuous file (24 kHz MP3, what edge-tts emits;
fine for speech)
- `audiobook_chapters.mp3` - same audio + ID3v2.4 CHAP chapters
- verify any time: `ffprobe -v error -show_chapters audiobook_chapters.mp3`
## Caveats
- One consistent voice; no expression/pacing control beyond SSML-free plain
text.
- OCR in reference tables, indices, and heavily illustrated pages will not
be fully repairable; the tool reports what remains instead of faking
100% clean.
- Chapters come from real headings in the text; if the scan ate the
heading numbers, some chapters will be missing (the tool would rather
leave a gap than mislabel a chapter).
- Respect copyright: use this for books you own or that are public domain.

38
configs/example.json Normal file
View File

@ -0,0 +1,38 @@
{
"include": {
"start": "^(Foreword|Preface|Introduction)$",
"end": "^(Glossary|Index|Bibliography|Acknowledg(?:ements)?)$",
"end_fallback": "^Appendix",
"min_line_start": 0,
"min_line_end": 2000
},
"drop_blocks": [
{
"start": "^Contents$",
"end": "^PART I$",
"end_followed_by": ["Introduction", "INTRODUCTION", "INTRO"]
}
],
"page_header_patterns": [
"^\\d{1,3}\\s+(Book title|Part [IVX]+)$",
"^(Book title)\\s*\\d{0,3}$"
],
"drop_line_patterns": [
"^(?:[Aa]sana|[Cc]hapter|[Ss]ection)\\s+\\d{1,3}$"
],
"strip_patterns": [
"\\s*\\((?:Plates?|Figure[s]?) [0-9IVXLC]+(?:\\s*(?:to|and) [0-9IVXLC]+)*\\)\\.?\\s*",
"\\s*(?:One|Two|Three|Four|Five|Six)\\*$"
],
"raw_fixes": {
"U+FFFD-containing token from `scan`": "corrected reading"
},
"regex_fixes": [
["\\bpalance\\b", "balance"]
],
"fixes": {
"exact token as it appears": "corrected text"
},
"heading_pattern": "^(?:PART [IVX]+\\b|\\d{1,3}\\.\\s+\\w|[A-Z][a-z]+\\s+[A-Z][a-z]+)$",
"heading_count_max": 150
}

8
pdf2audiobook/__init__.py Normal file
View File

@ -0,0 +1,8 @@
"""pdf2audiobook: turn a PDF into a narrated MP3 audiobook.
Pipeline: extract text -> scan/repair OCR defects -> voice sampling ->
sentence-aware chunking -> resumable edge-tts synthesis -> ffmpeg stitch ->
ID3v2.4 CHAP chapter markers.
"""
__version__ = "0.1.0"

201
pdf2audiobook/__main__.py Normal file
View File

@ -0,0 +1,201 @@
#!/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()

106
pdf2audiobook/chapters.py Normal file
View File

@ -0,0 +1,106 @@
"""Map chapter text positions onto audio time and tag the MP3 (ID3v2.4 CHAP).
Mapping: probe each chunk MP3's duration once (ffprobe) to get chunk start
times in ms; use chunk_offsets.json (start char-offset of each chunk in the
cleaned text) to binary-search each chapter's position. Each chapter spans
from its start to the next chapter's start (last: end of file).
Uses mutagen's CHAP class (1.x kwargs: start_time/end_time in ms,
element_id, sub_frames). Hand-rolling the CHAP bytes is error-prone.
Pitfalls baked in (all hit in the reference 2026-08 session):
- work on a COPY of the audio (ID3() + save() to a missing path writes a
tag-only file that ffprobe cannot read)
- TXXX uses text= (value= silently stores empty)
- many players display element_id, not the TXXX subframe, so the title
goes into element_id too (Latin1, CHnnn fallback for non-Latin1)
"""
import json
import os
import shutil
import subprocess
import sys
from bisect import bisect_right
def probe_chunk_times(audio_dir: str) -> tuple:
"""Return (offsets_ms, total_ms): start time of each chunk in ms."""
files = sorted(f for f in os.listdir(audio_dir) if f.endswith(".mp3"))
offsets = []
t = 0
for fn in files:
offsets.append(t)
r = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=nw=1:nk=1", os.path.join(audio_dir, fn)],
capture_output=True, text=True, check=True,
)
t += int(float(r.stdout.strip()) * 1000)
return offsets, t
def main(clean_path: str, audio_dir: str, in_mp3: str, out_mp3: str):
# clean.py writes <clean>.chapters.json / <clean>.chunk_offsets.json
stem = os.path.splitext(clean_path)[0]
chapters = json.load(open(stem + ".chapters.json"))
offsets_path = stem + ".chunk_offsets.json"
if os.path.exists(offsets_path):
char_offsets = json.load(open(offsets_path))
else:
# fallback: even distribution (crude, but keeps the tool working)
total_chars = len(open(clean_path, encoding="utf-8").read())
n = len([f for f in os.listdir(audio_dir) if f.endswith(".mp3")])
char_offsets = [int(i * total_chars / n) for i in range(n)]
print("WARNING: chunk_offsets.json missing; using even char distribution")
offsets_ms, total_ms = probe_chunk_times(audio_dir)
r = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=nw=1:nk=1", in_mp3],
capture_output=True, text=True, check=True,
)
file_end_ms = int(float(r.stdout.strip()) * 1000)
entries = []
for pos, title in chapters:
i = bisect_right(char_offsets, pos) - 1
i = max(0, min(i, len(offsets_ms) - 1))
entries.append([offsets_ms[i], title])
entries.sort(key=lambda e: e[0])
# drop two chapters that land in the same chunk (zero-length span)
dedup = []
for st, title in entries:
if dedup and dedup[-1][0] == st:
continue
dedup.append([st, title])
entries = dedup
if not entries:
raise SystemExit("no usable chapters; check heading_pattern in the config")
from mutagen.id3 import CHAP, ID3, ID3Tags, TXXX
shutil.copy(in_mp3, out_mp3)
tags = ID3(out_mp3)
for n, (st, title) in enumerate(entries, 1):
en = entries[n][0] if n < len(entries) else file_end_ms
elem_id = title if title.encode("latin-1", "replace") == title.encode("latin-1") \
else f"CH{n:03d}"
sub = ID3Tags()
sub.add(TXXX(desc="CHAPTER", text=title))
tags.add(CHAP(element_id=elem_id, start_time=int(st),
end_time=int(en), sub_frames=sub))
tags.save(out_mp3, v2_version=4)
print(json.dumps({
"out": out_mp3,
"chapters": len(entries),
"total_ms": total_ms,
"file_end_ms": file_end_ms,
"first": entries[0],
"last": entries[-1],
}, indent=1))
print("verify: ffprobe -v error -show_chapters " + out_mp3)
if __name__ == "__main__":
main(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4])

63
pdf2audiobook/chunk.py Normal file
View File

@ -0,0 +1,63 @@
"""Split cleaned text into sentence-aware chunks for edge-tts.
~2600 chars per chunk keeps each synthesis call comfortably under the
service's limits while preserving sentence boundaries. Overlong sentences
are sub-split on commas. Also records each chunk's start char-offset (chunk_offsets.json)
so chapter positions can later be mapped onto audio time.
"""
import json
import os
import re
def chunk_text(text: str, outdir: str, chunk_size: int = 2600,
offsets_path: str = None) -> int:
text = re.sub(r"\s+", " ", text)
parts = re.split(r"(?<=[.!?;:])\s+", text)
chunks = []
cur = ""
for p in parts:
if len(p) > chunk_size:
if cur:
chunks.append(cur.strip())
cur = ""
sub = re.split(r",\s+", p)
s = ""
for x in sub:
if s and len(s) + len(x) + 1 > chunk_size:
chunks.append(s.strip())
s = x
else:
s = (s + ", " + x) if s else x
cur = s
continue
if cur and len(cur) + len(p) + 1 > chunk_size:
chunks.append(cur.strip())
cur = p
else:
cur = (cur + " " + p) if cur else p
if cur.strip():
chunks.append(cur.strip())
# start char-offset of each chunk within the whitespace-normalized text
offsets = []
c = 0
for ch in chunks:
offsets.append(c)
c += len(ch) + 1 # +1: chunks are joined with a single space
os.makedirs(outdir, exist_ok=True)
# wipe stale chunks from a previous run with a different chunk size
for f in os.listdir(outdir):
if f.endswith(".txt"):
os.remove(os.path.join(outdir, f))
for i, ch in enumerate(chunks):
with open(os.path.join(outdir, f"{i:04d}.txt"), "w", encoding="utf-8") as f:
f.write(ch)
if offsets_path:
with open(offsets_path, "w", encoding="utf-8") as f:
json.dump(offsets, f)
return len(chunks)

207
pdf2audiobook/clean.py Normal file
View File

@ -0,0 +1,207 @@
"""Clean raw pdftotext output into TTS-ready text.
The pass order mirrors the pipeline proven on a 546-page scanned book:
1. exact-string raw fixes (typically U+FFFD glyph repairs)
2. keep window [include.start, include.end)
3. drop blocks (e.g. table of contents)
4. form feeds -> newlines
5. join hyphenated line breaks
6. line filter (noise lines, page headers, custom drop patterns)
7. join lines into one string
8. strip inline patterns (plate refs, running headers wedged mid-sentence)
9. regex fixes (order preserved from config)
10. exact-string fixes (longest first)
All patterns come from an optional JSON config; without one you still get
the generic noise/header removal and hyphen joining, which is enough for
digitally-born PDFs.
"""
import json
import os
import re
FFFD = "\ufffd"
# Lines that are pure layout noise: lone characters, dots, page numbers,
# roman numerals, bullet/dash lines.
NOISE_RE = re.compile(
r"^(\s*[\u2022\u00b7.]+|\s*[0-9]{1,3}\s*|\s*[IVXLCDM]{1,6}\s*"
r"|\s*I\s*|\s*\u00a7\s*|\s*[\-\u2013]\s*)$"
)
# Headings are expected to be short standalone lines; longer "headings" are
# usually corrupted body text.
MAX_HEADING_LEN = 90
def unhyphenate(t: str) -> str:
t = t.replace("\u00ad\n", "") # soft hyphen
t = t.replace("\u2010\n", "") # hyphen char
t = re.sub(r"(\w)-\n(\w)", r"\1\2", t)
t = re.sub(r"(\w)\\\-\n(\w)", r"\1\2", t) # OCR backslash-hyphen
t = re.sub(r"(\w)\\ \\\\\n", r"\1", t) # OCR backslash-space-backslash
return t
def _find_line(lines, pattern, min_line=0):
rx = re.compile(pattern)
for i in range(min_line, len(lines)):
# match on the stripped line: pdftotext prefixes the first line of
# each page with a form feed, which would defeat ^...$ anchors
if rx.search(lines[i].strip()):
return i
return None
def clean(raw_path: str, cfg_path: str = None, out_path: str = "clean.txt") -> dict:
cfg = {}
if cfg_path:
with open(cfg_path, encoding="utf-8") as f:
cfg = json.load(f)
text = open(raw_path, encoding="utf-8").read()
# 1. raw-level exact fixes (applied before any structure is touched)
rawfix = cfg.get("raw_fixes", {})
for k in sorted(rawfix, key=len, reverse=True):
if k in text:
text = text.replace(k, rawfix[k])
lines = text.split("\n")
# 3. drop blocks FIRST (a TOC entry can look like the include.start
# anchor; the old LOY run removed the TOC before cutting front matter)
for block in cfg.get("drop_blocks", []):
s = _find_line(lines, block["start"], block.get("min_line_start", 0))
if s is None:
continue
e = len(lines)
if block.get("end"):
followed = block.get("end_followed_by")
for i in range(s + 1, len(lines)):
if re.search(block["end"], lines[i]):
if followed and not any(
lines[k].strip() in followed
for k in range(i + 1, min(i + 4, len(lines)))
):
continue # keep looking
e = i
break
lines = lines[:s] + lines[e:]
# 2. include window
inc = cfg.get("include")
if inc and inc.get("start"):
s = _find_line(lines, inc["start"], inc.get("min_line_start", 0))
if s is None:
raise SystemExit(f"include.start pattern not found: {inc['start']!r}")
e = _find_line(lines, inc["end"], max(inc.get("min_line_end", 0), s + 1))
if e is None and inc.get("end_fallback"):
e = _find_line(lines, inc["end_fallback"], max(inc.get("min_line_end", 0), s + 1))
if e is None:
e = len(lines)
lines = lines[s:e]
# 4. form feeds
text = "\n".join(lines).replace("\f", "\n")
# 5. hyphenated line breaks
text = unhyphenate(text)
# 6. line filter (and record chapter heading positions)
hdr_pats = [re.compile(p) for p in cfg.get("page_header_patterns", [])]
drop_pats = [re.compile(p) for p in cfg.get("drop_line_patterns", [])]
heading_re = re.compile(cfg["heading_pattern"]) if cfg.get("heading_pattern") else None
# optional guard: a heading line only counts if it is followed (within a
# few lines) by this pattern. Filters running headers that share the
# heading's shape ("Appendix I" on every page of the appendix).
heading_next_re = re.compile(cfg["heading_next_pattern"]) \
if cfg.get("heading_next_pattern") else None
heading_next_skip = cfg.get("heading_next_skip", 3)
# lines matching these are never headings (but stay in the text):
# numbered instruction sentences that look like "N. Name" headings
heading_excl = [re.compile(p) for p in cfg.get("heading_exclude_patterns", [])]
heading_count_max = cfg.get("heading_count_max", 150)
out_lines = []
pos = 0 # char offset of the current line in the joined text
chapters = [] # (char_offset_in_clean_text, title)
lines2 = text.split("\n")
for i2, line in enumerate(lines2):
s = line.strip()
if not s:
continue
if NOISE_RE.match(s):
continue
if any(p.match(s) for p in hdr_pats):
continue
if any(p.match(s) for p in drop_pats):
continue
# (heading patterns are written to accept only proper-case or
# numbered lines, so no extra case check is needed here)
if (heading_re and len(chapters) < heading_count_max
and len(s) <= MAX_HEADING_LEN
and heading_re.match(s)
and not any(p.search(s) for p in heading_excl)):
if heading_next_re:
ok = any(
heading_next_re.search(lines2[k].strip())
for k in range(i2 + 1, min(i2 + 1 + heading_next_skip, len(lines2)))
)
if not ok:
continue
title = s.rstrip(" .:;-")
chapters.append((pos, title))
out_lines.append(s)
pos += len(s) + 1
joined = re.sub(r" +", " ", " ".join(out_lines))
# 8. inline strips (applied on the joined text)
for p in cfg.get("strip_patterns", []):
joined = re.sub(p, " ", joined)
# 9. regex fixes, in config order (runs BEFORE exact-string fixes,
# matching the reference pipeline: regexes may leave tokens that the
# exact map then clobbers correctly)
for pat, rep in cfg.get("regex_fixes", []):
joined = re.sub(pat, rep, joined)
# 10. exact-string fixes, longest first
for k in sorted(cfg.get("fixes", {}), key=len, reverse=True):
if k in joined:
joined = joined.replace(k, cfg["fixes"][k])
# 11. word-boundary regex fixes (anchored, last, like the reference run)
for pat, rep in cfg.get("word_fixes", []):
joined = re.sub(pat, rep, joined)
joined = re.sub(r" +", " ", joined).strip()
with open(out_path, "w", encoding="utf-8") as f:
f.write(joined)
# Re-anchor chapter positions in the final text: the string fixes above
# can shift offsets, so re-find each title near its recorded position.
# Also drop duplicate titles (running headers that share a heading's
# shape would otherwise create a chapter on every page).
seen_titles = set()
final_chapters = []
for pos, title in chapters:
if title in seen_titles:
continue
seen_titles.add(title)
i = joined.find(title, max(0, pos - 600))
final_chapters.append([i if i >= 0 else pos, title])
chapters_path = os.path.splitext(out_path)[0] + ".chapters.json"
with open(chapters_path, "w", encoding="utf-8") as f:
json.dump(final_chapters, f, indent=1, ensure_ascii=False)
return {
"out": out_path,
"chars": len(joined),
"words": len(joined.split()),
"fffd_left": joined.count(FFFD),
"chapters": len(final_chapters),
"chapters_path": chapters_path,
}

43
pdf2audiobook/extract.py Normal file
View File

@ -0,0 +1,43 @@
"""Text extraction from PDF (poppler pdftotext)."""
import os
import subprocess
FFFD = "\ufffd"
def extract(pdf_path: str, out_path: str) -> dict:
"""Run pdftotext and report basic text-layer statistics.
Returns a dict with path, chars, words, fffd_count, and a short
quality verdict (useful for deciding how much cleanup a book needs).
"""
if not os.path.exists(pdf_path):
raise FileNotFoundError(pdf_path)
if not os.path.exists(out_path):
subprocess.run(
["pdftotext", pdf_path, out_path],
check=True,
capture_output=True,
text=True,
)
text = open(out_path, encoding="utf-8").read()
words = len(text.split())
fffd = text.count(FFFD)
if words == 0:
verdict = "NO TEXT LAYER: the PDF appears to be a pure image scan. " \
"OCR it first (e.g. ocrmypdf) or the pipeline has nothing to read."
elif fffd > words * 0.01:
verdict = "HEAVY OCR DAMAGE: many unreadable glyphs (U+FFFD). " \
"Run `scan` and build a fix config before synthesizing."
elif fffd > 0:
verdict = "MINOR OCR DAMAGE: some unreadable glyphs. " \
"Run `scan` to see them; a fix config is recommended."
else:
verdict = "CLEAN TEXT LAYER: no replacement characters found."
return {
"path": out_path,
"chars": len(text),
"words": words,
"fffd": fffd,
"verdict": verdict,
}

99
pdf2audiobook/scan.py Normal file
View File

@ -0,0 +1,99 @@
"""Scan extracted text for OCR defects and generate a fix-config template.
Scanned PDFs mangle text in systematic ways:
- U+FFFD (replacement character) where the OCR engine gave up, usually on
diacritics and unusual consonants (foreign-language terms, umlauts, ...)
- Confused characters: ; -> s/h, J -> th/v, 9 -> d, ! -> l, < -> sh, ...
The OCR engine of a given scan misreads the SAME glyph the SAME way, so the
right tool is an exact-string replacement map built from the actual output.
`scan` enumerates the distinct suspicious tokens with counts and surrounding
context; you (or a language model) then fill in the correct reading, and
`clean` applies the map.
Output of scan:
- a human-readable report on stdout
- optionally a fix-config template JSON with the suspicious tokens as
empty values, ready to be filled in: {"fixes": {"mangled-token": ""}}
"""
import json
import re
import sys
FFFD = "\ufffd"
# Characters that rarely appear inside a real English word. A token that has
# one of these in the MIDDLE of the word (or FFFD anywhere) is suspect.
SUSPECT = set(";<>![]\\_`|~^@#%&*+=")
TOKEN_RE = re.compile(r"[\w" + FFFD + r";<>!\\[\]_`\-]+")
def is_suspect(token: str) -> bool:
if FFFD in token:
return True
body = token.strip("-_")
if len(body) < 3:
return False
# suspect char inside the word (not a leading/trailing punctuation)
return any(c in SUSPECT for c in body[1:-1])
def scan(text: str, top: int = 300) -> list:
"""Return [(count, token, context)] sorted by count desc, then token."""
counts = {}
context = {}
for m in TOKEN_RE.finditer(text):
tok = m.group(0)
if not is_suspect(tok):
continue
counts[tok] = counts.get(tok, 0) + 1
if tok not in context:
s = max(0, m.start() - 40)
e = min(len(text), m.end() + 40)
ctx = text[s:e].replace("\n", " ")
context[tok] = ctx
items = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
return [(c, t, context[t]) for t, c in items[:top]]
def cmd_scan(path: str, top: int = 300, template: str = None) -> None:
text = open(path, encoding="utf-8").read()
total_fffd = text.count(FFFD)
words = len(text.split())
items = scan(text, top=top)
print(f"text: {path}")
print(f"words: {words} U+FFFD glyphs: {total_fffd} "
f"suspect tokens (distinct): {len(items)}")
print()
print(f"{'count':>6} {'token':<32} context")
print("-" * 100)
for count, tok, ctx in items:
print(f"{count:>6} {tok!r:<32} ...{ctx}...")
if template:
fixes = {tok: "" for _, tok, _ in items}
cfg = {
"include": {"start": None, "end": None},
"drop_blocks": [],
"page_header_patterns": [],
"drop_line_patterns": [],
"strip_patterns": [],
"raw_fixes": {},
"regex_fixes": [],
"word_fixes": [],
"fixes": fixes,
"heading_pattern": None,
"heading_count_max": 150,
}
with open(template, "w", encoding="utf-8") as f:
json.dump(cfg, f, indent=2, ensure_ascii=False)
print(f"\nfix-config template written to {template} "
f"({len(fixes)} tokens to review). Fill in the empty values "
f"with the correct reading; empty values are ignored by clean.")
print("See README 'Fixing OCR defects' for the workflow.")
if __name__ == "__main__":
cmd_scan(sys.argv[1])

80
pdf2audiobook/stitch.py Normal file
View File

@ -0,0 +1,80 @@
"""Stitch chunk MP3s into one audiobook and verify it.
All chunks share edge-tts codec parameters (24 kHz MP3), so a stream-copy
concat (-c copy) is lossless and fast. Verification uses astats RMS at
several timestamps to catch dead/silent sections early.
"""
import json
import os
import subprocess
import sys
def stitch(audio_dir: str, out_path: str, verify: bool = True) -> dict:
files = sorted(f for f in os.listdir(audio_dir) if f.endswith(".mp3"))
if not files:
raise SystemExit(f"no mp3 files in {audio_dir}")
# the list file sits NEXT TO the chunks; the concat demuxer resolves
# names relative to the list file's directory and does NOT understand
# JSON-style double quotes (it treats them as part of the filename)
listfile = os.path.join(audio_dir, "_concat_list.txt")
with open(listfile, "w") as f:
for fn in files:
f.write(f"file '{fn}'\n")
raw = out_path + ".raw.mp3"
if os.path.exists(raw):
os.remove(raw)
subprocess.run(
["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", listfile, "-c", "copy", raw],
check=True,
)
os.replace(raw, out_path)
os.remove(listfile)
info = {"out": out_path, "chunks": len(files)}
if verify:
r = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=nw=1:nk=1", out_path],
capture_output=True, text=True, check=True,
)
dur = float(r.stdout.strip())
info["duration_s"] = dur
info["size_mb"] = round(os.path.getsize(out_path) / 1e6, 1)
# spot-check RMS at a few positions
checks = []
for frac in (0.05, 0.3, 0.6, 0.9):
t = dur * frac
a = subprocess.run(
["ffmpeg", "-hide_banner", "-ss", f"{t:.1f}", "-t", "4", "-i", out_path,
"-af", "astats=metadata=1", "-f", "null", "-"],
capture_output=True, text=True,
)
rms = None
for line in a.stderr.splitlines():
if "RMS level dB:" in line:
try:
rms = float(line.split("RMS level dB:")[1].strip())
except ValueError:
pass
break
checks.append((round(frac, 2), t and round(t), rms))
info["rms_checks"] = checks
silent = [c for c in checks if c[2] is not None and c[2] < -50]
info["silent_sections"] = len(silent)
return info
def main(audio_dir, out_path):
info = stitch(audio_dir, out_path)
print(json.dumps(info, indent=1))
if info.get("silent_sections"):
print("WARNING: silent sections detected; check the RMS checks above")
sys.exit(1)
if __name__ == "__main__":
main(sys.argv[1], sys.argv[2])

79
pdf2audiobook/synth.py Normal file
View File

@ -0,0 +1,79 @@
"""Resumable edge-tts batch synthesis.
- One MP3 per chunk, in <workdir>/audio/
- Skips chunks that already have a non-trivial MP3 (safe to re-run after a
crash or to pick up stragglers)
- 2-way concurrency + retries with backoff keeps the free service happy
(a 546-page book ~= 251 chunks finished in ~30-35 min wall time)
- --limit N synthesizes only the first N missing chunks (for quick checks)
"""
import asyncio
import glob
import os
import sys
import edge_tts
MIN_BYTES = 1000
RETRIES = 5
async def synth_one(text: str, voice: str, out: str) -> bool:
for attempt in range(RETRIES):
try:
await edge_tts.Communicate(text, voice).save(out)
if os.path.exists(out) and os.path.getsize(out) > MIN_BYTES:
return True
except Exception:
pass
if os.path.exists(out) and os.path.getsize(out) < MIN_BYTES:
os.remove(out)
await asyncio.sleep(2 * (attempt + 1))
return False
async def run(chunks_dir: str, outdir: str, voice: str, limit: int = None) -> None:
os.makedirs(outdir, exist_ok=True)
files = sorted(glob.glob(os.path.join(chunks_dir, "*.txt")))
todo = []
for f in files:
name = os.path.basename(f)[:-4]
if os.path.exists(os.path.join(outdir, f"{name}.mp3")):
continue
todo.append((name, open(f, encoding="utf-8").read()))
if limit:
todo = todo[:limit]
print(f"voice={voice} total_chunks={len(files)} todo={len(todo)}", flush=True)
if not todo:
print("nothing to do")
return
sem = asyncio.Semaphore(2)
done = 0
async def worker(name, text):
nonlocal done
async with sem:
ok = await synth_one(text, voice, os.path.join(outdir, f"{name}.mp3"))
done += 1
if not ok:
print(f"FAILED {name}", flush=True)
elif done % 10 == 0:
print(f"progress {done}/{len(todo)}", flush=True)
await asyncio.gather(*(worker(n, t) for n, t in todo))
# final consistency report
missing = [
os.path.basename(f)[:-4] for f in files
if not os.path.exists(os.path.join(outdir, os.path.basename(f)[:-4] + ".mp3"))
]
print(f"done: {done}/{len(todo)}; missing chunks overall: {len(missing)}", flush=True)
if missing:
print("re-run synth to pick up the missing ones", flush=True)
sys.exit(1)
def main(chunks_dir, outdir, voice, limit=None):
asyncio.run(run(chunks_dir, outdir, voice, limit))

105
pdf2audiobook/voices.py Normal file
View File

@ -0,0 +1,105 @@
"""Voice sampling for edge-tts.
`edge-tts --list-voices` returns the full catalogue (hundreds of voices in
many languages), so this module offers:
- filter voices by language / gender / gender-ish names
- render one identical sample text in N voices (resumable: existing
samples are skipped)
- optionally concatenate all samples into one comparison MP3 with a
1.2 s silence gap between voices, so a single playthrough compares them
"""
import asyncio
import os
import subprocess
import edge_tts
def list_voices(lang_prefix: str = None, gender: str = None) -> list:
"""List available edge-tts voices, optionally filtered.
lang_prefix: e.g. 'en' or 'en-GB'; gender: 'Male'/'Female'.
Returns list of dicts (name, shortName, gender, locale).
"""
r = subprocess.run(
["python", "-m", "edge_tts", "--list-voices"],
capture_output=True, text=True, check=True,
)
# the CLI prints 'Name Short Gender Age ...' lines after a header
out = []
for line in r.stdout.splitlines():
parts = line.split()
if len(parts) >= 4 and parts[0].endswith("Neural"):
v = {"name": parts[0], "short": parts[1], "gender": parts[2]}
if lang_prefix and not v["name"].lower().startswith(lang_prefix.lower()):
continue
if gender and v["gender"] != gender:
continue
out.append(v)
return out
async def _render_sample(text: str, voice: str, out: str) -> bool:
if os.path.exists(out) and os.path.getsize(out) > 1000:
return True
for attempt in range(4):
try:
await edge_tts.Communicate(text, voice).save(out)
if os.path.exists(out) and os.path.getsize(out) > 1000:
return True
except Exception:
pass
await asyncio.sleep(1.5 * (attempt + 1))
return False
async def sample(text: str, voices: list, outdir: str, compare: bool = True) -> list:
"""Render the same text in each voice. Returns [(voice, file, ok)]."""
os.makedirs(outdir, exist_ok=True)
sem = asyncio.Semaphore(4)
async def one(voice):
name = voice.replace("/", "_")
out = os.path.join(outdir, f"{name}.mp3")
async with sem:
ok = await _render_sample(text, voice, out)
return (voice, out, ok)
results = list(await asyncio.gather(*(one(v) for v in voices)))
if compare:
files = [f for _, f, ok in results if ok]
if len(files) >= 2:
silence = os.path.join(outdir, "_silence12.mp3")
if not os.path.exists(silence):
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i", "anullsrc=r=24000:cl=mono",
"-t", "1.2", "-ar", "24000", "-b:a", "48k", silence],
capture_output=True, check=True,
)
listfile = os.path.join(outdir, "_compare_list.txt")
with open(listfile, "w") as f:
# single quotes only: the concat demuxer does not parse
# JSON-style double quotes
f.write("file '_silence12.mp3'\n")
for v, fp, ok in results:
if ok:
f.write(f"file '{os.path.basename(fp)}'\n")
else:
f.write("file '_silence12.mp3'\n")
comp = os.path.join(outdir, "compare_all_voices.mp3")
subprocess.run(
["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", listfile,
"-c", "copy", comp],
capture_output=True, check=True,
)
print(f"comparison file: {comp}")
ok = 0
for voice, fp, good in results:
if good:
ok += 1
print(f"OK {voice:<28} {fp}")
else:
print(f"FAIL {voice:<28} (voice 404 or network error)")
print(f"{ok}/{len(voices)} samples rendered")
return results

2
requirements.txt Normal file
View File

@ -0,0 +1,2 @@
edge-tts>=7.0
mutagen>=1.40