app.py

"""EchoScript v1.0 UI.

Two independent, mutually exclusive processing pipelines, chosen per
click -- not two steps of one pipeline:

    Transcript & Translations pipeline:
        Upload Audio -> Select Audio Window (optional)
                      -> Detect Language & Generate Canonical Transcript
                      -> Preview Transcript + available translation languages
                      -> Generate Translations -> Copy / Download

    Phonetic Transcription (IPA) pipeline:
        Upload Audio -> Select Audio Window (optional)
                      -> Generate Phonetic Transcription -> Copy / Download

These never run together from the same click. That's deliberate: the
Transcript pipeline treats the Transcript as the single source of truth
for every translation and subtitle it produces, and the Phonetics
pipeline reads the audio directly with no dependency on -- or influence
from -- the Transcript at all (see services/phonetics.py for why that
independence matters). Mixing them into one combined action would mean
either running Whisper when someone only wanted IPA phones, or running
the phone recognizer when someone only wanted a transcript -- both
wasted work, and it would blur which of the two "touches the audio"
for a given result.

Within the Transcript pipeline, "Generate Transcript" and "Generate
Translations" remain two distinct actions:

1. "Generate Transcript" -- audio, time window, and an optional source-
   language hint go in; a canonical Transcript comes out (detected
   language, confidence, duration, word count, full text). This is the
   only step that touches the audio in this pipeline. Once it's done,
   the translation-language picker appears, scoped to the language that
   was *actually* detected (or forced) -- never a pre-detection guess.
2. "Generate Translations" -- pick which languages to translate the
   transcript into (the list depends on whether an Anthropic API key is
   present: more languages with a key, the offline-safe set without one)
   and click again. Always derived from the cached Transcript, never
   from the audio.

Caching: a `gr.State` holds the last Transcript plus the exact (audio,
window, source-language) signature that produced it -- clicking "Generate
Transcript" again with that signature unchanged reuses it instead of
re-running Whisper. A second `gr.State` dict caches each Translation by
language code, scoped to the current transcript, so adding one more
language to "Generate Translations" doesn't redo the others, and
deselecting a language doesn't drop it from the cache (just hides it).
The Phonetics pipeline has its own, separate cache/signature pair, keyed
only on (audio, window) -- it has no source-language concept at all.

Services are instantiated lazily (on first use) rather than at import
time, so the app can start up without needing model weights on disk yet.
"""

from __future__ import annotations

import tempfile
from pathlib import Path
from typing import Optional

import gradio as gr

from models.transcript import Transcript
from services.audio import AudioError, extract_window, resolve_window, validate_extension
from services.phonetics import PhoneticsError, transcribe_phonetics
from services.subtitles import generate_srt, generate_vtt
from services.transcription import SUPPORTED_LANGUAGES, TranscriptionService
from services.translation import (
    ANTHROPIC_TARGET_LANGUAGES,
    TranslationError,
    TranslationService,
    available_marian_targets,
)

# ---------------------------------------------------------------------------
# Lazy service singletons -- safe to share across requests/users; see
# services/translation.py for why TranslationService holds no key state.
# ---------------------------------------------------------------------------

_transcription_service: Optional[TranscriptionService] = None
_translation_service: Optional[TranslationService] = None


def get_transcription_service() -> TranscriptionService:
    global _transcription_service
    if _transcription_service is None:
        _transcription_service = TranscriptionService(
            model_size="base",
            device="cpu",
            compute_type="int8",
            download_root="/tmp/whisper_models",
        )
    return _transcription_service


def get_translation_service() -> TranslationService:
    global _translation_service
    if _translation_service is None:
        _translation_service = TranslationService()
    return _translation_service


# ---------------------------------------------------------------------------
# UI <-> service-layer vocabulary
# ---------------------------------------------------------------------------

# "Source Language" dropdown: display name -> ISO 639-1 code (None = auto).
_NAME_TO_CODE = {name: code for code, name in SUPPORTED_LANGUAGES.items()}
SOURCE_LANGUAGE_CHOICES = ["Auto Detect"] + list(SUPPORTED_LANGUAGES.values())

# Label -> ISO 639-1 code, built from the full (Anthropic) superset so it
# resolves correctly regardless of which list is currently offered.
# Labels are just language names (e.g. "French", not "French Translation").
_TRANSLATION_LABEL_TO_CODE = {
    name: code for code, name in ANTHROPIC_TARGET_LANGUAGES.items()
}

# Sections are pre-built for every language in the full superset (hidden by
# default) so that toggling the API key never needs to add/remove
# components -- only which ones are visible changes.
_TRANSLATION_SECTION_ORDER = list(_TRANSLATION_LABEL_TO_CODE.keys())


def _compute_translation_choices(has_key: bool, exclude_code: Optional[str]) -> list[str]:
    """The translate-to picker, cascaded from key presence + source language."""
    if has_key:
        pool = ANTHROPIC_TARGET_LANGUAGES
    else:
        pool = available_marian_targets(exclude_code) if exclude_code else {}
    return [name for code, name in pool.items() if code != exclude_code]


def _on_api_key_change(api_key: str, cached_transcript: Optional[Transcript], current_value: list[str]):
    """Re-cascade the translate-to picker live as the API key field changes.

    No-op until a transcript exists -- the picker isn't shown before then,
    so there's nothing yet to cascade.
    """
    if cached_transcript is None:
        return gr.update()
    has_key = bool((api_key or "").strip())
    choices = _compute_translation_choices(has_key, cached_transcript.language)
    filtered_value = [v for v in (current_value or []) if v in choices]
    return gr.update(choices=choices, value=filtered_value)


def _format_duration(seconds: float) -> str:
    seconds = max(0, int(round(seconds)))
    hours, remainder = divmod(seconds, 3600)
    minutes, secs = divmod(remainder, 60)
    return f"{hours:02}:{minutes:02}:{secs:02}"


def _write_text_file(text: str, tmp_dir: Path, filename: str) -> str:
    path = tmp_dir / filename
    path.write_text(text, encoding="utf-8")
    return str(path)


# ---------------------------------------------------------------------------
# Stage 1: Generate Transcript
# ---------------------------------------------------------------------------

def generate_transcript(
    audio_path: Optional[str],
    start_value: str,
    end_value: str,
    source_language_label: str,
    api_key: str,
    cached_transcript: Optional[Transcript],
    cached_signature,
    cached_translations: dict,
):
    """Transcript & Translations mode. Never touches the phonetics engine."""
    if not audio_path:
        raise gr.Error("Please upload an audio file first.")

    try:
        validate_extension(audio_path)
        start, end = resolve_window(start_value, end_value)
    except AudioError as exc:
        raise gr.Error(str(exc)) from exc

    source_code = _NAME_TO_CODE.get(source_language_label)  # None = auto-detect
    signature = (audio_path, start, end, source_code)
    regenerated = not (cached_transcript is not None and cached_signature == signature)

    if regenerated:
        working_path = audio_path
        if start is not None or end is not None:
            try:
                working_path = extract_window(audio_path, start, end)
            except AudioError as exc:
                raise gr.Error(str(exc)) from exc

        # The only step that touches the audio in this mode.
        transcript = get_transcription_service().transcribe(
            working_path,
            source_filename=Path(audio_path).name,
            language=source_code,
            window_start=start,
            window_end=end,
        )
        cached_signature = signature
        cached_translations = {}  # old translations were derived from a different transcript
    else:
        # Same audio, window, and source-language hint as last time --
        # skip Whisper entirely and reuse the cached Transcript.
        transcript = cached_transcript

    tmp_dir = Path(tempfile.mkdtemp(prefix="echoscript_"))

    detected_language = SUPPORTED_LANGUAGES.get(transcript.language, transcript.language)
    dashboard_md = (
        f"### \u2713 {detected_language} detected\n\n"
        f"**Confidence:** {transcript.language_probability:.0%}&nbsp;&nbsp;&nbsp;"
        f"**Duration:** {_format_duration(transcript.duration)}&nbsp;&nbsp;&nbsp;"
        f"**Words:** {transcript.word_count:,}"
    )

    transcript_text = transcript.text
    transcript_file = _write_text_file(transcript_text, tmp_dir, "transcript.txt")

    # Now that the actual language is known -- the whole point of doing
    # this as its own step -- compute the translate-to picker against it,
    # never against a pre-detection guess.
    has_key = bool((api_key or "").strip())
    translation_choices = _compute_translation_choices(has_key, transcript.language)
    default_targets = ["English"] if "English" in translation_choices else []
    translate_choices_update = gr.update(choices=translation_choices, value=default_targets)

    srt_path = _write_text_file(generate_srt(transcript.segments), tmp_dir, "transcript.srt")
    vtt_path = _write_text_file(generate_vtt(transcript.segments), tmp_dir, "transcript.vtt")

    # Per-language result sections: only reset (hide + clear) if the
    # transcript actually changed. If it was reused, leave whatever
    # translations are already showing exactly as they are.
    section_outputs = []
    if regenerated:
        for _ in _TRANSLATION_SECTION_ORDER:
            section_outputs += [gr.update(visible=False), "", None]
    else:
        for _ in _TRANSLATION_SECTION_ORDER:
            section_outputs += [gr.update(), gr.update(), gr.update()]

    outputs = [
        dashboard_md,
        transcript_text,
        transcript_file,
        translate_choices_update,
        gr.update(visible=True),   # reveal the "choose languages" picker
        gr.update(visible=False),  # hide the "generate a transcript first" placeholder
    ]
    outputs += section_outputs
    outputs += [srt_path, vtt_path, transcript, cached_signature, cached_translations]
    return outputs


# ---------------------------------------------------------------------------
# Alternate Stage 1: Generate Phonetic Transcription (IPA)
#
# Mutually exclusive with "Generate Transcript" -- this mode never touches
# Whisper, never produces a Transcript, and therefore never feeds
# translations or subtitles. It exists precisely so the Transcript can
# stay the single source of truth for everything downstream of it: if you
# want IPA phones, you get *only* IPA phones from this click, not a
# transcript-plus-phonetics bundle. See services/phonetics.py for why this
# needs to read the audio directly rather than derive from a transcript.
# ---------------------------------------------------------------------------

def generate_phonetics(
    audio_path: Optional[str],
    start_value: str,
    end_value: str,
    cached_phonetics: Optional[str],
    cached_phonetics_signature,
):
    if not audio_path:
        raise gr.Error("Please upload an audio file first.")

    try:
        validate_extension(audio_path)
        start, end = resolve_window(start_value, end_value)
    except AudioError as exc:
        raise gr.Error(str(exc)) from exc

    signature = (audio_path, start, end)
    if cached_phonetics is not None and cached_phonetics_signature == signature:
        # Same audio and window as last time -- reuse rather than
        # re-running the phone recognizer.
        phonetics_text = cached_phonetics
    else:
        working_path = audio_path
        if start is not None or end is not None:
            try:
                working_path = extract_window(audio_path, start, end)
            except AudioError as exc:
                raise gr.Error(str(exc)) from exc

        try:
            phonetics_text = transcribe_phonetics(working_path)
        except PhoneticsError as exc:
            phonetics_text = f"\u26a0\ufe0f Phonetic transcription failed: {exc}"
        cached_phonetics_signature = signature

    tmp_dir = Path(tempfile.mkdtemp(prefix="echoscript_"))
    phonetics_file = _write_text_file(phonetics_text, tmp_dir, "phonetics.txt")

    return [phonetics_text, phonetics_file, phonetics_text, cached_phonetics_signature]


# ---------------------------------------------------------------------------
# Stage 2: Generate Translations
# ---------------------------------------------------------------------------

def generate_translations(
    selected_languages: list[str],
    api_key: str,
    cached_transcript: Optional[Transcript],
    cached_translations: dict,
):
    """Generator: yield cached results immediately, then compute only new ones.

    This ensures the loading spinner only appears on sections that are
    actually being translated. Languages already in the cache are yielded
    instantly in the first pass; only genuinely new languages trigger
    model/API calls in the second pass. Gradio generators allow partial
    yields, so the UI updates progressively rather than waiting for the
    slowest language.
    """
    if cached_transcript is None:
        raise gr.Error("Generate a transcript first.")

    selected = set(selected_languages or [])
    translation_service = get_translation_service()
    tmp_dir = Path(tempfile.mkdtemp(prefix="echoscript_"))

    def _make_outputs(section_states: dict) -> list:
        """Build the flat output list from a dict of label -> (visible, text, file)."""
        result = []
        for label in _TRANSLATION_SECTION_ORDER:
            state = section_states.get(label)
            if state is None:
                # No decision yet for this label -- emit a no-op so Gradio
                # doesn't touch it (preserves whatever is already shown).
                result += [gr.update(), gr.update(), gr.update()]
            else:
                visible, text, file_path = state
                result += [gr.update(visible=visible), text if text is not None else gr.update(), file_path]
        result.append(cached_translations)
        return result

    # ------------------------------------------------------------------
    # Pass 1: Resolve every section immediately from the cache or by
    # hiding unselected ones. Sections that need a real translation show
    # a "⏳ Translating..." placeholder so the user sees all boxes right
    # away rather than having to wait for each one to appear.
    # ------------------------------------------------------------------
    section_states: dict[str, Optional[tuple]] = {}
    needs_translation: list[str] = []

    for label in _TRANSLATION_SECTION_ORDER:
        code = _TRANSLATION_LABEL_TO_CODE[label]
        if label not in selected:
            section_states[label] = (False, "", None)
        elif code in cached_translations:
            text = cached_translations[code]
            file_path = _write_text_file(text, tmp_dir, f"{code}_cached.txt")
            section_states[label] = (True, text, file_path)
        else:
            # Show the box immediately with a placeholder; fill it in pass 2.
            section_states[label] = (True, "⏳ Translating...", None)
            needs_translation.append(label)

    # Yield immediately so cached/placeholder results appear at once.
    yield _make_outputs(section_states)

    # ------------------------------------------------------------------
    # Pass 2: Translate only the languages that aren't cached yet,
    # yielding after each one completes.
    # ------------------------------------------------------------------
    for label in needs_translation:
        code = _TRANSLATION_LABEL_TO_CODE[label]
        try:
            translation = translation_service.translate(cached_transcript, code, api_key=api_key)
            text = translation.text
            cached_translations[code] = text
            file_path = _write_text_file(text, tmp_dir, f"{code}.txt")
            section_states[label] = (True, text, file_path)
        except TranslationError as exc:
            # Surface the failure for this language only. Deliberately not
            # cached, so the next click will retry.
            text = f"\u26a0\ufe0f Translation failed: {exc}"
            section_states[label] = (True, text, None)

        # Yield after each language so the UI updates progressively.
        yield _make_outputs(section_states)


def reset_transcript_state():
    """Clear cached transcript/translations and everything on screen for that mode."""
    ui_reset = [
        "Upload an audio file and click **Generate Transcript** to begin.",
        "",
        None,
        gr.update(choices=[], value=[]),
        gr.update(visible=False),
        gr.update(visible=True),
    ]
    for _ in _TRANSLATION_SECTION_ORDER:
        ui_reset += [gr.update(visible=False), "", None]
    ui_reset += [None, None, None, None, {}]
    return ui_reset


def reset_phonetics_state():
    """Clear cached phonetics and everything on screen for that mode."""
    return ["", None, None, None]


# ---------------------------------------------------------------------------
# UI layout
# ---------------------------------------------------------------------------

with gr.Blocks(title="EchoScript") as demo:

    transcript_state = gr.State(value=None)
    signature_state = gr.State(value=None)
    translations_state = gr.State(value={})
    phonetics_state = gr.State(value=None)
    phonetics_signature_state = gr.State(value=None)

    gr.Markdown(
        """
# EchoScript

**Upload Audio &rarr; Choose Transcript or Phonetics (independent pipelines) &rarr; Preview &
Choose Languages (Transcript mode only) &rarr; Generate Translations &rarr; Copy / Download**

<sub>build: 2026-07-06 22:10 UTC &middot; Transcript and Phonetics are now independent, mutually exclusive pipelines</sub>
        """
    )

    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### Upload Audio")
            audio_input = gr.Audio(
                label="Upload a file or record from microphone",
                sources=["upload", "microphone"],
                type="filepath",
                format="wav",
            )
            gr.Markdown("Upload: mp3 &middot; wav &middot; m4a &middot; flac &nbsp;|&nbsp; Microphone: recorded as wav")

            gr.Markdown("### Processing Window")
            with gr.Row():
                start_input = gr.Textbox(label="Start Time (optional)", placeholder="HH:MM:SS")
                end_input = gr.Textbox(label="End Time (optional)", placeholder="HH:MM:SS")
            gr.Markdown("Leave blank: entire file")

            gr.Markdown("### What do you want to generate?")
            gr.Markdown(
                "These are two independent pipelines -- choose one per click. "
                "**Transcript** is the source of truth for translations, subtitles, "
                "and editing. **Phonetics** reads the audio directly and has no "
                "connection to the transcript at all -- it won't 'correct' toward "
                "real words the way a transcript does, and picking it here never "
                "runs (or requires) the transcript pipeline."
            )

            with gr.Tabs():
                with gr.Tab("Transcript & Translations"):
                    language_input = gr.Dropdown(
                        choices=SOURCE_LANGUAGE_CHOICES,
                        value="Auto Detect",
                        label="Source Language",
                        info="A hint for transcription, not a guess at translation targets.",
                    )
                    api_key_input = gr.Textbox(
                        label="Anthropic API Key (optional)",
                        type="password",
                        placeholder="sk-ant-...",
                        info=(
                            "Provide your own key to translate into many more languages via "
                            "Claude. Without one, translation uses local offline models "
                            "(English, German, Persian, Spanish, and more). Used for this "
                            "session only -- never stored."
                        ),
                    )
                    generate_transcript_button = gr.Button("Generate Transcript", variant="primary")
                    reset_transcript_button = gr.Button("Reset Transcript (clear cache)", size="sm")

                with gr.Tab("Phonetic Transcription (IPA)"):
                    gr.Markdown(
                        "Produces the exact IPA sounds heard in the audio -- "
                        "independent of language, and independent of the transcript."
                    )
                    generate_phonetics_button = gr.Button("Generate Phonetic Transcription", variant="primary")
                    reset_phonetics_button = gr.Button("Reset Phonetics (clear cache)", size="sm")

        with gr.Column(scale=2):
            gr.Markdown("### Results Dashboard")
            dashboard_output = gr.Markdown("Upload an audio file and click **Generate Transcript** to begin.")

            with gr.Tabs():
                with gr.Tab("Transcript"):
                    transcript_box = gr.Textbox(
                        label="Transcript",
                        lines=16,
                        interactive=True,
                        buttons=["copy"],
                    )
                    transcript_download = gr.DownloadButton("Download TXT")

                with gr.Tab("Phonetics (IPA)"):
                    gr.Markdown(
                        "The exact sounds heard in the audio, written as IPA phones -- "
                        "independent of any language. This is **not** derived from the "
                        "transcript: it comes from a universal phone recognizer reading "
                        "the audio directly, so it won't correct itself toward real words "
                        "the way the transcript does."
                    )
                    phonetics_box = gr.Textbox(
                        label="Phonetic transcription (IPA)",
                        lines=10,
                        interactive=True,
                        buttons=["copy"],
                    )
                    phonetics_download = gr.DownloadButton("Download TXT")

                with gr.Tab("Translations"):
                    translations_placeholder = gr.Markdown(
                        "Generate a transcript first to see the languages available to "
                        "translate it into."
                    )
                    with gr.Group(visible=False) as translations_picker_group:
                        gr.Markdown("Translate to:")
                        with gr.Row():
                            select_all_btn = gr.Button("Select All", size="sm")
                            select_none_btn = gr.Button("Deselect All", size="sm")
                        translate_choices_input = gr.CheckboxGroup(choices=[], value=[], label=None)
                        generate_translations_button = gr.Button("Generate Translations", variant="primary")

                    translation_groups = {}
                    translation_boxes = {}
                    translation_downloads = {}
                    for label in _TRANSLATION_SECTION_ORDER:
                        short_name = label.replace(" Translation", "")
                        with gr.Group(visible=False) as group:
                            box = gr.Textbox(
                                label=short_name,
                                lines=10,
                                interactive=True,
                                buttons=["copy"],
                            )
                            download = gr.DownloadButton("Download TXT")
                        translation_groups[label] = group
                        translation_boxes[label] = box
                        translation_downloads[label] = download

                with gr.Tab("Subtitles"):
                    gr.Markdown(
                        "Subtitles are generated from the transcript "
                        "(source language) as soon as it's ready -- no "
                        "translation needed."
                    )
                    with gr.Row():
                        srt_download = gr.DownloadButton("Download SRT")
                        vtt_download = gr.DownloadButton("Download VTT")

    # Outputs for the Transcript & Translations mode (and its Reset).
    transcript_stage_outputs = [
        dashboard_output,
        transcript_box,
        transcript_download,
        translate_choices_input,
        translations_picker_group,
        translations_placeholder,
    ]
    for label in _TRANSLATION_SECTION_ORDER:
        transcript_stage_outputs += [
            translation_groups[label],
            translation_boxes[label],
            translation_downloads[label],
        ]
    transcript_stage_outputs += [
        srt_download,
        vtt_download,
        transcript_state,
        signature_state,
        translations_state,
    ]

    generate_transcript_button.click(
        fn=generate_transcript,
        inputs=[
            audio_input,
            start_input,
            end_input,
            language_input,
            api_key_input,
            transcript_state,
            signature_state,
            translations_state,
        ],
        outputs=transcript_stage_outputs,
    )
    reset_transcript_button.click(fn=reset_transcript_state, outputs=transcript_stage_outputs)

    # Outputs for the Phonetics mode (and its Reset) -- entirely separate
    # from the transcript pipeline; sharing only the audio/window inputs.
    phonetics_stage_outputs = [phonetics_box, phonetics_download, phonetics_state, phonetics_signature_state]

    generate_phonetics_button.click(
        fn=generate_phonetics,
        inputs=[audio_input, start_input, end_input, phonetics_state, phonetics_signature_state],
        outputs=phonetics_stage_outputs,
    )
    reset_phonetics_button.click(fn=reset_phonetics_state, outputs=phonetics_stage_outputs)

    # Outputs for Stage 2 (Generate Translations): just the per-language
    # sections plus the translation cache.
    translation_stage_outputs = []
    for label in _TRANSLATION_SECTION_ORDER:
        translation_stage_outputs += [
            translation_groups[label],
            translation_boxes[label],
            translation_downloads[label],
        ]
    translation_stage_outputs += [translations_state]

    generate_translations_button.click(
        fn=generate_translations,
        inputs=[translate_choices_input, api_key_input, transcript_state, translations_state],
        outputs=translation_stage_outputs,
    )

    # Cascade the translate-to picker live as the key changes, once a
    # transcript exists (no-op before that -- the picker isn't shown yet).
    api_key_input.input(
        fn=_on_api_key_change,
        inputs=[api_key_input, transcript_state, translate_choices_input],
        outputs=[translate_choices_input],
    )
    api_key_input.change(
        fn=_on_api_key_change,
        inputs=[api_key_input, transcript_state, translate_choices_input],
        outputs=[translate_choices_input],
    )

    select_all_btn.click(
        fn=lambda choices: gr.update(value=choices),
        inputs=[translate_choices_input],
        outputs=[translate_choices_input],
    )
    select_none_btn.click(
        fn=lambda: gr.update(value=[]),
        outputs=[translate_choices_input],
    )

if __name__ == "__main__":
    demo.launch()