aboutsummaryrefslogtreecommitdiff
path: root/app/backends/faster.py
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-24 18:57:09 -0400
committerhistoria <historiavg@proton.me>2026-08-24 18:57:09 -0400
commit0522e73b68291af62e43c387ab8f9b8ffa2cab47 (patch)
tree1800168c1ed5de7540624265945faba8251a7cab /app/backends/faster.py
parentd950fc8e64ee508334e608f6045d687d73a464be (diff)
downloadtts-audiobook-generator-0522e73b68291af62e43c387ab8f9b8ffa2cab47.tar.gz
feat: configure [backend] loads existing backend config rather than just overwriting it
Diffstat (limited to 'app/backends/faster.py')
-rwxr-xr-xapp/backends/faster.py161
1 files changed, 132 insertions, 29 deletions
diff --git a/app/backends/faster.py b/app/backends/faster.py
index e1249ca..7e1be74 100755
--- a/app/backends/faster.py
+++ b/app/backends/faster.py
@@ -14,10 +14,17 @@ Usage:
python app/backends/faster.py [--wavs WAV_DIR] [--output PATH]
[--language LANG] [--whisper-model NAME] [--force]
[--port PORT] [--voice NAME] [--skip-install] [--skip-clone]
+
+When the target ``voices.json`` already exists, the TUI wizard runs as a
+"modify": it loads the existing voices and pre-fills the language and
+wav directory from them instead of prompting to overwrite, asks whether
+to only transcribe new voices or re-transcribe everything, and writes
+back to the same file.
"""
import argparse
import json
+import shutil
import sys
from pathlib import Path
from typing import List, Optional
@@ -26,7 +33,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from backends import (
BackendStatus,
- ConfigureAction,
ServerSpec,
common,
envs,
@@ -95,9 +101,68 @@ def build_voices(wav_files: list, language: str, whisper_model: str) -> dict:
return voices
+def load_voices(path: Path) -> dict:
+ """Read voices.json into a name -> voice-entry dict, or {} when unusable.
+
+ Returns {} for a missing file, unreadable content, or a non-dict
+ document. Used by the wizard's modify flow to seed its defaults from an
+ existing voices.json instead of prompting to overwrite it.
+ """
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, ValueError):
+ return {}
+ if not isinstance(data, dict):
+ return {}
+ return data
+
+
+def _decide_faster_transcription(wav_files: list, existing_voices: dict,
+ confirm) -> Optional[dict]:
+ """Decide which voices to transcribe when a voices.json already exists.
+
+ CONFIRM asks the yes/no question (returning True/False, or None when the
+ user backs out). With new .wavs present it offers to transcribe only
+ those (default Yes); otherwise — and always, per the modify design — it
+ offers to re-transcribe everything (default No), so a stale transcript
+ can be refreshed even when every voice is already known. Returns a plan
+ dict: ``{"mode": "missing"|"all"|"keep", "missing": [...], "existing":
+ {...}}``, or None when CONFIRM cancelled.
+ """
+ existing = dict(existing_voices)
+ new_wavs = [wav for wav in wav_files if wav.stem not in existing]
+ if new_wavs:
+ choice = confirm("Existing voices.json found. Only transcribe the "
+ "new voices?", True)
+ if choice is None:
+ return None
+ if choice:
+ return {"mode": "missing", "missing": new_wavs,
+ "existing": existing}
+ return {"mode": "all", "missing": [], "existing": existing}
+ choice = confirm("All voices already in voices.json. Re-transcribe "
+ "anyway?", False)
+ if choice is None:
+ return None
+ if choice:
+ return {"mode": "all", "missing": [], "existing": existing}
+ return {"mode": "keep", "missing": [], "existing": existing}
+
+
def _write_voices_json(output_path: Path, wav_dir: Path, language: str,
- whisper_model: str, force: bool) -> Optional[dict]:
- """Transcribe the wav dir and write voices.json; return the voices dict."""
+ whisper_model: str, plan: Optional[dict]) -> Optional[dict]:
+ """Transcribe the wav dir and write voices.json; return the voices dict.
+
+ PLAN (built by ``_decide_faster_transcription`` in the wizard, or an
+ "all" plan for a fresh/flag run) decides whether every voice is
+ re-transcribed ("all"), only the new ones ("missing" — merged into the
+ existing entries), or nothing changes ("keep" — the existing file is
+ left untouched and returned as-is). None (cancelled) writes nothing.
+ """
+ if plan is None:
+ return None
+ if plan["mode"] == "keep":
+ return dict(plan["existing"])
wav_files = find_wav_files(wav_dir)
if not wav_files:
print(f"[ERROR] No .wav files found in {wav_dir}")
@@ -106,7 +171,11 @@ def _write_voices_json(output_path: Path, wav_dir: Path, language: str,
print("[WARNING] Neither faster_whisper nor whisper was found, so "
"transcripts will be empty — install one or edit voices.json "
"by hand.")
- voices = build_voices(wav_files, language, whisper_model)
+ if plan["mode"] == "missing":
+ voices = dict(plan["existing"])
+ voices.update(build_voices(plan["missing"], language, whisper_model))
+ else:
+ voices = build_voices(wav_files, language, whisper_model)
with output_path.open("w", encoding="utf-8") as handle:
json.dump(voices, handle, indent=4, ensure_ascii=False)
handle.write("\n")
@@ -142,17 +211,39 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
return None
do_clone = choice
- # Step 2: voices.json — wav dir, language, whisper model, output path.
+ # Step 2: voices.json — an existing one seeds the defaults (modify flow)
+ # instead of an overwrite prompt.
+ existing_voices = {}
+ default_output = args.output
+ if default_output is None and _is_cloned():
+ default_output = _checkout() / "voices.json"
+ if default_output is not None and default_output.exists() \
+ and not args.force:
+ existing_voices = load_voices(default_output)
+
+ wav_start = VOICES_DIR
+ if existing_voices:
+ ref_dirs = {Path(voice["ref_audio"]).parent
+ for voice in existing_voices.values()
+ if isinstance(voice, dict) and voice.get("ref_audio")}
+ if len(ref_dirs) == 1:
+ wav_start = next(iter(ref_dirs))
+
wav_dir = args.input_dir
if wav_dir is None:
wav_dir = tui.browse_directory(
stdscr, "Select the directory with your .wav voices",
info=common.wav_dir_info, preview=common.wav_dir_preview,
- start=VOICES_DIR)
+ start=wav_start)
language = args.language
if language is None:
+ default_language = config.LANGUAGE
+ for voice in existing_voices.values():
+ if isinstance(voice, dict) and voice.get("language"):
+ default_language = voice["language"]
+ break
lang_text = tui.line_edit(
- stdscr, "Language", config.LANGUAGE,
+ stdscr, "Language", default_language,
validate=lambda s: None if _try_language(s)
else "Unknown language (e.g. English, en)",
help_lines=["Language for every voice, as passed to the TTS "
@@ -170,12 +261,16 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
# when the checkout is not present (so a flag-only run still works).
output_path = (_checkout() / "voices.json") if _is_cloned() \
else (wav_dir / "voices.json")
- if output_path.exists() and not args.force:
- choice = confirm(f"{output_path} already exists. Overwrite?",
- default=True)
- if choice is None or choice is False:
- # Fall back to a path in the current directory.
- output_path = Path.cwd() / "voices.json"
+
+ # Transcription plan: re-transcribe only new voices (or all of them) —
+ # the "re-transcribe anyway?" offer appears even when nothing is new.
+ plan: Optional[dict] = {"mode": "all", "missing": [], "existing": {}}
+ wav_files = find_wav_files(wav_dir)
+ if wav_files and existing_voices and not args.force:
+ plan = _decide_faster_transcription(wav_files, existing_voices,
+ confirm)
+ if plan is None:
+ return None
# Step 3: port + default voice.
port = args.port
@@ -195,6 +290,7 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
"output_path": output_path,
"port": port,
"force": args.force,
+ "plan": plan,
}
@@ -226,7 +322,7 @@ def _execute(settings: dict) -> int:
voices = _write_voices_json(settings["output_path"], settings["wav_dir"],
settings["language"], settings["whisper_model"],
- settings["force"])
+ settings["plan"])
if voices is None:
return 1
@@ -308,6 +404,7 @@ def _collect_from_flags(args: argparse.Namespace,
"output_path": output_path,
"port": args.port if args.port is not None else _config_port(),
"force": args.force,
+ "plan": {"mode": "all", "missing": [], "existing": {}},
}
@@ -332,7 +429,8 @@ def build_parser() -> argparse.ArgumentParser:
"(default: base)")
parser.add_argument("--force", action="store_true",
help="Overwrite an existing voices.json without "
- "prompting")
+ "prompting; in the TUI, re-transcribe every "
+ "voice instead of reusing the existing file")
parser.add_argument("--port", type=int, default=None,
help="Server port to record in app/converter/config.py "
"(default: the port in FASTER_API_URL)")
@@ -394,22 +492,27 @@ def _detect_remote(managed: bool = False):
return False, {}
-def _run_voices_only_tui() -> int:
- """Rebuild voices.json via the TUI (the "configure" action).
+def uninstall() -> int:
+ """Remove the faster-qwen3-tts backend entirely.
- Runs the same wizard but skips the pip/clone prerequisites so it goes
- straight to picking the .wav directory and writing voices.json.
+ Uninstalls the pip package (``faster-qwen3-tts``) from the managed venv
+ and deletes the cloned checkout (``app/faster-qwen3-tts``, which holds
+ examples/openai_server.py and voices.json). A running server this tool
+ started is stopped first (best-effort). Returns the exit code.
"""
- args = build_parser().parse_args([])
- args.skip_install = True
- args.skip_clone = True
- return run_tui(args)
-
-
-configure_actions: List[ConfigureAction] = [
- ConfigureAction("Rebuild voices.json", _run_voices_only_tui),
- ConfigureAction("Reconfigure faster-qwen3-tts", run_tui),
-]
+ servers.stop("faster")
+ rc = common.pip_uninstall(["faster-qwen3-tts"])
+ if rc != 0:
+ print("[WARNING] pip uninstall failed (exit "
+ f"{rc}); remove faster-qwen3-tts from the managed venv manually")
+ else:
+ print("[OK] faster-qwen3-tts removed.")
+ checkout = _checkout()
+ if checkout.is_dir():
+ print(f"[INFO] Removing checkout {checkout}...")
+ shutil.rmtree(checkout, ignore_errors=True)
+ print("[OK] checkout removed.")
+ return 0
def main() -> int: