bf83bb35cc
The remote had one commit this branch did not (03f8a85), which adds the
Apache-2.0 LICENSE, the NOTICE and SPDX headers to every .py and .sh
file. The merge was clean - the headers sit at the top of each file and
my changes are further down.
The new files this branch added (languages.py, translations.py, the
cluster package, the new tests and scripts) were written after the
merge base, so the licence commit never saw them. They now carry the
same header, so the convention holds across the whole tree.
Nothing else changed. 883 tests passed / 8 failed - the same 8
pre-existing failures.
841 lines
32 KiB
Python
841 lines
32 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Self-healing chapter audio builder pipeline.
|
|
|
|
When triggered, runs a 6-phase pipeline:
|
|
1. Preflight — verify TTS server, model, voice refs, DB
|
|
2. Validate — identify missing/stale utterances
|
|
3. Generate — produce audio with TTS health monitoring + retry
|
|
4. Stitch — concatenate utterance WAVs into chapter WAV
|
|
5. Export — convert chapter WAV to MP3 via ffmpeg
|
|
6. Verify — run quality checks, report issues
|
|
|
|
Progress is written to ``ch{id}_build_status.json`` so the web UI can
|
|
poll ``/api/chapter/{id}/build-status`` for real-time updates.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PHASES = (
|
|
"preflight",
|
|
"validating",
|
|
"generating",
|
|
"stitching",
|
|
"exporting",
|
|
"verifying",
|
|
"complete",
|
|
"error",
|
|
"warning",
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class BuildResult:
|
|
phase: str = "idle"
|
|
detail: str = ""
|
|
progress: int = 0
|
|
total_utterances: int = 0
|
|
generated: int = 0
|
|
skipped: int = 0
|
|
failed: list[int] = field(default_factory=list)
|
|
errors: list[str] = field(default_factory=list)
|
|
chapter_wav: str | None = None
|
|
chapter_mp3: str | None = None
|
|
started_at: str = ""
|
|
elapsed: str = ""
|
|
quality_issues: list[dict] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"phase": self.phase,
|
|
"detail": self.detail,
|
|
"progress": self.progress,
|
|
"total_utterances": self.total_utterances,
|
|
"generated": self.generated,
|
|
"skipped": self.skipped,
|
|
"failed": self.failed,
|
|
"errors": self.errors,
|
|
"chapter_wav": self.chapter_wav,
|
|
"chapter_mp3": self.chapter_mp3,
|
|
"started_at": self.started_at,
|
|
"elapsed": self.elapsed,
|
|
"quality_issues": self.quality_issues,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _load_speakers(data_dir: Path) -> dict[str, dict]:
|
|
path = data_dir / "speakers.json"
|
|
if not path.exists():
|
|
return {}
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
return {s["primary_name"]: s for s in data.get("speakers", [])}
|
|
|
|
|
|
def _build_alias_map(speakers: dict[str, dict]) -> dict[str, dict]:
|
|
alias_map: dict[str, dict] = {}
|
|
for name, info in speakers.items():
|
|
alias_map[name] = info
|
|
for a in info.get("aliases", []):
|
|
if a:
|
|
alias_map[a] = info
|
|
return alias_map
|
|
|
|
|
|
def _resolve_voice_ref(
|
|
speaker: str,
|
|
alias_map: dict[str, dict],
|
|
voices_dir: Path,
|
|
) -> Path | None:
|
|
spk_info = alias_map.get(speaker)
|
|
candidates: list[str] = [speaker]
|
|
if spk_info and spk_info.get("primary_name"):
|
|
candidates.append(spk_info["primary_name"])
|
|
for name in candidates:
|
|
ref_path = voices_dir / (name.lower().replace(" ", "_") + "_ref.wav")
|
|
if ref_path.exists():
|
|
return ref_path
|
|
for name in candidates:
|
|
needle = name.lower().replace(" ", "_")
|
|
for vf in voices_dir.glob("*_ref.wav"):
|
|
if needle in vf.stem.lower():
|
|
return vf
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ChapterBuilder
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ChapterBuilder:
|
|
"""Self-healing chapter audio generation pipeline.
|
|
|
|
Usage::
|
|
|
|
builder = ChapterBuilder(config, chapter_id=6)
|
|
result = builder.run()
|
|
print(result.phase, result.detail)
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
config: Any,
|
|
chapter_id: int,
|
|
temperature: float = 0.5,
|
|
use_seed: bool = True,
|
|
max_retries: int = 3,
|
|
tts_health_interval: float = 30.0,
|
|
) -> None:
|
|
self.config = config
|
|
self.chapter_id = chapter_id
|
|
self.temperature = temperature
|
|
self.use_seed = use_seed
|
|
self.max_retries = max_retries
|
|
self.tts_health_interval = tts_health_interval
|
|
# Generation mode for cached voice prompts. Full-vmatrix ICL generation
|
|
# (prompt_slug -> cached ref_code) is measured at >15 min/utterance on
|
|
# the GPU box; embedding-mode generation (speaker embedding extracted
|
|
# from the current approved reference wav) is ~17s/utterance with the
|
|
# same voice identity. Default ON; disable via env to force slug mode.
|
|
self.use_embedding_mode = os.environ.get("UMBRA_TTS_EMBEDDING_MODE", "1") != "0"
|
|
|
|
self.data_dir: Path = config.data_dir
|
|
self.status_path: Path = (
|
|
self.data_dir / "chapter_audio" / f"ch{chapter_id}_build_status.json"
|
|
)
|
|
self.result = BuildResult()
|
|
self._start_time: float = 0.0
|
|
self._tts_healthy = True
|
|
self._health_stop = threading.Event()
|
|
self._health_thread: threading.Thread | None = None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Status writing
|
|
# ------------------------------------------------------------------
|
|
|
|
def _write_status(self) -> None:
|
|
elapsed_secs = time.time() - self._start_time
|
|
mins = int(elapsed_secs // 60)
|
|
secs = int(elapsed_secs % 60)
|
|
self.result.elapsed = f"{mins}m {secs:02d}s"
|
|
self.status_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(self.status_path, "w") as f:
|
|
json.dump(self.result.to_dict(), f)
|
|
|
|
def _set_phase(self, phase: str, detail: str, progress: int) -> None:
|
|
self.result.phase = phase
|
|
self.result.detail = detail
|
|
self.result.progress = progress
|
|
self._write_status()
|
|
|
|
# ------------------------------------------------------------------
|
|
# TTS health monitor
|
|
# ------------------------------------------------------------------
|
|
|
|
def _monitor_tts_health(self) -> None:
|
|
import httpx
|
|
|
|
url = f"{self.config.tts.qwen_tts_url}/health"
|
|
consecutive_failures = 0
|
|
while not self._health_stop.is_set():
|
|
try:
|
|
r = httpx.get(url, timeout=5)
|
|
healthy = r.status_code < 500
|
|
if healthy:
|
|
data = r.json()
|
|
healthy = data.get("base_loaded", False)
|
|
except Exception:
|
|
healthy = False
|
|
|
|
if healthy:
|
|
consecutive_failures = 0
|
|
if not self._tts_healthy:
|
|
logger.info("TTS server recovered")
|
|
self._tts_healthy = True
|
|
else:
|
|
consecutive_failures += 1
|
|
logger.warning(
|
|
"TTS health check failed (%d consecutive)", consecutive_failures
|
|
)
|
|
if consecutive_failures >= 5:
|
|
self._tts_healthy = False
|
|
self.result.errors.append(
|
|
f"TTS server unreachable after {consecutive_failures} checks"
|
|
)
|
|
self._write_status()
|
|
break
|
|
self._tts_healthy = False
|
|
|
|
self._health_stop.wait(self.tts_health_interval)
|
|
|
|
def _start_health_monitor(self) -> None:
|
|
self._health_stop.clear()
|
|
self._tts_healthy = True
|
|
self._health_thread = threading.Thread(
|
|
target=self._monitor_tts_health, daemon=True
|
|
)
|
|
self._health_thread.start()
|
|
|
|
def _stop_health_monitor(self) -> None:
|
|
self._health_stop.set()
|
|
if self._health_thread:
|
|
self._health_thread.join(timeout=5)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Phase 1: Preflight
|
|
# ------------------------------------------------------------------
|
|
|
|
def _check_prerequisites(self) -> None:
|
|
self._set_phase("preflight", "Checking prerequisites...", 2)
|
|
|
|
errors: list[str] = []
|
|
|
|
# TTS server reachable
|
|
import httpx
|
|
|
|
url = f"{self.config.tts.qwen_tts_url}/health"
|
|
health = None
|
|
try:
|
|
r = httpx.get(url, timeout=10)
|
|
if r.status_code >= 500:
|
|
errors.append(f"TTS server returned HTTP {r.status_code}")
|
|
else:
|
|
health = r.json()
|
|
except Exception as e:
|
|
errors.append(f"TTS server unreachable: {e}")
|
|
|
|
# Auto-load the TTS base model if it isn't already loaded so that the
|
|
# webUI chapter build is self-sufficient (no manual pre-load step).
|
|
if not errors and health is not None and not health.get("base_loaded"):
|
|
load_url = f"{self.config.tts.qwen_tts_url}/load"
|
|
try:
|
|
self._set_phase("preflight", "Loading TTS base model...", 3)
|
|
rr = httpx.post(load_url, json={"model_type": "base"}, timeout=600)
|
|
if rr.status_code >= 500:
|
|
errors.append(f"TTS server failed to load base model (HTTP {rr.status_code})")
|
|
else:
|
|
hr = httpx.get(url, timeout=10)
|
|
health = hr.json() if hr.status_code < 500 else {}
|
|
except Exception as e:
|
|
errors.append(f"Failed to load TTS base model: {e}")
|
|
if not errors and not health.get("base_loaded"):
|
|
errors.append("TTS base model not loaded")
|
|
|
|
if not errors and not health.get("base_loaded"):
|
|
errors.append("TTS base model not loaded")
|
|
|
|
# Chapter has utterances
|
|
db = sqlite3.connect(str(self.data_dir / "project.db"))
|
|
db.row_factory = sqlite3.Row
|
|
count = db.execute(
|
|
"SELECT COUNT(*) as n FROM utterances WHERE chapter_id = ?",
|
|
(self.chapter_id,),
|
|
).fetchone()["n"]
|
|
db.close()
|
|
if count == 0:
|
|
errors.append(f"No utterances found for chapter {self.chapter_id}")
|
|
|
|
# Voice refs exist for speakers in chapter
|
|
speakers = _load_speakers(self.data_dir)
|
|
alias_map = _build_alias_map(speakers)
|
|
voices_dir = self.data_dir / "voices"
|
|
|
|
db = sqlite3.connect(str(self.data_dir / "project.db"))
|
|
db.row_factory = sqlite3.Row
|
|
rows = db.execute(
|
|
"SELECT DISTINCT speaker_name FROM utterances "
|
|
"WHERE chapter_id = ? AND type = 'dialogue' AND speaker_name IS NOT NULL",
|
|
(self.chapter_id,),
|
|
).fetchall()
|
|
db.close()
|
|
|
|
missing_refs: list[str] = []
|
|
for row in rows:
|
|
spk = row["speaker_name"].strip()
|
|
if spk.lower() == "narrator":
|
|
continue
|
|
ref = _resolve_voice_ref(spk, alias_map, voices_dir)
|
|
if ref is None:
|
|
missing_refs.append(spk)
|
|
if missing_refs:
|
|
errors.append(f"Missing voice refs for: {', '.join(missing_refs)}")
|
|
|
|
# Output directory writable
|
|
audio_dir = self.data_dir / "chapter_audio" / f"ch{self.chapter_id}"
|
|
try:
|
|
audio_dir.mkdir(parents=True, exist_ok=True)
|
|
except OSError as e:
|
|
errors.append(f"Cannot create output directory: {e}")
|
|
|
|
if errors:
|
|
self.result.errors = errors
|
|
self._set_phase("error", f"Preflight failed: {'; '.join(errors)}", 0)
|
|
raise RuntimeError(f"Preflight failed: {errors}")
|
|
|
|
self._set_phase("preflight", "All prerequisites met", 5)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Phase 2: Validate
|
|
# ------------------------------------------------------------------
|
|
|
|
def _accepted_voice_modes(self, voice_ref: Path | None) -> set[str] | None:
|
|
"""Modes the validator accepts for an utterance.
|
|
|
|
A voiced speaker (voice ref present) is current under any voice-identity
|
|
generation mode (``"embedding"`` or ``"icl"``), which tolerates the
|
|
historical mixed-mode chapters while still invalidating a genuinely new
|
|
technology (a mode outside that set only triggers a rebuild when the
|
|
operator regenerates the chapter). An elastic/narrator utterance (no
|
|
voice ref) carries no mode, so the mode check is skipped.
|
|
"""
|
|
if voice_ref is None:
|
|
return None
|
|
return {"embedding", "icl"}
|
|
|
|
def _validate_utterances(self) -> list[dict]:
|
|
self._set_phase("validating", "Checking utterance staleness...", 8)
|
|
|
|
from umbraaudio.generation.staleness import is_current
|
|
from umbraaudio.pronunciation import (
|
|
load_phonetic_names,
|
|
load_proper_names,
|
|
prepare_tts_text,
|
|
)
|
|
from umbraaudio.source_units import ensure_chapter_keys
|
|
|
|
phonetic = load_phonetic_names(self.data_dir)
|
|
proper_names = load_proper_names(self.data_dir)
|
|
speakers = _load_speakers(self.data_dir)
|
|
alias_map = _build_alias_map(speakers)
|
|
voices_dir = self.data_dir / "voices"
|
|
|
|
db = sqlite3.connect(str(self.data_dir / "project.db"))
|
|
db.row_factory = sqlite3.Row
|
|
|
|
# Content-addressed source-unit identity, recorded at generation time per
|
|
# specs/21 §9.1 M4-04. Must run BEFORE the read so the keys are available to the
|
|
# rows below, and BEFORE the marker is written so the audio record can name the
|
|
# exact source unit it came from. Fills only missing keys (cheap, idempotent).
|
|
stamped = ensure_chapter_keys(db, self.chapter_id)
|
|
|
|
utterances = db.execute(
|
|
"SELECT id, utterance_index, text, speaker_name, emotion, pace, instruct, type, "
|
|
"source_unit_key, language_code "
|
|
"FROM utterances WHERE chapter_id = ? ORDER BY utterance_index",
|
|
(self.chapter_id,),
|
|
).fetchall()
|
|
# The source language of the book. An utterance with no language tag
|
|
# is written in this language (spec 31 section 6.1: NULL means the
|
|
# project source language).
|
|
from umbraaudio.translations import source_language
|
|
|
|
project_language = source_language(db)
|
|
db.close()
|
|
if stamped:
|
|
self._set_phase("validating", f"Recorded {stamped} source-unit keys", 8)
|
|
|
|
if not utterances:
|
|
self._set_phase("error", "No utterances found", 0)
|
|
raise RuntimeError("No utterances found")
|
|
|
|
self.result.total_utterances = len(utterances)
|
|
audio_dir = self.data_dir / "chapter_audio" / f"ch{self.chapter_id}"
|
|
audio_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
from umbraaudio.generation.scene_break import is_scene_break, needs_silence
|
|
|
|
missing: list[dict] = []
|
|
for u in utterances:
|
|
idx = u["utterance_index"]
|
|
text = u["text"] or ""
|
|
utt_type = u["type"] or "narration"
|
|
|
|
if not text.strip():
|
|
continue
|
|
|
|
fname = f"utt_{idx:04d}.wav"
|
|
|
|
if is_scene_break(text, utt_type):
|
|
if needs_silence(audio_dir / fname):
|
|
missing.append(dict(u))
|
|
continue
|
|
|
|
spoken_text = prepare_tts_text(text, phonetic, proper_names)
|
|
spk = (u["speaker_name"] or "narrator").strip()
|
|
vref = _resolve_voice_ref(spk, alias_map, voices_dir)
|
|
# The language the line is WRITTEN in. This is the source axis
|
|
# (spec 31), not the target translation: a book in English with a
|
|
# German line inside it is voiced in German for that line.
|
|
line_language = u["language_code"] or project_language
|
|
|
|
if not is_current(
|
|
spoken_text, audio_dir / fname,
|
|
speaker=spk, voice_ref=vref,
|
|
accepted_voice_modes=self._accepted_voice_modes(vref),
|
|
language=line_language,
|
|
):
|
|
missing.append(dict(u))
|
|
|
|
self.result.skipped = self.result.total_utterances - len(missing)
|
|
detail = f"{len(missing)} utterances need generation"
|
|
self._set_phase("validating", detail, 12)
|
|
return missing
|
|
|
|
# ------------------------------------------------------------------
|
|
# Phase 3: Generate
|
|
# ------------------------------------------------------------------
|
|
|
|
def _generate_utterances(self, missing: list[dict]) -> None:
|
|
if not missing:
|
|
self._set_phase("generating", "All utterances current", 70)
|
|
return
|
|
|
|
from umbraaudio.generation.audio_effects import (
|
|
apply_effects_to_file,
|
|
apply_effects_to_file_by_name,
|
|
load_speaker_distortion_effects,
|
|
load_speaker_traits,
|
|
)
|
|
from umbraaudio.generation.chapter import compose_utterance_instruct
|
|
from umbraaudio.generation.scene_break import (
|
|
is_scene_break,
|
|
write_scene_break_audio,
|
|
)
|
|
from umbraaudio.generation.staleness import write_marker
|
|
from umbraaudio.pronunciation import (
|
|
load_phonetic_names,
|
|
load_proper_names,
|
|
prepare_tts_text,
|
|
)
|
|
from umbraaudio.voices.qwen_tts import (
|
|
VOICE_REFERENCE_TEXT,
|
|
QwenTTSEngine,
|
|
QwenTTSUnavailableError,
|
|
)
|
|
|
|
self._set_phase("generating", f"Generating {len(missing)} utterances...", 15)
|
|
|
|
from umbraaudio.translations import source_language
|
|
|
|
_lang_db = sqlite3.connect(str(self.data_dir / "project.db"))
|
|
project_language = source_language(_lang_db)
|
|
_lang_db.close()
|
|
|
|
phonetic = load_phonetic_names(self.data_dir)
|
|
proper_names = load_proper_names(self.data_dir)
|
|
speakers = _load_speakers(self.data_dir)
|
|
alias_map = _build_alias_map(speakers)
|
|
voices_dir = self.data_dir / "voices"
|
|
audio_dir = self.data_dir / "chapter_audio" / f"ch{self.chapter_id}"
|
|
|
|
speaker_traits = load_speaker_traits(self.data_dir / "speakers.json")
|
|
distortion_effects = load_speaker_distortion_effects(self.data_dir / "speakers.json")
|
|
|
|
engine = QwenTTSEngine(qwen_tts_url=self.config.tts.qwen_tts_url, helper_url=self.config.tts.qwen_tts_helper_url)
|
|
|
|
# Capture the full vmatrix (the "8k matrix" ref_code + speaker embedding)
|
|
# once per distinct speaker, cached server-side as .pt under PROMPT_DIR.
|
|
# x_vector_only=False (explicit, and the engine default) captures the FULL
|
|
# vmatrix so the approved voice's exact numbers are persisted as the
|
|
# canonical record. Generation reuses that identity via embedding-mode
|
|
# (~17s/utt); full-vmatrix ICL generation is >15 min/utt, see
|
|
# use_embedding_mode.
|
|
prompt_map: dict[str, str] = {}
|
|
for u in missing:
|
|
if is_scene_break(u.get("text"), u.get("type")):
|
|
continue
|
|
spk = (u.get("speaker_name") or "narrator").strip()
|
|
if spk in prompt_map:
|
|
continue
|
|
vref = _resolve_voice_ref(spk, alias_map, voices_dir)
|
|
if not vref:
|
|
continue
|
|
prompt_map[spk] = engine.build_voice_prompt(
|
|
vref,
|
|
ref_text=VOICE_REFERENCE_TEXT,
|
|
x_vector_only=False,
|
|
)
|
|
|
|
self._start_health_monitor()
|
|
|
|
try:
|
|
for i, u in enumerate(missing):
|
|
idx = u["utterance_index"]
|
|
text = u["text"] or ""
|
|
speaker = (u["speaker_name"] or "narrator").strip()
|
|
emotion = u["emotion"] or "neutral"
|
|
instruct = u["instruct"] or ""
|
|
|
|
pct = int(15 + (i / max(len(missing), 1)) * 55)
|
|
self._set_phase(
|
|
"generating",
|
|
f"Generating {i + 1}/{len(missing)}: {speaker}",
|
|
pct,
|
|
)
|
|
|
|
if not text.strip():
|
|
continue
|
|
|
|
out = audio_dir / f"utt_{idx:04d}.wav"
|
|
|
|
if is_scene_break(text, u["type"]):
|
|
write_scene_break_audio(out)
|
|
self.result.generated += 1
|
|
continue
|
|
|
|
spoken_text = prepare_tts_text(text, phonetic, proper_names)
|
|
voice_ref = _resolve_voice_ref(speaker, alias_map, voices_dir)
|
|
# The language the line is written in (spec 31 source axis).
|
|
# NULL on the utterance means the project's source language.
|
|
line_language = u.get("language_code") or project_language
|
|
|
|
tts_instruct = compose_utterance_instruct(
|
|
instruct, emotion, u["pace"]
|
|
)
|
|
|
|
seed = (self.chapter_id * 10000 + idx) if self.use_seed else None
|
|
success = False
|
|
|
|
for attempt in range(self.max_retries):
|
|
# Wait for TTS health
|
|
wait_count = 0
|
|
while not self._tts_healthy and wait_count < 60:
|
|
self._set_phase(
|
|
"generating",
|
|
f"Waiting for TTS server... (utt {idx}, attempt {attempt + 1})",
|
|
pct,
|
|
)
|
|
time.sleep(5)
|
|
wait_count += 1
|
|
|
|
if not self._tts_healthy:
|
|
self.result.errors.append(
|
|
f"TTS server down, skipping utt_{idx:04d}"
|
|
)
|
|
break
|
|
|
|
try:
|
|
prompt_slug = prompt_map.get(speaker)
|
|
voice_mode = None
|
|
if voice_ref and self.use_embedding_mode:
|
|
engine.generate(
|
|
text=spoken_text,
|
|
output_path=out,
|
|
voice_ref=voice_ref,
|
|
ref_text=VOICE_REFERENCE_TEXT,
|
|
instruct=tts_instruct,
|
|
temperature=self.temperature,
|
|
seed=seed,
|
|
language=line_language,
|
|
)
|
|
voice_mode = "embedding"
|
|
elif prompt_slug:
|
|
engine.generate(
|
|
text=spoken_text,
|
|
output_path=out,
|
|
prompt_slug=prompt_slug,
|
|
instruct=tts_instruct,
|
|
temperature=self.temperature,
|
|
seed=seed,
|
|
language=line_language,
|
|
)
|
|
voice_mode = "icl"
|
|
elif voice_ref:
|
|
engine.generate(
|
|
text=spoken_text,
|
|
output_path=out,
|
|
voice_ref=voice_ref,
|
|
ref_text=VOICE_REFERENCE_TEXT,
|
|
instruct=tts_instruct,
|
|
temperature=self.temperature,
|
|
seed=seed,
|
|
language=line_language,
|
|
)
|
|
voice_mode = "embedding"
|
|
else:
|
|
engine.generate(
|
|
text=spoken_text,
|
|
output_path=out,
|
|
instruct=tts_instruct or "A calm, neutral voice speaking clearly.",
|
|
temperature=self.temperature,
|
|
seed=seed,
|
|
language=line_language,
|
|
)
|
|
|
|
if out.exists():
|
|
explicit = distortion_effects.get(speaker, [])
|
|
applied: list[str] = []
|
|
if explicit:
|
|
apply_effects_to_file_by_name(out, explicit)
|
|
applied = list(explicit)
|
|
else:
|
|
traits = speaker_traits.get(speaker, [])
|
|
if traits:
|
|
apply_effects_to_file(out, traits)
|
|
applied = list(traits)
|
|
|
|
# Marker written AFTER post-processing, so the recorded
|
|
# sha256 is the final bytes, and `effects` records what
|
|
# turned the raw TTS output into them.
|
|
write_marker(
|
|
spoken_text, out,
|
|
speaker=speaker, voice_ref=voice_ref,
|
|
voice_mode=voice_mode, instruct=tts_instruct,
|
|
source_unit_key=u.get("source_unit_key"),
|
|
effects=applied,
|
|
language=line_language,
|
|
)
|
|
self.result.generated += 1
|
|
success = True
|
|
break
|
|
|
|
except QwenTTSUnavailableError as e:
|
|
logger.warning(
|
|
"TTS unavailable for ch%d utt%d (attempt %d): %s",
|
|
self.chapter_id, idx, attempt + 1, e,
|
|
)
|
|
self._tts_healthy = False
|
|
time.sleep(5 * (attempt + 1))
|
|
except Exception as e:
|
|
logger.error(
|
|
"Generation failed for ch%d utt%d (attempt %d): %s",
|
|
self.chapter_id, idx, attempt + 1, e,
|
|
)
|
|
time.sleep(2 * (attempt + 1))
|
|
|
|
if not success:
|
|
self.result.failed.append(idx)
|
|
logger.error("All attempts exhausted for utt_%04d", idx)
|
|
|
|
finally:
|
|
self._stop_health_monitor()
|
|
|
|
if self.result.failed:
|
|
self.result.errors.append(
|
|
f"Failed utterances: {self.result.failed}"
|
|
)
|
|
self._set_phase(
|
|
"error",
|
|
f"Generation incomplete: {len(self.result.failed)} failures",
|
|
0,
|
|
)
|
|
raise RuntimeError(f"Generation failed for {len(self.result.failed)} utterances")
|
|
|
|
self._set_phase("generating", "All utterances generated", 70)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Phase 4: Stitch
|
|
# ------------------------------------------------------------------
|
|
|
|
def _stitch_chapter(self) -> None:
|
|
self._set_phase("stitching", "Stitching chapter audio...", 78)
|
|
|
|
from umbraaudio.assembly import AudioStitcher
|
|
|
|
audio_dir = self.data_dir / "chapter_audio" / f"ch{self.chapter_id}"
|
|
|
|
# Only stitch audio that belongs to a real, non-blank utterance.
|
|
# Leftover wavs for blank/stale utterances (text cleared or empty)
|
|
# must not be spliced into the chapter — silence/artifacts.
|
|
def _idx(p: Path) -> int:
|
|
try:
|
|
return int(p.stem.split("_")[1])
|
|
except (IndexError, ValueError):
|
|
return -1
|
|
|
|
valid_idx: set[int] = set()
|
|
db = sqlite3.connect(str(self.data_dir / "project.db"))
|
|
db.row_factory = sqlite3.Row
|
|
for u in db.execute(
|
|
"SELECT utterance_index, text, type FROM utterances WHERE chapter_id = ?",
|
|
(self.chapter_id,),
|
|
):
|
|
t = (u["text"] or "").strip()
|
|
if t:
|
|
valid_idx.add(u["utterance_index"])
|
|
db.close()
|
|
|
|
audio_files = sorted(
|
|
(p for p in audio_dir.glob("utt_*.wav") if _idx(p) in valid_idx),
|
|
key=_idx,
|
|
)
|
|
|
|
if not audio_files:
|
|
self._set_phase("error", "No audio files to stitch", 0)
|
|
raise RuntimeError("No audio files to stitch")
|
|
|
|
stitcher = AudioStitcher()
|
|
output_dir = self.data_dir / "chapter_audio"
|
|
output_path = output_dir / f"ch{self.chapter_id}_full.wav"
|
|
stitcher.stitch_utterances(audio_files, output_path)
|
|
|
|
self.result.chapter_wav = output_path.name
|
|
self._set_phase("stitching", f"Stitched {len(audio_files)} files → {output_path.name}", 85)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Phase 5: Export MP3
|
|
# ------------------------------------------------------------------
|
|
|
|
def _export_mp3(self) -> None:
|
|
self._set_phase("exporting", "Converting to AR-compliant MP3...", 88)
|
|
|
|
wav_path = (
|
|
self.data_dir / "chapter_audio" / f"ch{self.chapter_id}_full.wav"
|
|
)
|
|
if not wav_path.exists():
|
|
self._set_phase("exporting", "WAV not found, skipping MP3", 88)
|
|
return
|
|
|
|
mp3_path = wav_path.with_suffix(".mp3")
|
|
try:
|
|
from umbraaudio.assembly.ar_export import export_chapter_mp3
|
|
|
|
stats = export_chapter_mp3(self.data_dir, wav_path, mp3_path)
|
|
if mp3_path.exists():
|
|
self.result.chapter_mp3 = mp3_path.name
|
|
detail = f"AR MP3 exported: {mp3_path.name} ({stats['duration_s']}s"
|
|
if stats.get("rms_db") is not None:
|
|
detail += f", RMS {stats['rms_db']:.1f} dB"
|
|
detail += ")"
|
|
if stats.get("warning"):
|
|
detail += f" ⚠ {stats['warning']}"
|
|
self._set_phase("exporting", detail, 92)
|
|
else:
|
|
logger.warning("ffmpeg MP3 conversion failed")
|
|
self._set_phase("exporting", "MP3 conversion failed (non-fatal)", 92)
|
|
except FileNotFoundError:
|
|
logger.warning("ffmpeg not found, skipping MP3 export")
|
|
self._set_phase("exporting", "ffmpeg not available, skipping MP3", 92)
|
|
except Exception as e:
|
|
logger.warning("ffmpeg MP3 conversion failed: %s", e)
|
|
self._set_phase("exporting", f"MP3 conversion failed: {e}", 92)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Phase 6: Verify
|
|
# ------------------------------------------------------------------
|
|
|
|
def _verify_quality(self) -> None:
|
|
self._set_phase("verifying", "Running quality checks...", 94)
|
|
|
|
from umbraaudio.assembly import ChapterQualityChecker
|
|
|
|
speakers_path = self.data_dir / "speakers.json"
|
|
known_speakers: list[str] = []
|
|
if speakers_path.exists():
|
|
with open(speakers_path) as f:
|
|
data = json.load(f)
|
|
known_speakers = [s["primary_name"] for s in data.get("speakers", [])]
|
|
|
|
checker = ChapterQualityChecker(self.data_dir, known_speakers)
|
|
issues = checker.check_chapter(self.chapter_id)
|
|
self.result.quality_issues = [i.to_dict() for i in issues]
|
|
|
|
errors = sum(1 for i in issues if i.severity == "error")
|
|
warnings = sum(1 for i in issues if i.severity == "warning")
|
|
|
|
if errors:
|
|
self.result.phase = "warning"
|
|
self.result.detail = f"Completed with {errors} errors, {warnings} warnings"
|
|
self.result.progress = 100
|
|
elif warnings:
|
|
self.result.detail = f"Completed with {warnings} warnings"
|
|
else:
|
|
self.result.detail = "All quality checks passed"
|
|
|
|
self._set_phase(
|
|
self.result.phase if self.result.phase == "warning" else "complete",
|
|
self.result.detail,
|
|
100,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Main entry point
|
|
# ------------------------------------------------------------------
|
|
|
|
def run(self) -> BuildResult:
|
|
"""Run the full chapter build pipeline.
|
|
|
|
Returns the :class:`BuildResult` with final status.
|
|
"""
|
|
self._start_time = time.time()
|
|
self.result.started_at = datetime.now(UTC).isoformat()
|
|
|
|
try:
|
|
self._check_prerequisites()
|
|
missing = self._validate_utterances()
|
|
self._generate_utterances(missing)
|
|
self._stitch_chapter()
|
|
self._export_mp3()
|
|
self._verify_quality()
|
|
except Exception as e:
|
|
if self.result.phase != "error":
|
|
self._set_phase("error", str(e), 0)
|
|
logger.error("Chapter %d build failed: %s", self.chapter_id, e)
|
|
|
|
self._write_status()
|
|
return self.result
|