app.py

from __future__ import annotations

import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
from pathlib import Path

import gradio as gr
from dotenv import load_dotenv

load_dotenv()

MAIN = Path(__file__).parent / "main.py"

_ANSI = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]|\r")


_CLEANUP_DELAY = 30 * 60  # seconds to keep the output MKV available for download


def _strip(text: str) -> str:
    return _ANSI.sub("", text)


def _delete_later(path: str, delay: float = _CLEANUP_DELAY) -> None:
    """Delete a file or directory after `delay` seconds on a daemon thread."""
    def _do() -> None:
        shutil.rmtree(path, ignore_errors=True)
    threading.Timer(delay, _do).start()


def run(
    url: str,
    input_video: str | None,
    language: str,
    translate_to: str,
    resolution: int,
    max_length: int,
):
    source = (input_video or "").strip()
    url = (url or "").strip()

    if not source and not url:
        yield "Error: provide a YouTube URL or upload a video file.\n", gr.update(visible=False)
        return
    if source and url:
        yield "Error: provide either a YouTube URL or a video file, not both.\n", gr.update(visible=False)
        return

    tmpdir = tempfile.mkdtemp(prefix="transcriber-")

    cmd = [sys.executable, str(MAIN)]
    if source:
        cmd += ["--input-video", source]
    else:
        cmd += [url]

    cmd += ["--output-dir", tmpdir]
    if (language or "").strip():
        cmd += ["--language", language.strip()]
    if (translate_to or "").strip():
        cmd += ["--translate-to", translate_to.strip()]
    if resolution != 720:
        cmd += ["--resolution", str(int(resolution))]
    if max_length != 100:
        cmd += ["--max-length", str(int(max_length))]
    env = {**os.environ, "NO_COLOR": "1"}
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        env=env,
    )

    lines: list[str] = []
    try:
        for line in proc.stdout:  # type: ignore[union-attr]
            clean = _strip(line)
            if not clean.strip():
                continue  # skip blank lines and yt-dlp progress-bar clear lines
            # Collapse consecutive yt-dlp [download] progress lines in-place
            if clean.startswith("[download]") and lines and lines[-1].startswith("[download]"):
                lines[-1] = clean
            else:
                lines.append(clean)
            yield "".join(lines), gr.update(visible=False)
    except GeneratorExit:
        proc.terminate()
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()
        shutil.rmtree(tmpdir, ignore_errors=True)
        return
    finally:
        proc.wait()

    mkv_files = list(Path(tmpdir).glob("*.mkv"))
    if proc.returncode == 0 and mkv_files:
        # Copy MKV into a named temp dir so the download filename matches the
        # original title. Schedule the whole dir for deletion after the delay.
        out_dir = tempfile.mkdtemp(prefix="transcriber-out-")
        out_path = os.path.join(out_dir, mkv_files[0].name)
        shutil.copy2(mkv_files[0], out_path)
        shutil.rmtree(tmpdir, ignore_errors=True)
        _delete_later(out_dir)
        yield "".join(lines), gr.update(visible=True, value=out_path)
    else:
        shutil.rmtree(tmpdir, ignore_errors=True)
        if proc.returncode != 0:
            lines.append(f"\nProcess exited with code {proc.returncode}.\n")
        else:
            lines.append("\nNo output file produced.\n")
        yield "".join(lines), gr.update(visible=False)


_CSS = """
#log textarea { font-family: monospace; font-size: 12px; }
"""

with gr.Blocks(title="transcriber", delete_cache=(_CLEANUP_DELAY, _CLEANUP_DELAY)) as demo:
    browser_state = gr.BrowserState(
        {"language": "", "translate_to": ""},
        storage_key="transcriber-prefs",
        secret="transcriber-2026",
    )

    gr.Markdown("## transcriber\nTranscribe a YouTube video and download it with embedded subtitles.")

    with gr.Row():
        url_box = gr.Textbox(
            label="YouTube URL",
            placeholder="https://www.youtube.com/watch?v=...",
            scale=3,
        )
        file_input = gr.File(
            label="Or upload a local video",
            file_types=[".mp4", ".mkv", ".mov", ".avi", ".webm"],
            scale=1,
        )

    with gr.Accordion("Options", open=False):
        with gr.Row():
            language = gr.Textbox(
                label="Source language (ISO 639-3)",
                placeholder="e.g. srb; blank = auto-detect",
            )
            translate_to = gr.Textbox(
                label="Add subtitles translated to (ISO 639-3)",
                placeholder="e.g. rus; blank = skip translation",
            )
            resolution = gr.Slider(144, 2160, step=1, value=720, label="Video size on shorter side (px)")
            max_length = gr.Slider(40, 200, step=1, value=100, label="Max chars per subtitle cue")

    with gr.Row():
        run_btn = gr.Button("Run", variant="primary")
        stop_btn = gr.Button("Stop", variant="stop")

    log_box = gr.Textbox(
        lines=15,
        max_lines=40,
        interactive=False,
        autoscroll=True,
        label="Progress",
        elem_id="log",
    )
    output_file = gr.DownloadButton(label="Download MKV", visible=False, variant="primary", size="lg")

    # Restore persisted language settings on page load
    @demo.load(inputs=[browser_state], outputs=[language, translate_to])
    def _restore(state: dict) -> tuple[str, str]:
        return state.get("language", ""), state.get("translate_to", "")

    # Save language settings whenever they change
    @gr.on(
        [language.change, translate_to.change],
        inputs=[language, translate_to],
        outputs=[browser_state],
    )
    def _save(lang: str, tl: str) -> dict:
        return {"language": lang, "translate_to": tl}

    run_event = run_btn.click(
        fn=run,
        inputs=[url_box, file_input, language, translate_to, resolution, max_length],
        outputs=[log_box, output_file],
        show_progress="full",
    )
    stop_btn.click(fn=None, cancels=[run_event])

demo.queue()

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