1122 lines
44 KiB
Python
1122 lines
44 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
from __future__ import annotations
|
|
|
|
import gc
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import numpy as np
|
|
import torch
|
|
|
|
from umbraaudio.config import load_config
|
|
from umbraaudio.database import Database
|
|
|
|
if TYPE_CHECKING:
|
|
from umbraaudio.config import AppConfig
|
|
from umbraaudio.pipeline_manager import PipelineManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
NOISE_CHARACTERS = {
|
|
"people",
|
|
"everyone",
|
|
"someone",
|
|
"anyone",
|
|
"nobody",
|
|
"something",
|
|
"nothing",
|
|
"everything",
|
|
"the fighters",
|
|
"the fugitives",
|
|
"the pirates",
|
|
"Warbots",
|
|
"the Warbots",
|
|
"pirates",
|
|
}
|
|
|
|
|
|
def _author_watermark_bits(author_name: str, length: int = 128) -> np.ndarray:
|
|
digest = hashlib.sha256(author_name.encode()).digest()
|
|
bits = []
|
|
for byte in digest:
|
|
for i in range(7, -1, -1):
|
|
bits.append((byte >> i) & 1)
|
|
if len(bits) >= length:
|
|
return np.array(bits[:length], dtype=np.float32)
|
|
return np.array(bits[:length], dtype=np.float32)
|
|
|
|
|
|
ALIAS_MAP = {
|
|
"Hunter": "Deal Hunter",
|
|
"Captain Weston": "Captain",
|
|
"The captain": "Captain",
|
|
"the captain": "Captain",
|
|
"Princess": "Princess Kainda",
|
|
"the princess": "Princess Kainda",
|
|
"Kainda": "Princess Kainda",
|
|
"Vicky": "Aunt Vicky",
|
|
"Aunt Victoria": "Aunt Vicky",
|
|
"Victoria": "Aunt Vicky",
|
|
"the king": "Adric",
|
|
}
|
|
|
|
|
|
def _merge_characters(raw_characters: list[dict], quotes: list[dict]) -> list[dict]:
|
|
pronouns = {
|
|
"i",
|
|
"you",
|
|
"he",
|
|
"she",
|
|
"it",
|
|
"we",
|
|
"they",
|
|
"me",
|
|
"him",
|
|
"her",
|
|
"us",
|
|
"them",
|
|
"my",
|
|
"your",
|
|
"his",
|
|
"its",
|
|
"our",
|
|
"their",
|
|
}
|
|
|
|
merged: dict[str, dict] = {}
|
|
for char in raw_characters:
|
|
name = char["name"]
|
|
if name.lower() in pronouns:
|
|
continue
|
|
if name.lower() in NOISE_CHARACTERS:
|
|
continue
|
|
if len(name) <= 1:
|
|
continue
|
|
if name not in merged:
|
|
merged[name] = {"name": name, "mentions": 0, "context_snippets": []}
|
|
merged[name]["mentions"] += 1
|
|
|
|
for q in quotes:
|
|
speaker = q.get("speaker", "")
|
|
if speaker and speaker in merged:
|
|
merged[speaker]["context_snippets"].append(q.get("quote_text", "")[:200])
|
|
|
|
alias_groups: dict[str, list[str]] = {}
|
|
canonical: dict[str, str] = {}
|
|
for name in list(merged.keys()):
|
|
normalized = name.strip().lower()
|
|
if normalized not in alias_groups:
|
|
alias_groups[normalized] = []
|
|
alias_groups[normalized].append(name)
|
|
|
|
for _normalized, variants in alias_groups.items():
|
|
variants.sort(key=lambda n: -merged[n]["mentions"])
|
|
primary = variants[0]
|
|
for v in variants:
|
|
canonical[v] = primary
|
|
if len(variants) > 1:
|
|
merged[primary]["aliases"] = list(
|
|
set(merged[primary].get("aliases", []) + variants[1:])
|
|
)
|
|
|
|
deduped: dict[str, dict] = {}
|
|
for name, data in merged.items():
|
|
primary = canonical.get(name, name)
|
|
if primary not in deduped:
|
|
deduped[primary] = {
|
|
"name": primary,
|
|
"mentions": data["mentions"],
|
|
"context_snippets": list(data.get("context_snippets", [])),
|
|
"aliases": list(data.get("aliases", [])),
|
|
}
|
|
else:
|
|
deduped[primary]["mentions"] += data["mentions"]
|
|
if "context_snippets" in data:
|
|
deduped[primary]["context_snippets"].extend(data["context_snippets"])
|
|
if "aliases" in data:
|
|
existing = set(deduped[primary].get("aliases", []))
|
|
existing.update(data.get("aliases", []))
|
|
deduped[primary]["aliases"] = list(existing)
|
|
|
|
alias_keys_lower = {k.lower(): v for k, v in ALIAS_MAP.items()}
|
|
deduped_names = list(deduped.keys())
|
|
for name in deduped_names:
|
|
name_lower = name.lower()
|
|
if name_lower in alias_keys_lower:
|
|
canonical_name = alias_keys_lower[name_lower]
|
|
if canonical_name != name and canonical_name in {n.lower() for n in deduped}:
|
|
actual_canonical = next(n for n in deduped if n.lower() == canonical_name)
|
|
if actual_canonical != name:
|
|
deduped[actual_canonical]["mentions"] += deduped[name]["mentions"]
|
|
deduped[actual_canonical]["context_snippets"].extend(deduped[name].get("context_snippets", []))
|
|
existing = set(deduped[actual_canonical].get("aliases", []))
|
|
existing.add(name)
|
|
existing.update(deduped[name].get("aliases", []))
|
|
deduped[actual_canonical]["aliases"] = list(existing)
|
|
del deduped[name]
|
|
|
|
return sorted(deduped.values(), key=lambda c: -c["mentions"])
|
|
|
|
|
|
async def run_pipeline_background(
|
|
pm: PipelineManager,
|
|
cfg: AppConfig,
|
|
chapters: str | None = None,
|
|
author_name: str = "B.D. Murphy",
|
|
) -> None:
|
|
from umbraaudio.pipeline_manager import PipelineState, PipelineStep
|
|
|
|
pm.update(
|
|
step=PipelineStep.DISCOVER,
|
|
state=PipelineState.RUNNING,
|
|
detail="Starting discovery...",
|
|
)
|
|
|
|
try:
|
|
db = Database(cfg.db_path)
|
|
await db.connect()
|
|
|
|
from umbraaudio.voices.registry import VoiceRegistry
|
|
|
|
registry = VoiceRegistry(cfg.voice_registry_path)
|
|
await registry.connect()
|
|
|
|
pid = await db.active_project_id()
|
|
project = await db.get_project(pid)
|
|
if not project:
|
|
pm.update(state=PipelineState.FAILED, error="No project found")
|
|
return
|
|
|
|
chapters_db = await db.get_chapters(pid)
|
|
if not chapters_db:
|
|
pm.update(state=PipelineState.FAILED, error="No chapters found")
|
|
return
|
|
|
|
pm.update(detail=f"Project: {project['name']}, {len(chapters_db)} chapters")
|
|
|
|
from umbraaudio.voices.qwen_tts import QwenTTSEngine
|
|
from umbraaudio.voices.unified_model_manager import (
|
|
get_unified_model_manager,
|
|
)
|
|
|
|
tts = QwenTTSEngine(
|
|
device=cfg.tts.device,
|
|
sample_rate=cfg.tts.sample_rate,
|
|
voicedesign_model=cfg.tts.voicedesign_model,
|
|
base_model=cfg.tts.base_model,
|
|
qwen_tts_url=cfg.tts.qwen_tts_url,
|
|
)
|
|
|
|
model_manager = None
|
|
if cfg.tts.qwen_tts_url:
|
|
model_manager = await get_unified_model_manager(
|
|
cfg.tts.qwen_tts_url,
|
|
cfg.llm.lmstudio_base_url.replace("/v1", ""),
|
|
cfg.llm.lmstudio_model,
|
|
)
|
|
await model_manager.for_analysis() # Start with no models loaded
|
|
|
|
if not pm.should_skip_step(PipelineStep.DISCOVER):
|
|
from umbraaudio.analysis.booknlp_engine import BookNLPEngine
|
|
from umbraaudio.analysis.character_profiler import CharacterProfiler
|
|
|
|
pm.update(step=PipelineStep.DISCOVER, detail="Analyzing characters...", progress=10)
|
|
full_text = "\n\n".join(ch["source_text"] for ch in chapters_db)
|
|
booknlp = BookNLPEngine()
|
|
analysis_dir = cfg.data_dir / "analysis"
|
|
try:
|
|
result = booknlp.analyze(full_text, analysis_dir)
|
|
except Exception as e:
|
|
logger.warning("BookNLP failed: %s. Using fallback.", e)
|
|
result = {"characters": [], "quotes": []}
|
|
|
|
raw_characters = result.get("characters", [])
|
|
quotes = result.get("quotes", [])
|
|
|
|
merged = _merge_characters(raw_characters, quotes)
|
|
char_names = [c["name"] for c in merged[:30]]
|
|
|
|
profiler = CharacterProfiler(cfg.llm)
|
|
for i, cname in enumerate(char_names):
|
|
ctx_text = "\n".join(
|
|
merged[
|
|
next((j for j, c in enumerate(merged) if c["name"] == cname), 0)
|
|
].get("context_snippets", [])
|
|
)
|
|
if not ctx_text:
|
|
for ch in chapters_db:
|
|
text = ch["source_text"]
|
|
idx = text.find(cname)
|
|
if idx >= 0:
|
|
start = max(0, idx - 100)
|
|
end = min(len(text), idx + 200)
|
|
ctx_text = text[start:end]
|
|
break
|
|
|
|
existing_chars = await db.get_characters(pid)
|
|
existing_names = {c["name"] for c in existing_chars}
|
|
if cname in existing_names:
|
|
continue
|
|
|
|
voice_registry_entry = await registry.find_by_name(cname)
|
|
profile = None
|
|
|
|
if voice_registry_entry:
|
|
from umbraaudio.analysis import CharacterProfile
|
|
|
|
profile = CharacterProfile(
|
|
name=cname,
|
|
voice_description=voice_registry_entry.get("voice_description"),
|
|
is_narrator=(cname == char_names[0] if char_names else False),
|
|
)
|
|
registry_id = voice_registry_entry["id"]
|
|
else:
|
|
if model_manager:
|
|
await model_manager.for_lm_studio()
|
|
profile = await profiler.profile_character(
|
|
name=cname,
|
|
aliases=[],
|
|
context=ctx_text or f"Character {cname}.",
|
|
)
|
|
|
|
char_id = await db.insert_character(
|
|
project_id=pid,
|
|
name=cname,
|
|
registry_id=registry_id if voice_registry_entry else None,
|
|
)
|
|
|
|
if not voice_registry_entry:
|
|
registry_char = await registry.find_or_create(
|
|
profile, source_project=project["name"]
|
|
)
|
|
registry_id = registry_char[0]
|
|
await db.conn.execute(
|
|
"UPDATE characters SET registry_id = ? WHERE id = ?",
|
|
(registry_id, char_id),
|
|
)
|
|
|
|
vd = profile.voice_description
|
|
if isinstance(vd, dict):
|
|
vd = json.dumps(vd)
|
|
await db.update_character_profile(
|
|
char_id,
|
|
profile_json=profile.to_dict(),
|
|
voice_description=vd,
|
|
)
|
|
|
|
pct = int(10 + ((i + 1) / len(char_names)) * 40)
|
|
pm.update(
|
|
detail=f"Profiling character {i + 1}/{len(char_names)}: {cname}",
|
|
progress=pct,
|
|
)
|
|
|
|
pm.mark_chapter_complete(0)
|
|
pm.update(detail=f"Discovered {len(char_names)} characters", progress=50)
|
|
|
|
pm.update(step=PipelineStep.VOICES, detail="Generating voice references...", progress=0)
|
|
|
|
if not pm.should_skip_step(PipelineStep.VOICES):
|
|
if model_manager:
|
|
await model_manager.for_qwen_voice_design()
|
|
from umbraaudio.voices.voice_designer import VoiceDesigner
|
|
|
|
designer = VoiceDesigner(cfg.tts)
|
|
characters_db = await db.get_characters(pid)
|
|
characters_to_voice = [
|
|
c for c in characters_db if not c["is_narrator"] and c["voice_description"]
|
|
]
|
|
|
|
voices_dir = cfg.voices_dir
|
|
voices_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
for i, char in enumerate(characters_to_voice):
|
|
if char.get("voice_ref_path") and Path(char["voice_ref_path"]).exists():
|
|
continue
|
|
try:
|
|
from umbraaudio.analysis import CharacterProfile
|
|
|
|
cp = CharacterProfile(
|
|
name=char["name"],
|
|
voice_description=char["voice_description"],
|
|
is_narrator=bool(char["is_narrator"]),
|
|
energy=char.get("energy", "measured"),
|
|
pitch_register=char.get("pitch_register", "mid"),
|
|
timbre=char.get("timbre", "clear"),
|
|
)
|
|
ref_path = designer.generate_voice_reference(cp, voices_dir)
|
|
embedding = designer.extract_embedding(ref_path)
|
|
await db.update_character_voice(char["id"], str(ref_path), embedding)
|
|
except Exception as e:
|
|
logger.error("Voice ref failed for %s: %s", char["name"], e)
|
|
|
|
pct = int(((i + 1) / len(characters_to_voice)) * 100)
|
|
pm.update(
|
|
detail=f"Generating voice {i + 1}/{len(characters_to_voice)}: {char['name']}",
|
|
progress=pct,
|
|
)
|
|
|
|
pm.mark_chapter_complete(1)
|
|
pm.update(detail="Voice references complete", progress=100)
|
|
|
|
pm.update(step=PipelineStep.SEGMENT, detail="Segmenting utterances...", progress=0)
|
|
|
|
characters_db = await db.get_characters(pid)
|
|
|
|
if not pm.should_skip_step(PipelineStep.SEGMENT):
|
|
if model_manager:
|
|
await model_manager.for_segmentation()
|
|
from umbraaudio.mapping.manifest import build_generation_manifest, save_manifest
|
|
from umbraaudio.mapping.segmenter import Segmenter
|
|
|
|
segmenter = Segmenter()
|
|
characters_map = {}
|
|
for c in characters_db:
|
|
characters_map[c["name"]] = {
|
|
"voice_ref_path": c.get("voice_ref_path"),
|
|
"voice_description": c.get("voice_description"),
|
|
}
|
|
|
|
pronunciation_guide = await db.get_pronunciation_guide(pid)
|
|
char_names_no_narrator = [c["name"] for c in characters_db if not c["is_narrator"]]
|
|
alias_keys = [k for k in ALIAS_MAP if k.lower() not in {n.lower() for n in char_names_no_narrator}]
|
|
char_names_no_narrator.extend(alias_keys)
|
|
|
|
total_ch = len(chapters_db)
|
|
for idx, ch in enumerate(chapters_db):
|
|
utterances = segmenter.segment(
|
|
ch["source_text"],
|
|
ch["id"],
|
|
char_names_no_narrator,
|
|
chapter_title=ch.get("title"),
|
|
)
|
|
for u in utterances:
|
|
if u.get("character_name") is None:
|
|
u["character_name"] = "narrator"
|
|
canonical = ALIAS_MAP.get(u["character_name"])
|
|
if canonical:
|
|
u["character_name"] = canonical
|
|
manifest = build_generation_manifest(
|
|
ch["id"], utterances, characters_map, pronunciation_guide,
|
|
)
|
|
manifest_path = cfg.manifests_dir / f"ch{ch['id']}.json"
|
|
save_manifest(manifest, manifest_path)
|
|
pm.update_chapter_progress(ch["id"], total_ch, idx + 1)
|
|
|
|
pm.mark_chapter_complete(2)
|
|
pm.update(detail="Segmentation complete", progress=100)
|
|
|
|
# Pass 4: LLM Speaker Attribution with multi-pass context analysis
|
|
pm.update(step=PipelineStep.ATTRIBUTE, detail="Running LLM speaker attribution...", progress=0)
|
|
|
|
if not pm.should_skip_step(PipelineStep.ATTRIBUTE):
|
|
if model_manager:
|
|
await model_manager.for_analysis()
|
|
from scripts.run_llm_attribution import run_llm_attribution
|
|
|
|
attrib_result = await run_llm_attribution()
|
|
pm.update(
|
|
detail=f"Attribution complete: {attrib_result.get('attributed', 0)} utterances, "
|
|
f"{attrib_result.get('exceptions_found', 0)} exceptions",
|
|
progress=100,
|
|
)
|
|
|
|
# Check quality gate
|
|
if not attrib_result.get("quality_gate_passed", True):
|
|
pm.update(
|
|
detail=f"Quality gate FAILED: {attrib_result.get('exception_rate', 0):.2f}% exceptions",
|
|
state=PipelineState.FAILED,
|
|
error=f"Exception rate {attrib_result.get('exception_rate', 0):.2f}% exceeds 1% threshold",
|
|
)
|
|
logger.error(
|
|
"Quality gate failed: %.2f%% exceptions exceeds 1%% threshold",
|
|
attrib_result.get("exception_rate", 0),
|
|
)
|
|
return
|
|
|
|
pm.update(step=PipelineStep.GENERATE, detail="Generating audio...", progress=0)
|
|
|
|
if not pm.should_skip_step(PipelineStep.GENERATE):
|
|
if model_manager:
|
|
await model_manager.for_generation(needs_clone=True)
|
|
from umbraaudio.assembly.normalize import AudioNormalizer
|
|
from umbraaudio.generation.cache import GenerationCache
|
|
from umbraaudio.generation.chapter import ChapterGenerator
|
|
from umbraaudio.mapping.manifest import load_manifest
|
|
|
|
cache = GenerationCache(cfg.data_dir / "cache")
|
|
normalizer = AudioNormalizer(
|
|
target_rms_db=cfg.audio.target_rms_db,
|
|
peak_limit_db=cfg.audio.peak_limit_db,
|
|
sample_rate=cfg.tts.sample_rate,
|
|
)
|
|
generator = ChapterGenerator(
|
|
tts, cfg, cache=cache, normalizer=normalizer,
|
|
watermark_bits=_author_watermark_bits(author_name),
|
|
author_name=author_name,
|
|
)
|
|
|
|
speaker_profiles = {}
|
|
for c in characters_db:
|
|
speaker_profiles[c["name"]] = {
|
|
"energy": c.get("energy", "measured"),
|
|
"pitch_register": c.get("pitch_register", "mid"),
|
|
"timbre": c.get("timbre", "clear"),
|
|
}
|
|
|
|
total_gen = 0
|
|
for _i, ch in enumerate(chapters_db):
|
|
manifest_path = cfg.manifests_dir / f"ch{ch['id']}.json"
|
|
if not manifest_path.exists():
|
|
continue
|
|
manifest = load_manifest(manifest_path)
|
|
total_gen += len(manifest.get("utterances", []))
|
|
|
|
generated_count = 0
|
|
for i, ch in enumerate(chapters_db):
|
|
manifest_path = cfg.manifests_dir / f"ch{ch['id']}.json"
|
|
if not manifest_path.exists():
|
|
continue
|
|
manifest = load_manifest(manifest_path)
|
|
|
|
chapter_audio_dir = cfg.audio_dir / f"ch{ch['id']}"
|
|
chapter_audio_dir.mkdir(parents=True, exist_ok=True)
|
|
normalized_dir = cfg.data_dir / "audio_normalized" / f"ch{ch['id']}"
|
|
normalized_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
result = generator.generate_chapter(
|
|
manifest, chapter_audio_dir, normalized_dir,
|
|
speaker_profiles=speaker_profiles,
|
|
)
|
|
generated_count += len(result.get("generated_files", []))
|
|
|
|
gc.collect()
|
|
if torch.cuda.is_available():
|
|
torch.cuda.empty_cache()
|
|
|
|
pm.update_chapter_progress(ch["id"], len(chapters_db), i + 1)
|
|
|
|
pm.mark_chapter_complete(3)
|
|
pm.update(detail=f"Generated {generated_count} utterances", progress=100)
|
|
|
|
pm.update(step=PipelineStep.EXPORT, detail="Exporting audiobook...", progress=0)
|
|
|
|
if not pm.should_skip_step(PipelineStep.EXPORT):
|
|
from umbraaudio.assembly.export import AudiobookExporter
|
|
|
|
exporter = AudiobookExporter()
|
|
audio_dir = cfg.audio_dir
|
|
if not audio_dir.exists():
|
|
audio_dir = cfg.data_dir / "chapter_audio"
|
|
|
|
chapter_files = sorted(audio_dir.glob("*/utt_*.wav"))
|
|
|
|
if chapter_files:
|
|
m4b_path = cfg.export_dir / "audiobook.m4b"
|
|
try:
|
|
exporter.export_m4b(
|
|
chapter_files, m4b_path,
|
|
title=project["name"], author=author_name,
|
|
)
|
|
except Exception as e:
|
|
logger.error("M4B export failed: %s", e)
|
|
|
|
mp3_files = exporter.export_mp3(chapter_files, cfg.export_dir / "mp3")
|
|
pm.update(detail=f"Exported M4B + {len(mp3_files)} MP3", progress=100)
|
|
else:
|
|
pm.update(detail="No audio files to export", progress=100)
|
|
|
|
pm.mark_chapter_complete(4)
|
|
|
|
if model_manager:
|
|
await model_manager.cleanup()
|
|
await registry.close()
|
|
await db.close()
|
|
|
|
except Exception as e:
|
|
logger.error("Pipeline failed: %s", e)
|
|
pm.update(state=PipelineState.FAILED, error=str(e))
|
|
return
|
|
|
|
pm.update(state=PipelineState.COMPLETED, detail="Pipeline complete", progress=100)
|
|
|
|
|
|
async def run_pipeline(
|
|
project_dir: Path | str,
|
|
chapters: str | None = None,
|
|
author_name: str = "B.D. Murphy",
|
|
allow_voice_collisions: bool = False,
|
|
) -> dict[str, Any]:
|
|
cfg = load_config(project_dir)
|
|
db = Database(cfg.db_path)
|
|
await db.connect()
|
|
|
|
from umbraaudio.voices.registry import VoiceRegistry
|
|
|
|
registry = VoiceRegistry(cfg.voice_registry_path)
|
|
await registry.connect()
|
|
|
|
pid = await db.active_project_id()
|
|
|
|
results: dict[str, Any] = {
|
|
"stages": [],
|
|
"defects": [],
|
|
"summary": {},
|
|
"ambiguity_flags": [],
|
|
}
|
|
|
|
def log_stage(name: str, status: str, details: str = "") -> None:
|
|
results["stages"].append({"name": name, "status": status, "details": details})
|
|
logger.info("[%s] %s: %s", name, status, details)
|
|
|
|
def log_defect(stage: str, severity: str, message: str, context: str = "") -> None:
|
|
defect = {
|
|
"stage": stage,
|
|
"severity": severity,
|
|
"message": message,
|
|
"context": context,
|
|
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
|
|
}
|
|
results["defects"].append(defect)
|
|
logger.warning("[DEFECT][%s][%s] %s", stage, severity, message)
|
|
|
|
try:
|
|
project = await db.get_project(pid)
|
|
if not project:
|
|
log_stage("init", "FAILED", "No project found")
|
|
return results
|
|
|
|
chapters_db = await db.get_chapters(pid)
|
|
if not chapters_db:
|
|
log_stage("init", "FAILED", "No chapters found")
|
|
return results
|
|
|
|
log_stage("init", "OK", f"Project: {project['name']}, {len(chapters_db)} chapters")
|
|
|
|
from umbraaudio.voices.model_manager import get_pipeline_model_manager
|
|
from umbraaudio.voices.qwen_tts import QwenTTSEngine
|
|
|
|
tts = QwenTTSEngine(
|
|
device=cfg.tts.device,
|
|
sample_rate=cfg.tts.sample_rate,
|
|
voicedesign_model=cfg.tts.voicedesign_model,
|
|
base_model=cfg.tts.base_model,
|
|
qwen_tts_url=cfg.tts.qwen_tts_url,
|
|
)
|
|
log_stage("qwen_load", "OK", f"Qwen3-TTS configured (device={cfg.tts.device})")
|
|
|
|
model_manager = None
|
|
if cfg.tts.qwen_tts_url:
|
|
model_manager = get_pipeline_model_manager(cfg.tts.qwen_tts_url)
|
|
|
|
from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn
|
|
|
|
with Progress(
|
|
SpinnerColumn(),
|
|
TextColumn("{task.description}"),
|
|
BarColumn(),
|
|
TaskProgressColumn(),
|
|
console=__import__("rich.console", fromlist=["Console"]).Console(),
|
|
) as progress:
|
|
if model_manager:
|
|
await model_manager.for_analysis()
|
|
pass1_task = progress.add_task("Pass 1: Analysis (checking)...", total=1)
|
|
if project["status"] in ("initialized", "parsed"):
|
|
progress.update(pass1_task, description="Pass 1: Analysis...")
|
|
from umbraaudio.analysis.booknlp_engine import BookNLPEngine
|
|
from umbraaudio.analysis.character_profiler import CharacterProfiler
|
|
|
|
full_text = "\n\n".join(ch["source_text"] for ch in chapters_db)
|
|
booknlp = BookNLPEngine()
|
|
analysis_dir = cfg.data_dir / "analysis"
|
|
try:
|
|
result = booknlp.analyze(full_text, analysis_dir)
|
|
except Exception as e:
|
|
log_defect("analysis", "WARNING", f"BookNLP failed: {e}. Using fallback.")
|
|
result = {"characters": [], "quotes": []}
|
|
|
|
raw_characters = result.get("characters", [])
|
|
quotes = result.get("quotes", [])
|
|
|
|
merged = _merge_characters(raw_characters, quotes)
|
|
char_names = [c["name"] for c in merged[:30]]
|
|
|
|
profiler = CharacterProfiler(cfg.llm)
|
|
for cname in char_names:
|
|
ctx_text = "\n".join(
|
|
merged[
|
|
next((i for i, c in enumerate(merged) if c["name"] == cname), 0)
|
|
].get("context_snippets", [])
|
|
)
|
|
if not ctx_text:
|
|
for ch in chapters_db:
|
|
text = ch["source_text"]
|
|
idx = text.find(cname)
|
|
if idx >= 0:
|
|
start = max(0, idx - 100)
|
|
end = min(len(text), idx + 200)
|
|
ctx_text = text[start:end]
|
|
break
|
|
|
|
existing_chars = await db.get_characters(pid)
|
|
existing_names = {c["name"] for c in existing_chars}
|
|
if cname in existing_names:
|
|
continue
|
|
|
|
voice_registry_entry = await registry.find_by_name(cname)
|
|
profile = None
|
|
|
|
if voice_registry_entry:
|
|
from umbraaudio.analysis import CharacterProfile
|
|
|
|
profile = CharacterProfile(
|
|
name=cname,
|
|
voice_description=voice_registry_entry.get("voice_description"),
|
|
is_narrator=(cname == char_names[0] if char_names else False),
|
|
)
|
|
registry_id = voice_registry_entry["id"]
|
|
else:
|
|
if model_manager:
|
|
await model_manager.for_lm_studio()
|
|
profile = await profiler.profile_character(
|
|
name=cname,
|
|
aliases=[],
|
|
context=ctx_text or f"Character {cname}.",
|
|
)
|
|
|
|
char_id = await db.insert_character(
|
|
project_id=pid,
|
|
name=cname,
|
|
registry_id=registry_id if voice_registry_entry else None,
|
|
)
|
|
|
|
if not voice_registry_entry:
|
|
registry_char = await registry.find_or_create(
|
|
profile, source_project=project["name"]
|
|
)
|
|
registry_id = registry_char[0]
|
|
await db.conn.execute(
|
|
"UPDATE characters SET registry_id = ? WHERE id = ?",
|
|
(registry_id, char_id),
|
|
)
|
|
|
|
vd = profile.voice_description
|
|
if isinstance(vd, dict):
|
|
vd = json.dumps(vd)
|
|
await db.update_character_profile(
|
|
char_id,
|
|
profile_json=profile.to_dict(),
|
|
voice_description=vd,
|
|
)
|
|
|
|
from umbraaudio.analysis.pronunciation import PronunciationGenerator
|
|
|
|
pron = PronunciationGenerator(cfg.llm)
|
|
unusual = pron.find_unusual_terms(full_text)
|
|
unusual = [t for t in unusual if t not in existing_names][:20]
|
|
if unusual:
|
|
try:
|
|
if model_manager:
|
|
await model_manager.for_lm_studio()
|
|
guide = await profiler.generate_pronunciation_guide(
|
|
unusual, full_text[:3000]
|
|
)
|
|
for term, phonetic in guide.items():
|
|
await db.insert_pronunciation(1, term, phonetic)
|
|
except Exception as e:
|
|
log_defect("analysis", "WARNING", f"Pronunciation guide failed: {e}")
|
|
|
|
await db.update_project_status(1, "analyzed")
|
|
progress.update(
|
|
pass1_task,
|
|
advance=1,
|
|
description=f"Pass 1: {len(char_names)} characters analyzed",
|
|
)
|
|
else:
|
|
progress.update(pass1_task, advance=1, description="Pass 1: Already analyzed")
|
|
|
|
if model_manager:
|
|
await model_manager.for_qwen_voice_design()
|
|
pass2_task = progress.add_task("Pass 2: Voice references...", total=1)
|
|
characters_db = await db.get_characters(pid)
|
|
characters_to_voice = [
|
|
c for c in characters_db if not c["is_narrator"] and c["voice_description"]
|
|
]
|
|
|
|
from umbraaudio.voices.voice_designer import VoiceDesigner
|
|
|
|
designer = VoiceDesigner(cfg.tts)
|
|
voice_refs: dict[str, Path] = {}
|
|
speaker_profiles: dict[str, dict[str, Any]] = {}
|
|
|
|
for char in characters_to_voice:
|
|
profile_dict = {
|
|
"name": char["name"],
|
|
"energy": char.get("energy", "measured"),
|
|
"voice_description": char.get("voice_description"),
|
|
"pitch_register": char.get("pitch_register", "mid"),
|
|
"timbre": char.get("timbre", "clear"),
|
|
"texture": char.get("texture", "smooth"),
|
|
"formality": char.get("formality", "conversational"),
|
|
}
|
|
speaker_profiles[char["name"]] = profile_dict
|
|
|
|
if char.get("voice_ref_path") and Path(char["voice_ref_path"]).exists():
|
|
voice_refs[char["name"]] = Path(char["voice_ref_path"])
|
|
continue
|
|
try:
|
|
from umbraaudio.analysis import CharacterProfile
|
|
|
|
cp = CharacterProfile(
|
|
name=char["name"],
|
|
voice_description=char["voice_description"],
|
|
is_narrator=bool(char["is_narrator"]),
|
|
energy=char.get("energy", "measured"),
|
|
pitch_register=char.get("pitch_register", "mid"),
|
|
timbre=char.get("timbre", "clear"),
|
|
)
|
|
ref_path = designer.generate_voice_reference(cp, cfg.voices_dir)
|
|
voice_refs[char["name"]] = ref_path
|
|
embedding = designer.extract_embedding(ref_path)
|
|
|
|
await db.update_character_voice(char["id"], str(ref_path), embedding)
|
|
if char.get("registry_id"):
|
|
await registry.update_voice_ref(
|
|
char["registry_id"], str(ref_path), embedding
|
|
)
|
|
else:
|
|
registry_entry = await registry.find_by_name(char["name"])
|
|
if registry_entry:
|
|
await registry.update_voice_ref(
|
|
registry_entry["id"], str(ref_path), embedding
|
|
)
|
|
|
|
except Exception as e:
|
|
log_defect("voice_design", "ERROR", f"Voice ref failed for {char['name']}: {e}")
|
|
|
|
progress.update(
|
|
pass2_task,
|
|
advance=1,
|
|
description=f"Pass 2: {len(voice_refs)} voice refs created",
|
|
)
|
|
|
|
if model_manager:
|
|
await model_manager.for_segmentation()
|
|
pass3_task = progress.add_task("Pass 3: Segment & manifest...", total=len(chapters_db))
|
|
from umbraaudio.mapping.manifest import build_generation_manifest, save_manifest
|
|
from umbraaudio.mapping.segmenter import Segmenter
|
|
|
|
segmenter = Segmenter()
|
|
characters_map = {}
|
|
for c in characters_db:
|
|
characters_map[c["name"]] = {
|
|
"voice_ref_path": c.get("voice_ref_path"),
|
|
"voice_description": c.get("voice_description"),
|
|
}
|
|
|
|
real_character_names = []
|
|
seen_canonical = set()
|
|
for c in characters_db:
|
|
if (
|
|
c["is_narrator"]
|
|
or c["name"].lower() in {n.lower() for n in NOISE_CHARACTERS}
|
|
or len(c["name"]) <= 2
|
|
):
|
|
continue
|
|
canonical = ALIAS_MAP.get(c["name"], c["name"])
|
|
if canonical not in seen_canonical:
|
|
seen_canonical.add(canonical)
|
|
real_character_names.append(c["name"])
|
|
|
|
alias_keys = [k for k in ALIAS_MAP if k.lower() not in {n.lower() for n in real_character_names}]
|
|
real_character_names.extend(alias_keys)
|
|
|
|
pronunciation_guide = await db.get_pronunciation_guide(pid)
|
|
|
|
for ch in chapters_db:
|
|
utterances = segmenter.segment(
|
|
ch["source_text"],
|
|
ch["id"],
|
|
real_character_names,
|
|
chapter_title=ch.get("title"),
|
|
)
|
|
for u in utterances:
|
|
if u.get("character_name") is None:
|
|
u["character_name"] = "narrator"
|
|
canonical = ALIAS_MAP.get(u["character_name"])
|
|
if canonical:
|
|
u["character_name"] = canonical
|
|
|
|
from umbraaudio.analysis.ambiguity import detect_ambiguities
|
|
|
|
flags = detect_ambiguities(utterances, real_character_names, ch.get("title", ""))
|
|
for flag in flags:
|
|
results["ambiguity_flags"].append(flag.to_dict())
|
|
log_defect(
|
|
"segmentation",
|
|
"WARNING",
|
|
f"Ambiguous attribution ch{ch['id']} utt#{flag.utterance_index}: "
|
|
f"{flag.reason} (confidence={flag.confidence:.2f})",
|
|
)
|
|
|
|
# Clear existing utterances for this chapter and insert new ones
|
|
await db.conn.execute("DELETE FROM utterances WHERE chapter_id = ?", (ch["id"],))
|
|
await db.conn.commit()
|
|
|
|
# Insert utterances into database for LLM attribution
|
|
char_name_to_id = {c["name"]: c["id"] for c in characters_db}
|
|
for u in utterances:
|
|
char_id = char_name_to_id.get(u.get("character_name"))
|
|
await db.insert_utterance(
|
|
chapter_id=ch["id"],
|
|
character_id=char_id,
|
|
index=u["index"],
|
|
text=u["text"],
|
|
type=u.get("type", "narration"),
|
|
emotion=u.get("emotion"),
|
|
pace=u.get("pace"),
|
|
instruct=u.get("instruct"),
|
|
)
|
|
|
|
manifest = build_generation_manifest(
|
|
ch["id"],
|
|
utterances,
|
|
characters_map,
|
|
pronunciation_guide,
|
|
)
|
|
manifest_path = cfg.manifests_dir / f"ch{ch['id']}.json"
|
|
save_manifest(manifest, manifest_path)
|
|
progress.update(pass3_task, advance=1)
|
|
|
|
# Pass 4: LLM Speaker Attribution with multi-pass context analysis
|
|
if model_manager:
|
|
await model_manager.for_analysis()
|
|
pass4_task = progress.add_task("Pass 4: LLM speaker attribution...", total=len(chapters_db))
|
|
from scripts.run_llm_attribution import run_llm_attribution
|
|
|
|
try:
|
|
attrib_result = await run_llm_attribution()
|
|
except Exception as e:
|
|
import traceback
|
|
tb_str = traceback.format_exc()
|
|
logger.error("LLM attribution failed with exception: %s", e)
|
|
logger.error("Full traceback:\n%s", tb_str)
|
|
print("LLM ATTRIBUTION ERROR:", file=sys.stderr)
|
|
print(tb_str, file=sys.stderr)
|
|
log_defect("attribution", "CRITICAL", f"LLM attribution failed: {e}")
|
|
raise
|
|
progress.update(pass4_task, advance=len(chapters_db))
|
|
|
|
# Check quality gate
|
|
if not attrib_result.get("quality_gate_passed", True):
|
|
log_defect(
|
|
"attribution",
|
|
"ERROR",
|
|
f"Quality gate FAILED: {attrib_result.get('exception_rate', 0):.2f}% exceptions exceeds 1% threshold",
|
|
)
|
|
progress.update(pass4_task, description="Quality gate FAILED")
|
|
raise RuntimeError(
|
|
f"Quality gate failed: {attrib_result.get('exception_rate', 0):.2f}% exceptions exceeds 1% threshold"
|
|
)
|
|
|
|
from umbraaudio.assembly.normalize import AudioNormalizer
|
|
from umbraaudio.generation.cache import GenerationCache
|
|
from umbraaudio.generation.chapter import ChapterGenerator
|
|
|
|
cache = GenerationCache(cfg.data_dir / "cache")
|
|
normalizer = AudioNormalizer(
|
|
target_rms_db=cfg.audio.target_rms_db,
|
|
peak_limit_db=cfg.audio.peak_limit_db,
|
|
sample_rate=cfg.tts.sample_rate,
|
|
)
|
|
generator = ChapterGenerator(
|
|
tts,
|
|
cfg,
|
|
cache=cache,
|
|
normalizer=normalizer,
|
|
watermark_bits=_author_watermark_bits(author_name),
|
|
author_name=author_name,
|
|
)
|
|
|
|
if model_manager:
|
|
await model_manager.for_generation(needs_clone=True)
|
|
|
|
from umbraaudio.mapping.manifest import load_manifest
|
|
|
|
if chapters:
|
|
chapter_range = _parse_chapter_range(chapters, len(chapters_db))
|
|
else:
|
|
chapter_range = range(len(chapters_db))
|
|
|
|
pass5_task = progress.add_task(
|
|
"Pass 5: Generate audio...",
|
|
total=sum(
|
|
len(
|
|
load_manifest(cfg.manifests_dir / f"ch{chapters_db[i]['id']}.json").get(
|
|
"utterances", []
|
|
)
|
|
)
|
|
if (cfg.manifests_dir / f"ch{chapters_db[i]['id']}.json").exists()
|
|
else 0
|
|
for i in chapter_range
|
|
),
|
|
)
|
|
generated_chapter_files: list[Path] = []
|
|
|
|
for i in chapter_range:
|
|
ch = chapters_db[i]
|
|
manifest_path = cfg.manifests_dir / f"ch{ch['id']}.json"
|
|
if not manifest_path.exists():
|
|
progress.update(pass5_task, description=f"Skipping ch{ch['id']} (no manifest)")
|
|
continue
|
|
|
|
manifest = load_manifest(manifest_path)
|
|
|
|
chapter_audio_dir = cfg.audio_dir / f"ch{ch['id']}"
|
|
chapter_audio_dir.mkdir(parents=True, exist_ok=True)
|
|
normalized_dir = cfg.data_dir / "audio_normalized" / f"ch{ch['id']}"
|
|
normalized_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
result = generator.generate_chapter(
|
|
manifest,
|
|
chapter_audio_dir,
|
|
normalized_dir,
|
|
speaker_profiles=speaker_profiles,
|
|
)
|
|
generated_files = result["generated_files"]
|
|
|
|
gc.collect()
|
|
if torch.cuda.is_available():
|
|
torch.cuda.empty_cache()
|
|
|
|
if generated_files:
|
|
generated_chapter_files.extend(generated_files)
|
|
|
|
progress.update(pass5_task, advance=len(generated_files))
|
|
|
|
pass6_task = progress.add_task("Quality check...", total=len(generated_chapter_files))
|
|
defects_found = 0
|
|
|
|
for ch_audio in generated_chapter_files:
|
|
if not ch_audio.exists():
|
|
log_defect("quality", "ERROR", f"Missing audio file: {ch_audio}")
|
|
defects_found += 1
|
|
progress.update(pass6_task, advance=1)
|
|
continue
|
|
|
|
try:
|
|
compliance = normalizer.check_acx_compliance(ch_audio)
|
|
if not compliance["compliant"]:
|
|
log_defect(
|
|
"quality",
|
|
"WARNING",
|
|
f"ACX non-compliant: {ch_audio.name} "
|
|
f"(RMS={compliance['rms_db']}dB, peak={compliance['peak_db']}dB)",
|
|
)
|
|
defects_found += 1
|
|
except Exception as e:
|
|
log_defect(
|
|
"quality", "WARNING", f"Quality check failed for {ch_audio.name}: {e}"
|
|
)
|
|
defects_found += 1
|
|
|
|
progress.update(pass6_task, advance=1)
|
|
|
|
export_task = progress.add_task("Export final audiobook...", total=1)
|
|
from umbraaudio.assembly.export import AudiobookExporter
|
|
|
|
exporter = AudiobookExporter()
|
|
|
|
if generated_chapter_files:
|
|
m4b_path = cfg.export_dir / "audiobook.m4b"
|
|
try:
|
|
exporter.export_m4b(
|
|
generated_chapter_files,
|
|
m4b_path,
|
|
title=project["name"],
|
|
author=author_name,
|
|
)
|
|
log_stage("export", "OK", f"Exported M4B: {m4b_path}")
|
|
except Exception as e:
|
|
log_defect("export", "WARNING", f"M4B export failed: {e}")
|
|
|
|
mp3_files = exporter.export_mp3(
|
|
generated_chapter_files, cfg.export_dir / "mp3"
|
|
)
|
|
if mp3_files:
|
|
log_stage("export", "OK", f"Exported {len(mp3_files)} MP3 chapters")
|
|
|
|
wav_files = exporter.export_wav(
|
|
generated_chapter_files, cfg.export_dir / "wav"
|
|
)
|
|
if wav_files:
|
|
log_stage("export", "OK", f"Exported {len(wav_files)} WAV chapters")
|
|
else:
|
|
log_defect("export", "ERROR", "No audio files to export")
|
|
|
|
progress.update(export_task, advance=1)
|
|
|
|
total_utt_count = 0
|
|
for i in range(len(chapters_db)):
|
|
mp = cfg.manifests_dir / f"ch{chapters_db[i]['id']}.json"
|
|
if mp.exists():
|
|
m = load_manifest(mp)
|
|
total_utt_count += len(m.get("utterances", []))
|
|
|
|
total_gen_count = sum(
|
|
len(list((cfg.audio_dir / f"ch{chapters_db[i]['id']}").glob("*.wav")))
|
|
for i in range(len(chapters_db))
|
|
if (cfg.audio_dir / f"ch{chapters_db[i]['id']}").exists()
|
|
)
|
|
|
|
results["summary"] = {
|
|
"project": project["name"],
|
|
"characters": len(characters_db),
|
|
"chapters": len(chapters_db),
|
|
"utterances_total": total_utt_count,
|
|
"audio_generated": total_gen_count,
|
|
"voice_refs": len(voice_refs),
|
|
"defects": len(results["defects"]),
|
|
"ambiguity_flags": len(results["ambiguity_flags"]),
|
|
}
|
|
|
|
await db.update_project_status(1, "generated")
|
|
log_stage("pipeline", "COMPLETE", json.dumps(results["summary"], indent=2))
|
|
|
|
except Exception as e:
|
|
log_stage("pipeline", "FAILED", str(e))
|
|
log_defect("pipeline", "CRITICAL", f"Pipeline failed: {e}")
|
|
|
|
finally:
|
|
if model_manager:
|
|
await model_manager.cleanup()
|
|
await registry.close()
|
|
await db.close()
|
|
|
|
return results
|
|
|
|
|
|
def _parse_chapter_range(chapters_str: str, total: int) -> range:
|
|
if chapters_str == "all":
|
|
return range(total)
|
|
parts = chapters_str.split(",")
|
|
indices: list[int] = []
|
|
for part in parts:
|
|
part = part.strip()
|
|
if "-" in part:
|
|
start, end = part.split("-", 1)
|
|
start_i = max(0, int(start) - 1)
|
|
end_i = min(total, int(end))
|
|
indices.extend(range(start_i, end_i))
|
|
else:
|
|
idx = int(part) - 1
|
|
if 0 <= idx < total:
|
|
indices.append(idx)
|
|
return range(max(indices), max(indices) + 1) if indices else range(0)
|