refactor: audiocpp.py setup flow

1 files changed, 146 insertions, 0 deletions

diff --git a/app/backends/audiocpp/voices.py b/app/backends/audiocpp/voices.py
new file mode 100644
index 0000000..2f0fdd7
--- /dev/null
+++ b/app/backends/audiocpp/voices.py
@@ -0,0 +1,146 @@
+"""Reference-.wav transcription planning and execution."""
+
+import argparse
+from pathlib import Path
+from typing import Callable, Dict, List, Optional, Tuple
+
+from backends.common import (PROMPT_TEXT_FILENAME, find_wav_files,
+                             read_prompt_text)
+from converter.clients import (transcribe_reference_audio,
+                               whisper_backend_available)
+
+def transcribe_wav_dir(wav_files: list, whisper_model: str,
+                       cancel=None) -> Dict[str, str]:
+    """Transcribe each wav file and return a mapping of stem -> transcript.
+
+    CANCEL (a ``threading.Event``) is checked between files so the in-TUI
+    task view can stop a long transcription early.
+    """
+    transcripts: Dict[str, str] = {}
+    for wav_file in wav_files:
+        if cancel is not None and cancel.is_set():
+            print("[INFO] Transcription cancelled")
+            break
+        name = wav_file.stem
+        print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...")
+        text = transcribe_reference_audio(str(wav_file), model_name=whisper_model)
+        if text:
+            print(f"[OK] {name}: {text}")
+        else:
+            print(f"[WARNING] No transcript for '{name}'; cloning works best "
+                  "with an accurate transcript — consider editing prompt_text "
+                  "by hand before starting the server")
+        transcripts[name] = text or ""
+    return transcripts
+
+
+def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None:
+    """Print a loud, final warning for voices whose transcript is empty."""
+    empty = sorted(name for name, text in transcripts.items() if not text)
+    if not empty:
+        return
+    bar = "=" * 70
+    print()
+    print(bar)
+    print("[WARNING] MANUAL TRANSCRIPTION REQUIRED")
+    print(bar)
+    listing = "  - " + "\n  - ".join(empty) if len(empty) > 1 else f"  - {empty[0]}"
+    print(f"The following voice(s) have an EMPTY transcript in prompt_text:\n"
+          f"{listing}")
+    print("Those voices will NOT work until you add an accurate transcript.")
+    print(f"Edit {PROMPT_TEXT_FILENAME} in your voice directory and fill in the "
+          "text after '|' for each voice above.")
+    print(bar)
+
+
+def _decide_transcription(wav_files: list, existing: Dict[str, str],
+                          prompt_exists: bool, force: bool,
+                          confirm: Callable[[str, bool], bool]) -> dict:
+    """Decide which voices to transcribe; CONFIRM asks the plan questions.
+
+    Returns a plan dict: {"mode": "all"|"missing"|"keep", "missing":
+    [...], "existing": {...}} — "existing" carries the prompt_text
+    mapping read while deciding, so the caller can reuse it instead of
+    reading the file again.
+    """
+    mode = "all"
+    missing: List[Path] = []
+    if prompt_exists and not force:
+        missing = [wav for wav in wav_files
+                   if not existing.get(wav.stem, "").strip()]
+        if not missing:
+            if confirm("All voices already transcribed in prompt_text. "
+                       "Re-transcribe anyway?", False):
+                mode = "all"
+            else:
+                mode = "keep"
+        elif confirm("Existing transcription and new .wavs detected, "
+                     "only transcribe new voices?", True):
+            mode = "missing"
+        else:
+            mode = "all"
+    return {"mode": mode, "missing": missing, "existing": existing}
+
+
+def _transcribe(args: argparse.Namespace, plan: Optional[dict],
+                cancel=None) -> Tuple[Dict[str, str], bool]:
+    """Transcribe the wav directory into a stem -> transcript mapping.
+
+    Returns the mapping and a flag indicating whether it should be written to
+    prompt_text (False when an existing, complete prompt_text is kept as-is).
+    PLAN is always pre-collected — by the TUI (via _decide_transcription and
+    its confirm callbacks) or by _flag_plan for a non-interactive run — so no
+    questions are asked here; a None PLAN defaults to "transcribe everything".
+    CANCEL is checked between files.
+    """
+    wav_files = find_wav_files(args.input_dir)
+    if not wav_files:
+        print(f"[WARNING] No .wav files found in {args.input_dir}; writing the "
+              "config without a voice_dir")
+        return {}, False
+
+    prompt_path = args.input_dir / PROMPT_TEXT_FILENAME
+    existing = dict((plan or {}).get("existing") or {})
+    mode = plan["mode"] if plan else "all"
+
+    if mode == "keep":
+        print(f"[INFO] Kept existing {prompt_path}; all voices were "
+              "already transcribed, nothing new to transcribe")
+        return existing, False
+
+    if whisper_backend_available() is None:
+        print("[WARNING] Neither faster_whisper nor whisper was found, so "
+              "reference .wav files cannot be transcribed automatically and "
+              "every transcript will be empty.")
+        print("          Install whisper (or faster_whisper) in your "
+              "audiobook environment to transcribe automatically; otherwise "
+              "transcripts must be added by hand (see the warning at the end).")
+
+    if plan["mode"] == "missing":
+        new_transcripts = transcribe_wav_dir(plan["missing"], args.whisper_model,
+                                             cancel=cancel)
+        transcripts = dict(existing)
+        transcripts.update(new_transcripts)
+    else:
+        transcripts = transcribe_wav_dir(wav_files, args.whisper_model,
+                                         cancel=cancel)
+    return transcripts, True
+
+
+def _flag_plan(wav_files: list, prompt_path: Path, force: bool) -> dict:
+    """Build a transcription plan for a non-interactive (flag-only) run.
+
+    With --force everything is re-transcribed; otherwise an existing
+    prompt_text is reused and only voices with an empty transcript are
+    re-transcribed, mirroring what the TUI confirms interactively.
+    """
+    if prompt_path.exists() and not force:
+        existing = read_prompt_text(prompt_path)
+        missing = [wav for wav in wav_files
+                   if not existing.get(wav.stem, "").strip()]
+        if not missing:
+            return {"mode": "keep", "missing": [], "existing": existing}
+        return {"mode": "missing", "missing": missing, "existing": existing}
+    return {"mode": "all", "missing": [], "existing": {}}
+
+