aboutsummaryrefslogtreecommitdiff
path: root/app/backends/audiocpp/voices.py
blob: 2f0fdd78f2a642a82c501a9d1800e7b124a77965 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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": {}}