diff options
Diffstat (limited to 'app/backends/faster.py')
| -rwxr-xr-x | app/backends/faster.py | 161 |
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: |