Audiobook Producer
95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
"""Voice client: script segments -> wav pieces, via an OpenAI-compatible /v1/audio/speech endpoint (URL in settings).
|
|
Pieces are sentence-sized. Output goes to scratch (work_dir), with a manifest the assembler reads."""
|
|
import json
|
|
import os
|
|
import re
|
|
import urllib.request
|
|
|
|
import bookio
|
|
import settings
|
|
|
|
ABBREV = re.compile(r"^(Dr|Mr|Mrs|Ms|St|Mt|Jr|Sr|vs|[A-Z])\.$")
|
|
|
|
|
|
def sentences(text):
|
|
"""Split at . ! ? (optionally followed by a quote) and a space, but not after abbreviations like 'Dr.' or 'E.'."""
|
|
out, start = [], 0
|
|
for m in re.finditer(r"[.!?][\"']?\s+", text):
|
|
before = text[start:m.start() + 1].split()[-1] if text[start:m.start() + 1].split() else ""
|
|
if ABBREV.match(before):
|
|
continue
|
|
out.append(text[start:m.end()].strip())
|
|
start = m.end()
|
|
if text[start:].strip():
|
|
out.append(text[start:].strip())
|
|
return out
|
|
|
|
|
|
def pieces(text, max_chars):
|
|
"""Pack whole sentences into pieces of at most max_chars (a longer sentence stays whole)."""
|
|
out, cur = [], ""
|
|
for s in sentences(text):
|
|
if cur and len(cur) + 1 + len(s) > max_chars:
|
|
out.append(cur)
|
|
cur = s
|
|
else:
|
|
cur = (cur + " " + s).strip()
|
|
if cur:
|
|
out.append(cur)
|
|
return out
|
|
|
|
|
|
def plan(cfg, segs):
|
|
"""Ordered items: {'text', 'gap_after_s'} for speech, {'cue': 'chime', 'gap_after_s'} for cues."""
|
|
v = cfg["voice"]
|
|
items = []
|
|
for s in segs:
|
|
if s["kind"] == "cue":
|
|
items.append({"cue": "chime", "gap_after_s": v["gap_sentence_s"]})
|
|
continue
|
|
ps = pieces(s["text"], v["max_chunk_chars"])
|
|
for i, p in enumerate(ps):
|
|
items.append({"text": p, "gap_after_s": v["gap_paragraph_s"] if i == len(ps) - 1 else v["gap_sentence_s"]})
|
|
return items
|
|
|
|
|
|
def speak(cfg, text):
|
|
v = cfg["voice"]
|
|
if v.get("api") == "chatterbox_tts":
|
|
# Native Chatterbox server route: the only one that accepts exaggeration, cfg_weight and temperature per request.
|
|
body = {"text": text, "voice_mode": "predefined", "predefined_voice_id": v["voice"], "output_format": "wav"}
|
|
else:
|
|
body = {"model": v["model"], "input": text, "voice": v["voice"], "response_format": "wav"}
|
|
body.update(v.get("extra") or {})
|
|
req = urllib.request.Request(v["endpoint"], json.dumps(body).encode(), {"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=v["timeout_s"]) as r:
|
|
data = r.read()
|
|
if data[:4] != b"RIFF":
|
|
raise RuntimeError("voice endpoint did not return a wav file")
|
|
return data
|
|
|
|
|
|
def work_dir_for(cfg, book_dir, name):
|
|
return os.path.join(settings.resolve(cfg, "work_dir"), os.path.basename(book_dir), "audio", name)
|
|
|
|
|
|
def run(cfg, book_dir, only=None):
|
|
book = bookio.read_book(book_dir)
|
|
for name in book["audio_files"]:
|
|
if only and name != only:
|
|
continue
|
|
with open(os.path.join(book_dir, "script", name + ".json"), encoding="utf-8") as f:
|
|
items = plan(cfg, json.load(f))
|
|
out = work_dir_for(cfg, book_dir, name)
|
|
os.makedirs(out, exist_ok=True)
|
|
for i, it in enumerate(items):
|
|
if "text" in it:
|
|
it["wav"] = os.path.join(out, "%04d.wav" % i)
|
|
if not os.path.exists(it["wav"]): # resume: finished pieces are kept
|
|
data = speak(cfg, it["text"])
|
|
with open(it["wav"], "wb") as f:
|
|
f.write(data)
|
|
with open(os.path.join(out, "manifest.json"), "w", encoding="utf-8") as f:
|
|
json.dump(items, f, indent=1, ensure_ascii=False)
|
|
print("%s: %d pieces" % (name, len(items)))
|