bitscuit
a1796e82aa
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).
81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""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])
|