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).
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""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)
|