diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..42a6538 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,32 @@ +# Git +.git +.gitignore + +# Python bytecode and caches +__pycache__ +*.py[cod] +*.pyo +.pytest_cache +.mypy_cache +.ruff_cache + +# uv caches (built inside the image; don't copy from host) +.uv_cache + +# Virtual environments +.venv + +# Output and upload dirs (mounted as volumes at runtime) +apps/backend/outs +apps/backend/uploads +apps/backend/cache + +# Dev / editor +.env +*.env.local +.vscode +.idea + +# Docs +CONTRIBUTING.md +README.md diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml new file mode 100644 index 0000000..f7176f8 --- /dev/null +++ b/.github/workflows/docker.yml @@ -0,0 +1,67 @@ +name: Docker + +on: + push: + branches: + - main + - develop + - feature/** + pull_request: + branches: + - "*" + +jobs: + build-backend: + runs-on: ubuntu-latest + timeout-minutes: 60 + strategy: + matrix: + include: + - service: asr + target: final-asr + - service: translation + target: final-translation + - service: tts + target: final-tts + - service: orchestrator + target: final-orchestrator + fail-fast: false + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Build ${{ matrix.service }} image + uses: docker/build-push-action@v6 + with: + context: . + dockerfile: Dockerfile + target: final-${{ matrix.service }} + push: false + tags: ghcr.io/codemowers/bluez-dubbing/${{ matrix.service }}:${{ github.sha }} + cache-from: type=gha,scope=${{ matrix.service }} + cache-to: type=gha,scope=${{ matrix.service }},mode=max + + build-frontend: + runs-on: ubuntu-latest + timeout-minutes: 10 + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Build frontend image + uses: docker/build-push-action@v6 + with: + context: . + dockerfile: Dockerfile.frontend + push: false + tags: ghcr.io/codemowers/bluez-dubbing/frontend:${{ github.sha }} + cache-from: type=gha,scope=frontend + cache-to: type=gha,scope=frontend,mode=max diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..779fe44 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,80 @@ +# Single parameterised Dockerfile for the four backend services. +# Usage: +# docker build --build-arg SERVICE=orchestrator -t bluez-orchestrator . +# docker build --build-arg SERVICE=asr -t bluez-asr . +# docker build --build-arg SERVICE=translation -t bluez-translation . +# docker build --build-arg SERVICE=tts -t bluez-tts . +# +# The orchestrator uses beveradb/audio-separator as its base so that +# audio-separator, ffmpeg, rubberband, and torch ship pre-installed. +# All other services use python:3.11-slim. + +ARG SERVICE=orchestrator + +# ── orchestrator base ──────────────────────────────────────────────────────── +FROM beveradb/audio-separator:latest AS base-orchestrator +# audio-separator image ships Python 3.11, ffmpeg, rubberband, torch, onnxruntime. +# Install uv on top. +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +# ── all other services base ────────────────────────────────────────────────── +FROM python:3.11-slim AS base-default +RUN apt-get update && apt-get install -y --no-install-recommends \ + ffmpeg \ + rubberband-cli \ + git \ + && rm -rf /var/lib/apt/lists/* +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +# ── select base ───────────────────────────────────────────────────────────── +# Docker does not support conditional FROM, so we build both bases above and +# select via the SERVICE arg at final-stage copy time. The actual selection +# happens in the service-specific compose / k8s build targets. +# For a single-image build the default target is used; override with +# DOCKER_BUILDKIT=1 docker build --build-arg SERVICE=orchestrator \ +# --target final-orchestrator . + +FROM base-orchestrator AS final-orchestrator +WORKDIR /app +COPY . . +ENV UV_CACHE_DIR=/root/.cache/uv +RUN uv sync --frozen --project apps/backend/services/orchestrator +ENV PYTHONPATH=/app/apps/backend:/app +EXPOSE 8000 +CMD ["uv", "run", "--project", "apps/backend/services/orchestrator", \ + "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] + +FROM base-default AS final-asr +WORKDIR /app +COPY . . +ENV UV_CACHE_DIR=/root/.cache/uv +RUN uv sync --frozen --project apps/backend/services/asr \ + && uv sync --frozen --project apps/backend/services/asr/models/whisperxModel +ENV PYTHONPATH=/app/apps/backend:/app +EXPOSE 8000 +CMD ["uv", "run", "--project", "apps/backend/services/asr", \ + "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8001"] + +FROM base-default AS final-translation +WORKDIR /app +COPY . . +ENV UV_CACHE_DIR=/root/.cache/uv +RUN uv sync --frozen --project apps/backend/services/translation \ + && uv sync --frozen --project apps/backend/services/translation/models/deepTranslationModel \ + && uv sync --frozen --project apps/backend/services/translation/models/facebook_m2m100Model +ENV PYTHONPATH=/app/apps/backend:/app +EXPOSE 8000 +CMD ["uv", "run", "--project", "apps/backend/services/translation", \ + "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8002"] + +FROM base-default AS final-tts +WORKDIR /app +COPY . . +ENV UV_CACHE_DIR=/root/.cache/uv +RUN uv sync --frozen --project apps/backend/services/tts \ + && uv sync --frozen --project apps/backend/services/tts/models/chatterboxModel \ + && uv sync --frozen --project apps/backend/services/tts/models/edgeTTsModel +ENV PYTHONPATH=/app/apps/backend:/app +EXPOSE 8000 +CMD ["uv", "run", "--project", "apps/backend/services/tts", \ + "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8003"] diff --git a/Dockerfile.frontend b/Dockerfile.frontend new file mode 100644 index 0000000..c847608 --- /dev/null +++ b/Dockerfile.frontend @@ -0,0 +1,16 @@ +FROM nginx:1.27-alpine + +# Copy static assets +COPY apps/frontend /usr/share/nginx/html + +# nginx config: listen on 8080 (non-root compatible), proxy /api and /outs +COPY docker/nginx.conf /etc/nginx/conf.d/default.conf + +# Allow nginx to bind 8080 as non-root +RUN chown -R nginx:nginx /var/cache/nginx /var/log/nginx /etc/nginx/conf.d \ + && chmod -R g+w /var/cache/nginx /var/log/nginx \ + && touch /var/run/nginx.pid \ + && chown nginx:nginx /var/run/nginx.pid + +USER nginx +EXPOSE 8080 diff --git a/apps/backend/services/orchestrator/app/main.py b/apps/backend/services/orchestrator/app/main.py index 5b1e553..f2bad7a 100644 --- a/apps/backend/services/orchestrator/app/main.py +++ b/apps/backend/services/orchestrator/app/main.py @@ -76,9 +76,7 @@ logger = logging.getLogger("bluez.orchestrator") if not logger.handlers: - # Read from control center config (default to INFO if not set) log_level = general_cfg.get("log_level", "INFO").upper() - # Configure logging logging.basicConfig(level=getattr(logging, log_level, logging.INFO)) API_PREFIX = "/api" @@ -105,9 +103,11 @@ allow_headers=["*"], ) -ASR_URL = "http://localhost:8001/v1/transcribe" -TR_URL = "http://localhost:8002/v1/translate" -TTS_URL = "http://localhost:8003/v1/synthesize" +# Service URLs — override via environment for Docker / container deployments. +# Defaults keep the original localhost:800x behaviour for bare-metal / Makefile use. +ASR_URL = os.getenv("ASR_URL", "http://localhost:8001/v1/transcribe") +TR_URL = os.getenv("TR_URL", "http://localhost:8002/v1/translate") +TTS_URL = os.getenv("TTS_URL", "http://localhost:8003/v1/synthesize") OUTS = BASE / "outs" SEPARATION_CACHE = BASE / "cache" / "audio_separation" @@ -419,7 +419,6 @@ def emit_progress(event: Dict[str, Any]) -> None: #-------------------------------------------------------------------------------------------------------------# async def persist_uploaded_file(file: UploadFile, uploads_dir: Path) -> Path: - # Offload large upload hashing/copying work so we don't block the event loop for UI requests. return await asyncio.to_thread(_persist_uploaded_file_sync, file, uploads_dir) async def load_cached_raw_audio(cache_key: str, target_path: Path) -> bool: @@ -539,7 +538,6 @@ def _run(): async def has_video_stream(path: str | Path) -> bool: p = Path(path) - # quick short-circuit by extension if p.suffix.lower() in AUDIO_EXTENSIONS: return False cmd = [ @@ -692,7 +690,6 @@ async def extract_audio_to_workspace(source_url: str, raw_audio_path: Path) -> N raise HTTPException(500, "Audio extraction produced an empty file") -# find a way to delete the cached files after some time or size limit async def load_cached_separation(cache_key: str, vocals_target: Path, background_target: Path) -> bool: cache_dir = SEPARATION_CACHE / cache_key vocals_cache = cache_dir / "vocals.wav" @@ -711,7 +708,6 @@ async def store_separation_cache(cache_key: str, vocals_source: Path, background await run_in_thread(shutil.copy, vocals_source, cache_dir / "vocals.wav") await run_in_thread(shutil.copy, background_source, cache_dir / "background.wav") -# see if we can implement automatic noise level detection in future improvements work. to judge if separation is needed for some tasks async def maybe_run_audio_separation( preprocessing_dir: Path, raw_audio_path: Path, @@ -719,7 +715,7 @@ async def maybe_run_audio_separation( audio_sep: bool, dubbing_strategy: str, ) -> Tuple[Optional[Path], Optional[Path], str]: - if not audio_sep and dubbing_strategy != "full_replacement": # here we just supposed that if audio separation is disabled and the strategy is not full replacement we don't need to do separation + if not audio_sep and dubbing_strategy != "full_replacement": return None, None, dubbing_strategy vocals_path = preprocessing_dir / "vocals.wav" @@ -895,7 +891,7 @@ async def run_tts_review_session( if key in TTS_REVIEW_SESSIONS: raise HTTPException(409, "TTS review already in progress for this run and language.") - translation_map: Dict[str, Segment] = {seg.segment_id: seg for seg in translation.segments if seg.segment_id} # used to access the value linked to the id in 0(1) + translation_map: Dict[str, Segment] = {seg.segment_id: seg for seg in translation.segments if seg.segment_id} segments_state: Dict[str, SegmentAudioOut] = { seg.segment_id: seg for seg in tts_result.segments if seg.segment_id} worker = TTS_WORKERS.get(tts_model) @@ -974,7 +970,7 @@ async def regenerate_tts_segment_audio( segment_id: str, text: str, lang: Optional[str], - audio_prompt_url: Optional[str] = None, # if we want to give different audio prompt than the original one + audio_prompt_url: Optional[str] = None, ) -> SegmentAudioOut: seg_lock = session.segment_locks.get(segment_id) if seg_lock is None: @@ -1149,635 +1145,178 @@ async def finalize_media( -@app.on_event("startup") -async def startup_event() -> None: - timeout = httpx.Timeout(connect=10.0, read=1200.0, write=10.0, pool=None) # think changing the value of the pool in the future for more robustness - app.state.http_client = httpx.AsyncClient(timeout=timeout) - OUTS.mkdir(parents=True, exist_ok=True) - SEPARATION_CACHE.mkdir(parents=True, exist_ok=True) - RAW_AUDIO_CACHE.mkdir(parents=True, exist_ok=True) - UPLOADS_DIR.mkdir(parents=True, exist_ok=True) - +async def dub( + video_url: str, + target_work: str = Query( + "dub", + description="Target work type, e.g., 'dub': for full dubbing or 'sub': for subtitles only", + ), + target_langs: Optional[List[str] | str] = Query(None), + source_lang: Optional[str] = None, + min_speakers: Optional[int] = None, + max_speakers: Optional[int] = None, + sep_model: str = Query("melband_roformer_big_beta5e.ckpt"), + asr_model: str = Query("whisperx"), + tr_model: str = Query("facebook_m2m100"), + tts_model: str = Query("chatterbox"), + audio_sep: bool = Query(True, description="Whether to perform audio source separation"), + perform_vad_trimming: bool = Query(True, description="Whether to perform VAD-based silence trimming after TTS"), + translation_strategy: str = Query( + "default", + description="Translation strategy to use: either translate directly over the short ASR aligned segments or translate the full text and then align the translated result after", + ), + dubbing_strategy: str = Query( + "default", + description="Dubbing strategy to use, either translation over (original audio ducked) or full replacement", + ), + sophisticated_dub_timing: bool = Query( + True, + description="Whether to use sophisticated timing for full replacement dubbing strategy", + ), + subtitle_style: Optional[str] = Query( + None, + description="Subtitle style preset: default, minimal, bold, netflix", + ), + persist_intermediate: bool = Query( + True, + description="Persist intermediate artifacts (disable for lower latency and disk usage)", + ), + involve_mode: bool = Query( + False, + description="Enable involve-mode workflow with manual transcription review between stages.", + ), + run_id: Optional[str] = Query( + None, + description="Optional run identifier when invoked from the job runner (required for involve mode).", + ), +): + """ + Complete dubbing pipeline orchestrator. + """ -@app.on_event("shutdown") -async def shutdown_event() -> None: - client = getattr(app.state, "http_client", None) - if client: - await client.aclose() + if involve_mode and not run_id: + raise HTTPException(400, "Involve mode requires an active run context (run_id).") + original_source = video_url + video_url = video_url.strip() + if target_work == "sub" and subtitle_style is None: + subtitle_style = "default_mobile" + if dubbing_strategy != "translation_over": + dubbing_strategy = "full_replacement" # other values default to full_replacement -@app.get(OPTIONS_ROUTE) -async def pipeline_options() -> JSONResponse: - return JSONResponse( - { - "asr_models": list_worker_models(ASR_WORKERS), - "translation_models": list_worker_models(TR_WORKERS), - "tts_models": list_worker_models(TTS_WORKERS), - "audio_separation_models": list_audio_separation_models(), - "translation_strategies": TRANSLATION_STRATEGIES, - "dubbing_strategies": DUBBING_STRATEGIES, - "subtitle_styles": sorted(STYLE_PRESETS.keys()), - } - ) + source_lang = (source_lang or "").strip() or None + target_languages = normalize_language_codes((list(target_langs)) if target_langs else []) + sep_model = (sep_model or "").strip() -@app.get(FILE_ROUTE) -async def pipeline_file(path: str) -> FileResponse: - resolved = Path(path).resolve() - try: - resolved.relative_to(OUTS) - except ValueError as exc: - raise HTTPException(403, "Invalid path") from exc + requested_tr_model = (tr_model or "").strip() + requested_tts_model = (tts_model or "").strip() - if not resolved.exists() or not resolved.is_file(): - raise HTTPException(404, "File not found") - return FileResponse(resolved) + asr_model = resolve_model_choice(asr_model, ASR_WORKERS, source_lang, fallback= general_cfg.get("default_models", {}).get("asr", "whisperx")) + translation_models_by_lang: Dict[str, str] = {} + tts_models_by_lang: Dict[str, str] = {} + per_language_models: Dict[str, Dict[str, str]] = {} + for lang in target_languages: + translation_models_by_lang[lang] = resolve_model_choice( + requested_tr_model, + TR_WORKERS, + lang or source_lang, + fallback= general_cfg.get("default_models", {}).get("tr", "deep_translator"), + ) + tts_models_by_lang[lang] = resolve_model_choice( + requested_tts_model, + TTS_WORKERS, + lang, + fallback= general_cfg.get("default_models", {}).get("tts", "chatterbox"), + ) -@app.post(RELEASE_ROUTE) -async def pipeline_release_media(token: str = Form(...)) -> JSONResponse: - token = (token or "").strip() - if not token: - return JSONResponse({"status": "ignored"}) - try: - target = resolve_cached_media_token(token) - except HTTPException: - return JSONResponse({"status": "invalid"}, status_code=200) - if target.exists(): - cleanup_cached_media_path(target) - return JSONResponse({"status": "released"}) - return JSONResponse({"status": "missing"}) + selected_models = { + "asr": asr_model, + "translation": translation_models_by_lang, + "tts": tts_models_by_lang, + "separation": sep_model, + } + if translation_strategy not in TRANSLATION_STRATEGIES: + translation_strategy = TRANSLATION_STRATEGIES[0] + if dubbing_strategy not in DUBBING_STRATEGIES: + dubbing_strategy = DUBBING_STRATEGIES[0] -@app.post(STOP_ROUTE) -async def pipeline_stop(run_id: str = Form(...)) -> JSONResponse: - run_id = (run_id or "").strip() - if not run_id: - raise HTTPException(400, "run_id is required") - task = ACTIVE_JOBS.get(run_id) - if task is None: - return JSONResponse({"status": "missing"}) - if task.done(): - ACTIVE_JOBS.pop(run_id, None) - return JSONResponse({"status": "completed"}) - task.cancel() - return JSONResponse({"status": "cancelling"}) + workspace = WorkspaceManager.create(OUTS, persist_intermediate) + step_timer = StepTimer() + client = get_http_client() + subtitles_dir: Optional[Path] = None -@app.post(f"{JOBS_PREFIX}/transcription_review") -async def pipeline_submit_transcription_review(review: TranscriptionReviewRequest) -> JSONResponse: - run_id = (review.run_id or "").strip() - if not run_id: - raise HTTPException(400, "run_id is required") - future = TRANSCRIPTION_REVIEW_WAITERS.get(run_id) - if future is None: - raise HTTPException(404, "No pending transcription review for this run.") - if future.done(): - raise HTTPException(409, "Transcription review already submitted for this run.") - session = TRANSCRIPTION_REVIEW_SESSIONS.get(run_id) - audio_duration = session.audio_duration if session else None - tolerance = session.tolerance if session else TRANSCRIPTION_SEGMENT_TOLERANCE - if audio_duration is not None and audio_duration <= tolerance: - audio_duration = None + resolved_video_path, media_digest = await prepare_media_source(video_url, workspace) + source_media_local_path: Optional[Path] = resolved_video_path - allowed_languages = set((session.languages or []) if session else []) - segments = review.transcription.segments or [] - sanitized: List[Segment] = [] - for segment in segments: - start_raw = segment.start if segment.start is not None else 0.0 - end_raw = segment.end if segment.end is not None else start_raw - try: - start = float(start_raw) - except (TypeError, ValueError): - raise HTTPException(400, f"Invalid start time for segment {segment.segment_id or '?'}") - try: - end = float(end_raw) - except (TypeError, ValueError): - raise HTTPException(400, f"Invalid end time for segment {segment.segment_id or '?'}") - if audio_duration is not None: - if start < -tolerance or start > audio_duration + tolerance: - raise HTTPException( - 400, - f"Segment start {start:.3f}s exceeds audio bounds (duration {audio_duration:.3f}s).", - ) - if end < -tolerance or end > audio_duration + tolerance: - raise HTTPException( - 400, - f"Segment end {end:.3f}s exceeds audio bounds (duration {audio_duration:.3f}s).", - ) - start = max(0.0, min(start, audio_duration + tolerance)) - end = max(0.0, min(end, audio_duration + tolerance)) - if end < start: - if end + tolerance >= start: - end = start - else: - raise HTTPException(400, f"Segment end {end:.3f}s precedes start {start:.3f}s.") - text = (segment.text or "").strip() - lang = (segment.lang or "").strip().lower() or None - if lang: - allowed_languages.add(lang) - segment.start = round(start, 3) - segment.end = round(end, 3) - segment.text = text - segment.lang = lang - segment.words = None - sanitized.append(segment) - sanitized.sort(key=lambda seg: seg.start if seg.start is not None else 0.0) - review.transcription.segments = sanitized - ensure_segment_ids(review.transcription) - TRANSCRIPTION_REVIEW_SESSIONS.pop(run_id, None) - future.set_result(review.transcription) - return JSONResponse({"status": "accepted"}) + source_has_video = await has_video_stream(resolved_video_path) # better check to avoid issues with audio-only inputs being treated as videos + if not source_has_video: + target_work = "dub" + subtitle_style = None + dubbing_strategy = "full_replacement" -@app.post(f"{JOBS_PREFIX}/alignment_review") -async def pipeline_submit_alignment_review(review: AlignmentReviewRequest) -> JSONResponse: - run_id = (review.run_id or "").strip() - if not run_id: - raise HTTPException(400, "run_id is required") - future = ALIGNMENT_REVIEW_WAITERS.get(run_id) - if future is None: - raise HTTPException(404, "No pending alignment review for this run.") - if future.done(): - raise HTTPException(409, "Alignment review already submitted for this run.") - future.set_result(review.alignment) - return JSONResponse({"status": "accepted"}) + if workspace.persist_intermediate: + preprocessing_dir = workspace.ensure_dir("preprocessing") + else: + preprocessing_dir = workspace.make_temp_dir("preprocessing") + raw_audio_path = preprocessing_dir / "raw_audio.wav" -@app.post(f"{JOBS_PREFIX}/tts_review") -async def pipeline_submit_tts_review(review: TTSReviewRequest) -> JSONResponse: - run_id = (review.run_id or "").strip() - if not run_id: - raise HTTPException(400, "run_id is required") - key = tts_session_key(run_id, review.language) - session = TTS_REVIEW_SESSIONS.get(key) - if session is None: - raise HTTPException(404, "No pending TTS review for this run and language.") - future = session.future - if future.done(): - raise HTTPException(409, "TTS review already submitted for this language.") - - updates = {item.segment_id: item for item in review.segments} - async with session.lock: - for segment in session.translation.values(): - if not segment.segment_id: - continue - update = updates.get(segment.segment_id) - if not update: - continue - segment.text = update.text.strip() - if update.lang: - segment.lang = update.lang - state = session.segments.get(segment.segment_id) - if state: - state.text = segment.text - state.lang = segment.lang or state.lang - for audio_segment in session.tts_response.segments: - if not audio_segment.segment_id: - continue - update = updates.get(audio_segment.segment_id) - if not update: - continue - if update.lang: - audio_segment.lang = update.lang - if not future.done(): - future.set_result(True) - emit_progress( - { - "type": "tts_review_complete", - "run_id": run_id, - "language": review.language, - } - ) - return JSONResponse({"status": "accepted"}) - - -@app.post(f"{JOBS_PREFIX}/tts_review/regenerate") -async def pipeline_regenerate_tts_segment( - run_id: str = Form(...), - segment_id: str = Form(...), - text: str = Form(...), - language: Optional[str] = Form(None), - lang: Optional[str] = Form(None), - audio_prompt_file: Optional[UploadFile] = File(None), -) -> JSONResponse: - run_id = (run_id or "").strip() - if not run_id: - raise HTTPException(400, "run_id is required") - if not segment_id: - raise HTTPException(400, "segment_id is required") - key = tts_session_key(run_id, language) - session = TTS_REVIEW_SESSIONS.get(key) - if session is None: - raise HTTPException(404, "No pending TTS review for this run and language.") - if session.future.done(): - raise HTTPException(409, "TTS review already submitted for this language.") - - # Handle custom audio prompt upload if provided - audio_prompt_url = None - if audio_prompt_file and audio_prompt_file.filename: - try: - # Store the uploaded audio file in the session workspace - audio_prompts_dir = session.workspace / "audio_prompts" - audio_prompts_dir.mkdir(parents=True, exist_ok=True) - - # Generate a safe filename - safe_name = f"{segment_id}_{safe_filename(audio_prompt_file.filename)}" - audio_prompt_path = audio_prompts_dir / safe_name - - # Save the uploaded file - with audio_prompt_path.open("wb") as dest: - content = await audio_prompt_file.read() - dest.write(content) - - audio_prompt_url = str(audio_prompt_path) - logger.info(f"Custom audio prompt uploaded for segment {segment_id}: {audio_prompt_url}") - except Exception as exc: - logger.warning(f"Failed to process audio prompt upload: {exc}") - # Continue without custom prompt if upload fails - - state = await regenerate_tts_segment_audio(session, segment_id, text, lang, audio_prompt_url) - segment_payload = serialize_tts_review_segment(state) - emit_progress( - { - "type": "tts_review_regenerated", - "run_id": run_id, - "language": language, - "segment": segment_payload, - } - ) - return JSONResponse({"status": "ok", "segment": segment_payload}) + cancelled = False + try: + raw_audio_cached = False + raw_audio_cache_token: Optional[str] = None + if media_digest: + raw_audio_cache_token = raw_audio_cache_key(media_digest) + raw_audio_cached = await load_cached_raw_audio(raw_audio_cache_token, raw_audio_path) + if raw_audio_cached: + logger.info("Loaded raw audio from cache for media digest %s", media_digest) -@app.post(RUN_ROUTE) -async def pipeline_run( - file: UploadFile | None = File(None), - video_url: Optional[str] = Form(None), - target_work: str = Form("dub"), - target_langs: Optional[List[str]] = Form(None), - source_lang: Optional[str] = Form(None), - min_speakers: Optional[int] = Form(None), - max_speakers: Optional[int] = Form(None), - reuse_media_token: Optional[str] = Form(None), - asr_model: Optional[str] = Form("auto"), - tr_model: Optional[str] = Form("auto"), - tts_model: Optional[str] = Form("auto"), - sep_model: Optional[str] = Form("auto"), - audio_sep: str = Form("true"), - perform_vad_trimming: str = Form("true"), - translation_strategy: str = Form("default"), - dubbing_strategy: str = Form("default"), - sophisticated_dub_timing: str = Form("true"), - subtitle_style: Optional[str] = Form(None), - persist_intermediate: str = Form("false"), - involve_mode: str = Form("false"), - ) -> StreamingResponse: - uploads_dir = UPLOADS_DIR - uploads_dir.mkdir(parents=True, exist_ok=True) # ensure uploads dir exists, just in case normally should be there already because of app startup + if not raw_audio_cached: + with step_timer.time("extract_audio"): + await extract_audio_to_workspace(str(resolved_video_path), raw_audio_path) + if raw_audio_cache_token: + await store_raw_audio_cache(raw_audio_cache_token, raw_audio_path) - provided_video_url = (video_url or "").strip() or None - source_media: Optional[str] = None - upload_path: Optional[Path] = None - upload_token: Optional[str] = None - remote_input_used = False - subtitle_style = (subtitle_style or "").strip() or None # since the Ui might send empty string we convert it to None here - if sep_model == "auto": - sep_model = general_cfg.get("default_models", {}).get("sep", "melband_roformer_big_beta5e.ckpt") + raw_audio_duration = get_audio_duration(raw_audio_path) - if file and file.filename: - upload_path = await persist_uploaded_file(file, uploads_dir) - source_media = str(upload_path) - upload_token = str(upload_path.relative_to(uploads_dir)) - elif provided_video_url: - source_media = provided_video_url - remote_input_used = True - elif reuse_media_token: - cached_path = resolve_cached_media_token(reuse_media_token) - if not cached_path.exists(): - raise HTTPException(400, "Cached media not found; please re-upload your file.") - source_media = str(cached_path) - upload_token = str(Path(reuse_media_token)) - else: - raise HTTPException(400, "Provide either a media file or a video link.") + vocals_path: Optional[Path] = None + background_path: Optional[Path] = None - run_id = str(uuid.uuid4()) - queue: asyncio.Queue[Dict[str, Any]] = asyncio.Queue() + vocal_for_transcript = general_cfg.get("vocal_only_for_transcription", True) + logger.info("Vocal only for transcription: %s", vocal_for_transcript) - def report(event: Dict[str, Any]) -> None: - try: - queue.put_nowait(event) - except asyncio.QueueFull: - logger.debug("Progress queue full; dropping event: %s", event) + if target_work != "sub" or vocal_for_transcript: # in subtitle-only mode, no need to separate audio for now: in future we might want to do it for better ASR performance if we succeed to implement automatic noise level detection + with step_timer.time("audio_separation"): - async def run_pipeline() -> None: - nonlocal upload_token - await queue.put({"type": "run_id", "run_id": run_id}) - token = PROGRESS_REPORTER.set(report) - try: - result = await dub( - video_url=str(source_media), - target_work=target_work, - target_langs=target_langs, - source_lang=source_lang, - min_speakers=min_speakers, - max_speakers=max_speakers, - sep_model=sep_model, - asr_model=asr_model, - tr_model=tr_model, - tts_model=tts_model, - audio_sep=parse_bool(audio_sep), - perform_vad_trimming=parse_bool(perform_vad_trimming), - translation_strategy=translation_strategy, - dubbing_strategy=dubbing_strategy, - sophisticated_dub_timing=parse_bool(sophisticated_dub_timing), - subtitle_style=subtitle_style, - persist_intermediate=parse_bool(persist_intermediate), - involve_mode=parse_bool(involve_mode), - run_id=run_id, + vocals_path, background_path, dubbing_strategy = await maybe_run_audio_separation( + preprocessing_dir, + raw_audio_path, + sep_model, + audio_sep, + dubbing_strategy, ) - if not upload_token: - local_source = Path(result.get("source_media_local_path", "") or "") - if local_source.exists(): - if remote_input_used: - digest = hashlib.sha1((provided_video_url or str(local_source)).encode("utf-8")).hexdigest() - cache_dir = uploads_dir / "remote" / digest - cache_dir.mkdir(parents=True, exist_ok=True) - cache_target = cache_dir / local_source.name - else: - cache_target = uploads_dir / local_source.name - cache_target.parent.mkdir(parents=True, exist_ok=True) - if local_source.resolve() != cache_target.resolve(): - await run_in_thread(shutil.copy2, local_source, cache_target) - upload_token = str(cache_target.relative_to(uploads_dir)) - languages_payload = {} - language_outputs = result.get("language_outputs") or {} - for lang, data in language_outputs.items(): - subtitles_aligned = data.get("subtitles", {}).get("aligned", {}) - intermediate_payload = {} - for key, value in (data.get("intermediate_files") or {}).items(): - intermediate_payload[key] = build_file_payload(value) - languages_payload[lang] = { - "final_video": build_file_payload(data.get("final_video_path")), - "final_audio": build_file_payload(data.get("final_audio_path")), - "speech_track": build_file_payload(data.get("speech_track")), - "subtitles": { - "aligned": { - "srt": build_file_payload(subtitles_aligned.get("srt")), - "vtt": build_file_payload(subtitles_aligned.get("vtt")), - } - }, - "intermediate_files": intermediate_payload, - "models": data.get("models", {}), - } - - def build_subtitle_section(section: Dict[str, Any]) -> Dict[str, Optional[Dict[str, str]]]: - return { - "srt": build_file_payload(section.get("srt")), - "vtt": build_file_payload(section.get("vtt")), - } - - subtitles_result = result.get("subtitles", {}) or {} - subtitles_payload = { - "original": build_subtitle_section(subtitles_result.get("original", {})), - "aligned": build_subtitle_section(subtitles_result.get("aligned", {})), - } - per_language_subtitles = {} - for lang, subs in (subtitles_result.get("per_language") or {}).items(): - per_language_subtitles[lang] = { - key: build_subtitle_section(values) for key, values in subs.items() - } - if per_language_subtitles: - subtitles_payload["per_language"] = per_language_subtitles - - payload = { - "type": "result", - "result": { - "run_id": run_id, - "workspace_id": result.get("workspace_id"), - "source_media": result.get("source_media"), - "source_video": build_file_payload(result.get("source_video")), - "final_video": build_file_payload(result.get("final_video_path")), - "final_audio": build_file_payload(result.get("final_audio_path")), - "speech_track": build_file_payload(result.get("speech_track")), - "subtitles": subtitles_payload, - "default_language": result.get("default_language"), - "available_languages": result.get("available_languages", []), - "languages": languages_payload, - "models": result.get("models", {}), - "timings": result.get("timings", {}), - "upload_token": upload_token, - "source_media_local_path": result.get("source_media_local_path"), - }, - } - await queue.put(payload) - except asyncio.CancelledError: - await queue.put({"type": "cancelled", "run_id": run_id}) - raise - except HTTPException as exc: - await queue.put({"type": "error", "status": exc.status_code, "message": exc.detail}) - except Exception as exc: # noqa: BLE001 - logger.exception("Pipeline run failed: %s", exc) - await queue.put({"type": "error", "message": str(exc)}) - finally: - PROGRESS_REPORTER.reset(token) - await queue.put({"type": "complete"}) - async def event_stream(): - task = asyncio.create_task(run_pipeline()) - ACTIVE_JOBS[run_id] = task - try: - while True: - event = await queue.get() - yield f"data: {json.dumps(event)}\n\n" - if event.get("type") == "complete": - break - finally: - ACTIVE_JOBS.pop(run_id, None) - if not task.done(): - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - except Exception: - logger.exception("Pipeline task raised after completion", exc_info=True) + transcript_audio = vocals_path if vocals_path and vocal_for_transcript else raw_audio_path - return StreamingResponse(event_stream(), media_type="text/event-stream") + with step_timer.time("asr"): + raw_asr_result, aligned_asr_result = await run_asr_step( + client, + transcript_audio, + asr_model, + source_lang, + min_speakers, + max_speakers, + perform_alignment=not involve_mode, + ) - -@app.post("/v1/dub") -async def dub( - video_url: str, - target_work: str = Query( - "dub", - description="Target work type, e.g., 'dub': for full dubbing or 'sub': for subtitles only", - ), - target_langs: Optional[List[str] | str] = Query(None), - source_lang: Optional[str] = None, - min_speakers: Optional[int] = None, - max_speakers: Optional[int] = None, - sep_model: str = Query("melband_roformer_big_beta5e.ckpt"), - asr_model: str = Query("whisperx"), - tr_model: str = Query("facebook_m2m100"), - tts_model: str = Query("chatterbox"), - audio_sep: bool = Query(True, description="Whether to perform audio source separation"), - perform_vad_trimming: bool = Query(True, description="Whether to perform VAD-based silence trimming after TTS"), - translation_strategy: str = Query( - "default", - description="Translation strategy to use: either translate directly over the short ASR aligned segments or translate the full text and then align the translated result after", - ), - dubbing_strategy: str = Query( - "default", - description="Dubbing strategy to use, either translation over (original audio ducked) or full replacement", - ), - sophisticated_dub_timing: bool = Query( - True, - description="Whether to use sophisticated timing for full replacement dubbing strategy", - ), - subtitle_style: Optional[str] = Query( - None, - description="Subtitle style preset: default, minimal, bold, netflix", - ), - persist_intermediate: bool = Query( - True, - description="Persist intermediate artifacts (disable for lower latency and disk usage)", - ), - involve_mode: bool = Query( - False, - description="Enable involve-mode workflow with manual transcription review between stages.", - ), - run_id: Optional[str] = Query( - None, - description="Optional run identifier when invoked from the job runner (required for involve mode).", - ), -): - """ - Complete dubbing pipeline orchestrator. - """ - - if involve_mode and not run_id: - raise HTTPException(400, "Involve mode requires an active run context (run_id).") - - original_source = video_url - video_url = video_url.strip() - if target_work == "sub" and subtitle_style is None: - subtitle_style = "default_mobile" - if dubbing_strategy != "translation_over": - dubbing_strategy = "full_replacement" # other values default to full_replacement - - source_lang = (source_lang or "").strip() or None - target_languages = normalize_language_codes((list(target_langs)) if target_langs else []) - - sep_model = (sep_model or "").strip() - - requested_tr_model = (tr_model or "").strip() - requested_tts_model = (tts_model or "").strip() - - asr_model = resolve_model_choice(asr_model, ASR_WORKERS, source_lang, fallback= general_cfg.get("default_models", {}).get("asr", "whisperx")) - - translation_models_by_lang: Dict[str, str] = {} - tts_models_by_lang: Dict[str, str] = {} - per_language_models: Dict[str, Dict[str, str]] = {} - for lang in target_languages: - translation_models_by_lang[lang] = resolve_model_choice( - requested_tr_model, - TR_WORKERS, - lang or source_lang, - fallback= general_cfg.get("default_models", {}).get("tr", "deep_translator"), - ) - tts_models_by_lang[lang] = resolve_model_choice( - requested_tts_model, - TTS_WORKERS, - lang, - fallback= general_cfg.get("default_models", {}).get("tts", "chatterbox"), - ) - - selected_models = { - "asr": asr_model, - "translation": translation_models_by_lang, - "tts": tts_models_by_lang, - "separation": sep_model, - } - - if translation_strategy not in TRANSLATION_STRATEGIES: - translation_strategy = TRANSLATION_STRATEGIES[0] - if dubbing_strategy not in DUBBING_STRATEGIES: - dubbing_strategy = DUBBING_STRATEGIES[0] - - workspace = WorkspaceManager.create(OUTS, persist_intermediate) - step_timer = StepTimer() - client = get_http_client() - - subtitles_dir: Optional[Path] = None - - resolved_video_path, media_digest = await prepare_media_source(video_url, workspace) - source_media_local_path: Optional[Path] = resolved_video_path - - source_has_video = await has_video_stream(resolved_video_path) # better check to avoid issues with audio-only inputs being treated as videos - - if not source_has_video: - target_work = "dub" - subtitle_style = None - dubbing_strategy = "full_replacement" - - if workspace.persist_intermediate: - preprocessing_dir = workspace.ensure_dir("preprocessing") - else: - preprocessing_dir = workspace.make_temp_dir("preprocessing") - - raw_audio_path = preprocessing_dir / "raw_audio.wav" - - cancelled = False - - try: - raw_audio_cached = False - raw_audio_cache_token: Optional[str] = None - if media_digest: - raw_audio_cache_token = raw_audio_cache_key(media_digest) - raw_audio_cached = await load_cached_raw_audio(raw_audio_cache_token, raw_audio_path) - if raw_audio_cached: - logger.info("Loaded raw audio from cache for media digest %s", media_digest) - - if not raw_audio_cached: - with step_timer.time("extract_audio"): - await extract_audio_to_workspace(str(resolved_video_path), raw_audio_path) - if raw_audio_cache_token: - await store_raw_audio_cache(raw_audio_cache_token, raw_audio_path) - - raw_audio_duration = get_audio_duration(raw_audio_path) - - vocals_path: Optional[Path] = None - background_path: Optional[Path] = None - - vocal_for_transcript = general_cfg.get("vocal_only_for_transcription", True) - logger.info("Vocal only for transcription: %s", vocal_for_transcript) - - if target_work != "sub" or vocal_for_transcript: # in subtitle-only mode, no need to separate audio for now: in future we might want to do it for better ASR performance if we succeed to implement automatic noise level detection - with step_timer.time("audio_separation"): - - vocals_path, background_path, dubbing_strategy = await maybe_run_audio_separation( - preprocessing_dir, - raw_audio_path, - sep_model, - audio_sep, - dubbing_strategy, - ) - - transcript_audio = vocals_path if vocals_path and vocal_for_transcript else raw_audio_path - - with step_timer.time("asr"): - raw_asr_result, aligned_asr_result = await run_asr_step( - client, - transcript_audio, - asr_model, - source_lang, - min_speakers, - max_speakers, - perform_alignment=not involve_mode, - ) - - original_raw_dump = raw_asr_result.model_dump() - asr_raw_path = "" + original_raw_dump = raw_asr_result.model_dump() + asr_raw_path = "" if involve_mode: languages_set: set[str] = set() @@ -2257,73 +1796,505 @@ def keep_if_persistent(path: Optional[str | Path]) -> str: if data.get("subtitles") } - if per_language_models: - selected_models["per_language"] = per_language_models + if per_language_models: + selected_models["per_language"] = per_language_models + + default_intermediate = (language_outputs_serialized.get(default_language) or {}).get("intermediate_files", {}) + + intermediate_files_payload: Dict[str, Any] = { + "asr_original": keep_if_persistent(asr_raw_path), + "asr_aligned": keep_if_persistent(asr_aligned_path), + "translation": default_intermediate.get("translation", ""), + "translation_aligned_W_origin": default_intermediate.get("translation_aligned_W_origin", ""), + "translation_aligned_W_dubbedvoice": default_intermediate.get("translation_aligned_W_dubbedvoice", ""), + "tts": default_intermediate.get("tts", ""), + "vocals": keep_if_persistent(vocals_path), + "background": keep_if_persistent(background_path), + "per_language": { + lang: data["intermediate_files"] for lang, data in language_outputs_serialized.items() + }, + } + + subtitles_payload: Dict[str, Any] = { + "original": {"srt": keep_if_persistent(srt_path_0), "vtt": keep_if_persistent(vtt_path_0)}, + "aligned": {"srt": keep_if_persistent(srt_path_1), "vtt": keep_if_persistent(vtt_path_1)}, + } + if subtitles_per_language_payload: + subtitles_payload["per_language"] = { + lang: {"aligned": subs.get("aligned", {})} + for lang, subs in subtitles_per_language_payload.items() + if subs.get("aligned") + } + + final_result: Dict[str, Any] = { + "workspace_id": workspace.workspace_id, + "final_video_path": final_video_path, + "final_audio_path": default_audio_path, + "speech_track": default_speech_track, + "source_media": original_source, + "source_video": keep_if_persistent(source_media_local_path), + "source_media_local_path": str(source_media_local_path) if source_media_local_path else "", + "default_language": default_language, + "available_languages": list(language_outputs_serialized.keys()), + "language_outputs": language_outputs_serialized, + "models": selected_models, + "subtitles": subtitles_payload, + "intermediate_files": intermediate_files_payload, + "timings": step_timer.timings, + } + + workspace.maybe_dump_json("final_result.json", final_result) + + return final_result + + except asyncio.CancelledError: + cancelled = True + raise + except HTTPException: + raise + except Exception as exc: # noqa: BLE001 + logger.exception("Pipeline failed: %s", exc) + raise HTTPException(500, f"Pipeline failed: {exc}") from exc + finally: + if not workspace.persist_intermediate: + for path in workspace.temp_dirs: + shutil.rmtree(path, ignore_errors=True) + temp_root = workspace.workspace / "_temp" + shutil.rmtree(temp_root, ignore_errors=True) + if cancelled: + try: + shutil.rmtree(workspace.workspace, ignore_errors=True) + except Exception: # noqa: BLE001 + logger.warning("Failed to remove workspace %s after cancellation", workspace.workspace, exc_info=True) + +@app.on_event("startup") +async def startup_event() -> None: + timeout = httpx.Timeout(connect=10.0, read=1200.0, write=10.0, pool=None) + app.state.http_client = httpx.AsyncClient(timeout=timeout) + OUTS.mkdir(parents=True, exist_ok=True) + SEPARATION_CACHE.mkdir(parents=True, exist_ok=True) + RAW_AUDIO_CACHE.mkdir(parents=True, exist_ok=True) + UPLOADS_DIR.mkdir(parents=True, exist_ok=True) + + +@app.on_event("shutdown") +async def shutdown_event() -> None: + client = getattr(app.state, "http_client", None) + if client: + await client.aclose() + + +@app.get(OPTIONS_ROUTE) +async def pipeline_options() -> JSONResponse: + return JSONResponse( + { + "asr_models": list_worker_models(ASR_WORKERS), + "translation_models": list_worker_models(TR_WORKERS), + "tts_models": list_worker_models(TTS_WORKERS), + "audio_separation_models": list_audio_separation_models(), + "translation_strategies": TRANSLATION_STRATEGIES, + "dubbing_strategies": DUBBING_STRATEGIES, + "subtitle_styles": sorted(STYLE_PRESETS.keys()), + } + ) + + +@app.get(FILE_ROUTE) +async def pipeline_file(path: str) -> FileResponse: + resolved = Path(path).resolve() + try: + resolved.relative_to(OUTS) + except ValueError as exc: + raise HTTPException(403, "Invalid path") from exc + + if not resolved.exists() or not resolved.is_file(): + raise HTTPException(404, "File not found") + return FileResponse(resolved) + + +@app.post(RELEASE_ROUTE) +async def pipeline_release_media(token: str = Form(...)) -> JSONResponse: + token = (token or "").strip() + if not token: + return JSONResponse({"status": "ignored"}) + try: + target = resolve_cached_media_token(token) + except HTTPException: + return JSONResponse({"status": "invalid"}, status_code=200) + if target.exists(): + cleanup_cached_media_path(target) + return JSONResponse({"status": "released"}) + return JSONResponse({"status": "missing"}) + + +@app.post(STOP_ROUTE) +async def pipeline_stop(run_id: str = Form(...)) -> JSONResponse: + run_id = (run_id or "").strip() + if not run_id: + raise HTTPException(400, "run_id is required") + task = ACTIVE_JOBS.get(run_id) + if task is None: + return JSONResponse({"status": "missing"}) + if task.done(): + ACTIVE_JOBS.pop(run_id, None) + return JSONResponse({"status": "completed"}) + task.cancel() + return JSONResponse({"status": "cancelling"}) + + +@app.post(f"{JOBS_PREFIX}/transcription_review") +async def pipeline_submit_transcription_review(review: TranscriptionReviewRequest) -> JSONResponse: + run_id = (review.run_id or "").strip() + if not run_id: + raise HTTPException(400, "run_id is required") + future = TRANSCRIPTION_REVIEW_WAITERS.get(run_id) + if future is None: + raise HTTPException(404, "No pending transcription review for this run.") + if future.done(): + raise HTTPException(409, "Transcription review already submitted for this run.") + session = TRANSCRIPTION_REVIEW_SESSIONS.get(run_id) + audio_duration = session.audio_duration if session else None + tolerance = session.tolerance if session else TRANSCRIPTION_SEGMENT_TOLERANCE + if audio_duration is not None and audio_duration <= tolerance: + audio_duration = None + + allowed_languages = set((session.languages or []) if session else []) + segments = review.transcription.segments or [] + sanitized: List[Segment] = [] + for segment in segments: + start_raw = segment.start if segment.start is not None else 0.0 + end_raw = segment.end if segment.end is not None else start_raw + try: + start = float(start_raw) + except (TypeError, ValueError): + raise HTTPException(400, f"Invalid start time for segment {segment.segment_id or '?'}") + try: + end = float(end_raw) + except (TypeError, ValueError): + raise HTTPException(400, f"Invalid end time for segment {segment.segment_id or '?'}") + if audio_duration is not None: + if start < -tolerance or start > audio_duration + tolerance: + raise HTTPException( + 400, + f"Segment start {start:.3f}s exceeds audio bounds (duration {audio_duration:.3f}s).", + ) + if end < -tolerance or end > audio_duration + tolerance: + raise HTTPException( + 400, + f"Segment end {end:.3f}s exceeds audio bounds (duration {audio_duration:.3f}s).", + ) + start = max(0.0, min(start, audio_duration + tolerance)) + end = max(0.0, min(end, audio_duration + tolerance)) + if end < start: + if end + tolerance >= start: + end = start + else: + raise HTTPException(400, f"Segment end {end:.3f}s precedes start {start:.3f}s.") + text = (segment.text or "").strip() + lang = (segment.lang or "").strip().lower() or None + if lang: + allowed_languages.add(lang) + segment.start = round(start, 3) + segment.end = round(end, 3) + segment.text = text + segment.lang = lang + segment.words = None + sanitized.append(segment) + sanitized.sort(key=lambda seg: seg.start if seg.start is not None else 0.0) + review.transcription.segments = sanitized + ensure_segment_ids(review.transcription) + TRANSCRIPTION_REVIEW_SESSIONS.pop(run_id, None) + future.set_result(review.transcription) + return JSONResponse({"status": "accepted"}) + + +@app.post(f"{JOBS_PREFIX}/alignment_review") +async def pipeline_submit_alignment_review(review: AlignmentReviewRequest) -> JSONResponse: + run_id = (review.run_id or "").strip() + if not run_id: + raise HTTPException(400, "run_id is required") + future = ALIGNMENT_REVIEW_WAITERS.get(run_id) + if future is None: + raise HTTPException(404, "No pending alignment review for this run.") + if future.done(): + raise HTTPException(409, "Alignment review already submitted for this run.") + future.set_result(review.alignment) + return JSONResponse({"status": "accepted"}) + + +@app.post(f"{JOBS_PREFIX}/tts_review") +async def pipeline_submit_tts_review(review: TTSReviewRequest) -> JSONResponse: + run_id = (review.run_id or "").strip() + if not run_id: + raise HTTPException(400, "run_id is required") + key = tts_session_key(run_id, review.language) + session = TTS_REVIEW_SESSIONS.get(key) + if session is None: + raise HTTPException(404, "No pending TTS review for this run and language.") + future = session.future + if future.done(): + raise HTTPException(409, "TTS review already submitted for this language.") + + updates = {item.segment_id: item for item in review.segments} + async with session.lock: + for segment in session.translation.values(): + if not segment.segment_id: + continue + update = updates.get(segment.segment_id) + if not update: + continue + segment.text = update.text.strip() + if update.lang: + segment.lang = update.lang + state = session.segments.get(segment.segment_id) + if state: + state.text = segment.text + state.lang = segment.lang or state.lang + for audio_segment in session.tts_response.segments: + if not audio_segment.segment_id: + continue + update = updates.get(audio_segment.segment_id) + if not update: + continue + if update.lang: + audio_segment.lang = update.lang + if not future.done(): + future.set_result(True) + emit_progress( + { + "type": "tts_review_complete", + "run_id": run_id, + "language": review.language, + } + ) + return JSONResponse({"status": "accepted"}) + - default_intermediate = (language_outputs_serialized.get(default_language) or {}).get("intermediate_files", {}) +@app.post(f"{JOBS_PREFIX}/tts_review/regenerate") +async def pipeline_regenerate_tts_segment( + run_id: str = Form(...), + segment_id: str = Form(...), + text: str = Form(...), + language: Optional[str] = Form(None), + lang: Optional[str] = Form(None), + audio_prompt_file: Optional[UploadFile] = File(None), +) -> JSONResponse: + run_id = (run_id or "").strip() + if not run_id: + raise HTTPException(400, "run_id is required") + if not segment_id: + raise HTTPException(400, "segment_id is required") + key = tts_session_key(run_id, language) + session = TTS_REVIEW_SESSIONS.get(key) + if session is None: + raise HTTPException(404, "No pending TTS review for this run and language.") + if session.future.done(): + raise HTTPException(409, "TTS review already submitted for this language.") - intermediate_files_payload: Dict[str, Any] = { - "asr_original": keep_if_persistent(asr_raw_path), - "asr_aligned": keep_if_persistent(asr_aligned_path), - "translation": default_intermediate.get("translation", ""), - "translation_aligned_W_origin": default_intermediate.get("translation_aligned_W_origin", ""), - "translation_aligned_W_dubbedvoice": default_intermediate.get("translation_aligned_W_dubbedvoice", ""), - "tts": default_intermediate.get("tts", ""), - "vocals": keep_if_persistent(vocals_path), - "background": keep_if_persistent(background_path), - "per_language": { - lang: data["intermediate_files"] for lang, data in language_outputs_serialized.items() - }, - } + audio_prompt_url = None + if audio_prompt_file and audio_prompt_file.filename: + try: + audio_prompts_dir = session.workspace / "audio_prompts" + audio_prompts_dir.mkdir(parents=True, exist_ok=True) + safe_name = f"{segment_id}_{safe_filename(audio_prompt_file.filename)}" + audio_prompt_path = audio_prompts_dir / safe_name + with audio_prompt_path.open("wb") as dest: + content = await audio_prompt_file.read() + dest.write(content) + audio_prompt_url = str(audio_prompt_path) + logger.info(f"Custom audio prompt uploaded for segment {segment_id}: {audio_prompt_url}") + except Exception as exc: + logger.warning(f"Failed to process audio prompt upload: {exc}") - subtitles_payload: Dict[str, Any] = { - "original": {"srt": keep_if_persistent(srt_path_0), "vtt": keep_if_persistent(vtt_path_0)}, - "aligned": {"srt": keep_if_persistent(srt_path_1), "vtt": keep_if_persistent(vtt_path_1)}, + state = await regenerate_tts_segment_audio(session, segment_id, text, lang, audio_prompt_url) + segment_payload = serialize_tts_review_segment(state) + emit_progress( + { + "type": "tts_review_regenerated", + "run_id": run_id, + "language": language, + "segment": segment_payload, } - if subtitles_per_language_payload: - subtitles_payload["per_language"] = { - lang: {"aligned": subs.get("aligned", {})} - for lang, subs in subtitles_per_language_payload.items() - if subs.get("aligned") - } + ) + return JSONResponse({"status": "ok", "segment": segment_payload}) - final_result: Dict[str, Any] = { - "workspace_id": workspace.workspace_id, - "final_video_path": final_video_path, - "final_audio_path": default_audio_path, - "speech_track": default_speech_track, - "source_media": original_source, - "source_video": keep_if_persistent(source_media_local_path), - "source_media_local_path": str(source_media_local_path) if source_media_local_path else "", - "default_language": default_language, - "available_languages": list(language_outputs_serialized.keys()), - "language_outputs": language_outputs_serialized, - "models": selected_models, - "subtitles": subtitles_payload, - "intermediate_files": intermediate_files_payload, - "timings": step_timer.timings, - } - workspace.maybe_dump_json("final_result.json", final_result) +@app.post(RUN_ROUTE) +async def pipeline_run( + file: UploadFile | None = File(None), + video_url: Optional[str] = Form(None), + target_work: str = Form("dub"), + target_langs: Optional[List[str]] = Form(None), + source_lang: Optional[str] = Form(None), + min_speakers: Optional[int] = Form(None), + max_speakers: Optional[int] = Form(None), + reuse_media_token: Optional[str] = Form(None), + asr_model: Optional[str] = Form("auto"), + tr_model: Optional[str] = Form("auto"), + tts_model: Optional[str] = Form("auto"), + sep_model: Optional[str] = Form("auto"), + audio_sep: str = Form("true"), + perform_vad_trimming: str = Form("true"), + translation_strategy: str = Form("default"), + dubbing_strategy: str = Form("default"), + sophisticated_dub_timing: str = Form("true"), + subtitle_style: Optional[str] = Form(None), + persist_intermediate: str = Form("false"), + involve_mode: str = Form("false"), + ) -> StreamingResponse: + uploads_dir = UPLOADS_DIR + uploads_dir.mkdir(parents=True, exist_ok=True) - return final_result + provided_video_url = (video_url or "").strip() or None + source_media: Optional[str] = None + upload_path: Optional[Path] = None + upload_token: Optional[str] = None + remote_input_used = False + subtitle_style = (subtitle_style or "").strip() or None + if sep_model == "auto": + sep_model = general_cfg.get("default_models", {}).get("sep", "melband_roformer_big_beta5e.ckpt") - except asyncio.CancelledError: - cancelled = True - raise - except HTTPException: - raise - except Exception as exc: # noqa: BLE001 - logger.exception("Pipeline failed: %s", exc) - raise HTTPException(500, f"Pipeline failed: {exc}") from exc - finally: - if not workspace.persist_intermediate: - for path in workspace.temp_dirs: - shutil.rmtree(path, ignore_errors=True) - temp_root = workspace.workspace / "_temp" - shutil.rmtree(temp_root, ignore_errors=True) - if cancelled: + if file and file.filename: + upload_path = await persist_uploaded_file(file, uploads_dir) + source_media = str(upload_path) + upload_token = str(upload_path.relative_to(uploads_dir)) + elif provided_video_url: + source_media = provided_video_url + remote_input_used = True + elif reuse_media_token: + cached_path = resolve_cached_media_token(reuse_media_token) + if not cached_path.exists(): + raise HTTPException(400, "Cached media not found; please re-upload your file.") + source_media = str(cached_path) + upload_token = str(Path(reuse_media_token)) + else: + raise HTTPException(400, "Provide either a media file or a video link.") + + run_id = str(uuid.uuid4()) + queue: asyncio.Queue[Dict[str, Any]] = asyncio.Queue() + + def report(event: Dict[str, Any]) -> None: try: - shutil.rmtree(workspace.workspace, ignore_errors=True) - except Exception: # noqa: BLE001 - logger.warning("Failed to remove workspace %s after cancellation", workspace.workspace, exc_info=True) + queue.put_nowait(event) + except asyncio.QueueFull: + logger.debug("Progress queue full; dropping event: %s", event) + + async def run_pipeline() -> None: + nonlocal upload_token + await queue.put({"type": "run_id", "run_id": run_id}) + token = PROGRESS_REPORTER.set(report) + try: + result = await dub( + video_url=str(source_media), + target_work=target_work, + target_langs=target_langs, + source_lang=source_lang, + min_speakers=min_speakers, + max_speakers=max_speakers, + sep_model=sep_model, + asr_model=asr_model, + tr_model=tr_model, + tts_model=tts_model, + audio_sep=parse_bool(audio_sep), + perform_vad_trimming=parse_bool(perform_vad_trimming), + translation_strategy=translation_strategy, + dubbing_strategy=dubbing_strategy, + sophisticated_dub_timing=parse_bool(sophisticated_dub_timing), + subtitle_style=subtitle_style, + persist_intermediate=parse_bool(persist_intermediate), + involve_mode=parse_bool(involve_mode), + run_id=run_id, + ) + if not upload_token: + local_source = Path(result.get("source_media_local_path", "") or "") + if local_source.exists(): + if remote_input_used: + digest = hashlib.sha1((provided_video_url or str(local_source)).encode("utf-8")).hexdigest() + cache_dir = uploads_dir / "remote" / digest + cache_dir.mkdir(parents=True, exist_ok=True) + cache_target = cache_dir / local_source.name + else: + cache_target = uploads_dir / local_source.name + cache_target.parent.mkdir(parents=True, exist_ok=True) + if local_source.resolve() != cache_target.resolve(): + await run_in_thread(shutil.copy2, local_source, cache_target) + upload_token = str(cache_target.relative_to(uploads_dir)) + languages_payload = {} + language_outputs = result.get("language_outputs") or {} + for lang, data in language_outputs.items(): + subtitles_aligned = data.get("subtitles", {}).get("aligned", {}) + intermediate_payload = {} + for key, value in (data.get("intermediate_files") or {}).items(): + intermediate_payload[key] = build_file_payload(value) + languages_payload[lang] = { + "final_video": build_file_payload(data.get("final_video_path")), + "final_audio": build_file_payload(data.get("final_audio_path")), + "speech_track": build_file_payload(data.get("speech_track")), + "subtitles": { + "aligned": { + "srt": build_file_payload(subtitles_aligned.get("srt")), + "vtt": build_file_payload(subtitles_aligned.get("vtt")), + } + }, + "intermediate_files": intermediate_payload, + } + event: Dict[str, Any] = { + "type": "complete", + "run_id": run_id, + "workspace_id": result.get("workspace_id"), + "final_video": build_file_payload(result.get("final_video_path")), + "final_audio": build_file_payload(result.get("final_audio_path")), + "speech_track": build_file_payload(result.get("speech_track")), + "source_video": build_file_payload(result.get("source_video")), + "default_language": result.get("default_language"), + "available_languages": result.get("available_languages", []), + "language_outputs": languages_payload, + "models": result.get("models", {}), + "subtitles": { + "original": { + "srt": build_file_payload(result.get("subtitles", {}).get("original", {}).get("srt")), + "vtt": build_file_payload(result.get("subtitles", {}).get("original", {}).get("vtt")), + }, + "aligned": { + "srt": build_file_payload(result.get("subtitles", {}).get("aligned", {}).get("srt")), + "vtt": build_file_payload(result.get("subtitles", {}).get("aligned", {}).get("vtt")), + }, + }, + "intermediate_files": { + key: build_file_payload(value) + for key, value in (result.get("intermediate_files") or {}).items() + if isinstance(value, (str, Path)) + }, + "timings": result.get("timings", {}), + "cached_media_token": upload_token, + } + await queue.put(event) + except asyncio.CancelledError: + await queue.put({"type": "cancelled", "run_id": run_id}) + except HTTPException as exc: + await queue.put({"type": "error", "run_id": run_id, "detail": exc.detail}) + except Exception as exc: # noqa: BLE001 + logger.exception("Pipeline run_pipeline failed unexpectedly") + await queue.put({"type": "error", "run_id": run_id, "detail": str(exc)}) + finally: + ACTIVE_JOBS.pop(run_id, None) + PROGRESS_REPORTER.reset(token) + + task = asyncio.create_task(run_pipeline()) + ACTIVE_JOBS[run_id] = task + + async def event_stream(): + sentinel = object() + while True: + try: + event = await asyncio.wait_for(queue.get(), timeout=30.0) + except asyncio.TimeoutError: + yield "event: ping\ndata: {}\n\n" + continue + if event is sentinel: + break + yield f"data: {json.dumps(event)}\n\n" + if event.get("type") in ("complete", "error", "cancelled"): + break + + return StreamingResponse(event_stream(), media_type="text/event-stream") diff --git a/deploy/k8s/asr.yaml b/deploy/k8s/asr.yaml new file mode 100644 index 0000000..ddc0b1e --- /dev/null +++ b/deploy/k8s/asr.yaml @@ -0,0 +1,74 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: asr + namespace: bluez-dubbing +spec: + replicas: 1 + selector: + matchLabels: + app: asr + template: + metadata: + labels: + app: asr + spec: + automountServiceAccountToken: false + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + containers: + - name: asr + image: ghcr.io/codemowers/bluez-dubbing/asr:latest + imagePullPolicy: Always + ports: + - containerPort: 8001 + env: + - name: PYTHONPATH + value: /app/apps/backend:/app + - name: HF_TOKEN + valueFrom: + secretKeyRef: + name: bluez-secrets + key: HF_TOKEN + volumeMounts: + - name: model-cache + mountPath: /root/.cache + resources: + requests: + cpu: "500m" + memory: "2Gi" + limits: + memory: "8Gi" + livenessProbe: + httpGet: + path: /docs + port: 8001 + initialDelaySeconds: 60 + periodSeconds: 30 + failureThreshold: 5 + readinessProbe: + httpGet: + path: /docs + port: 8001 + initialDelaySeconds: 30 + periodSeconds: 10 + volumes: + - name: model-cache + persistentVolumeClaim: + claimName: model-cache-pvc +--- +apiVersion: v1 +kind: Service +metadata: + name: asr + namespace: bluez-dubbing +spec: + selector: + app: asr + ports: + - port: 8001 + targetPort: 8001 + clusterIP: None # headless; orchestrator talks directly to pod diff --git a/deploy/k8s/frontend.yaml b/deploy/k8s/frontend.yaml new file mode 100644 index 0000000..8a2d09a --- /dev/null +++ b/deploy/k8s/frontend.yaml @@ -0,0 +1,83 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: frontend + namespace: bluez-dubbing +spec: + replicas: 1 + selector: + matchLabels: + app: frontend + template: + metadata: + labels: + app: frontend + spec: + automountServiceAccountToken: false + securityContext: + runAsNonRoot: true + runAsUser: 101 # nginx user + runAsGroup: 101 + containers: + - name: frontend + image: ghcr.io/codemowers/bluez-dubbing/frontend:latest + imagePullPolicy: Always + ports: + - containerPort: 8080 + volumeMounts: + - name: outs + mountPath: /app/apps/backend/outs + readOnly: true + resources: + requests: + cpu: "50m" + memory: "64Mi" + limits: + memory: "128Mi" + livenessProbe: + httpGet: + path: / + port: 8080 + initialDelaySeconds: 5 + periodSeconds: 15 + readinessProbe: + httpGet: + path: / + port: 8080 + initialDelaySeconds: 3 + periodSeconds: 5 + volumes: + - name: outs + persistentVolumeClaim: + claimName: outs-pvc +--- +apiVersion: v1 +kind: Service +metadata: + name: frontend + namespace: bluez-dubbing +spec: + selector: + app: frontend + ports: + - port: 80 + targetPort: 8080 +--- +# Uncomment and configure for your ingress controller. +# apiVersion: networking.k8s.io/v1 +# kind: Ingress +# metadata: +# name: frontend +# namespace: bluez-dubbing +# spec: +# rules: +# - host: dubbing.example.com +# http: +# paths: +# - path: / +# pathType: Prefix +# backend: +# service: +# name: frontend +# port: +# number: 80 diff --git a/deploy/k8s/kustomization.yaml b/deploy/k8s/kustomization.yaml new file mode 100644 index 0000000..6c5cf49 --- /dev/null +++ b/deploy/k8s/kustomization.yaml @@ -0,0 +1,12 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +namespace: bluez-dubbing +resources: + - namespace.yaml + - secret.yaml + - pvc.yaml + - asr.yaml + - translation.yaml + - tts.yaml + - orchestrator.yaml + - frontend.yaml diff --git a/deploy/k8s/namespace.yaml b/deploy/k8s/namespace.yaml new file mode 100644 index 0000000..77e0650 --- /dev/null +++ b/deploy/k8s/namespace.yaml @@ -0,0 +1,4 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: bluez-dubbing diff --git a/deploy/k8s/orchestrator.yaml b/deploy/k8s/orchestrator.yaml new file mode 100644 index 0000000..bfc0b46 --- /dev/null +++ b/deploy/k8s/orchestrator.yaml @@ -0,0 +1,90 @@ +# The orchestrator is based on beveradb/audio-separator which ships +# audio-separator, ffmpeg, rubberband, torch, and onnxruntime pre-installed. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: orchestrator + namespace: bluez-dubbing +spec: + replicas: 1 + selector: + matchLabels: + app: orchestrator + template: + metadata: + labels: + app: orchestrator + spec: + automountServiceAccountToken: false + # audio-separator upstream image runs as root by default; + # override to non-root once confirmed working. + containers: + - name: orchestrator + image: ghcr.io/codemowers/bluez-dubbing/orchestrator:latest + imagePullPolicy: Always + ports: + - containerPort: 8000 + env: + - name: PYTHONPATH + value: /app/apps/backend:/app + - name: ASR_URL + value: http://asr:8001/v1/transcribe + - name: TR_URL + value: http://translation:8002/v1/translate + - name: TTS_URL + value: http://tts:8003/v1/synthesize + - name: ORCHESTRATOR_ALLOWED_ORIGINS + value: "*" + - name: HF_TOKEN + valueFrom: + secretKeyRef: + name: bluez-secrets + key: HF_TOKEN + volumeMounts: + - name: outs + mountPath: /app/apps/backend/outs + - name: uploads + mountPath: /app/apps/backend/uploads + - name: model-cache + mountPath: /root/.cache + resources: + requests: + cpu: "1000m" + memory: "4Gi" + limits: + memory: "16Gi" + livenessProbe: + httpGet: + path: /api/options + port: 8000 + initialDelaySeconds: 120 + periodSeconds: 30 + failureThreshold: 5 + readinessProbe: + httpGet: + path: /api/options + port: 8000 + initialDelaySeconds: 60 + periodSeconds: 10 + volumes: + - name: outs + persistentVolumeClaim: + claimName: outs-pvc + - name: uploads + persistentVolumeClaim: + claimName: uploads-pvc + - name: model-cache + persistentVolumeClaim: + claimName: model-cache-pvc +--- +apiVersion: v1 +kind: Service +metadata: + name: orchestrator + namespace: bluez-dubbing +spec: + selector: + app: orchestrator + ports: + - port: 8000 + targetPort: 8000 diff --git a/deploy/k8s/pvc.yaml b/deploy/k8s/pvc.yaml new file mode 100644 index 0000000..44760e7 --- /dev/null +++ b/deploy/k8s/pvc.yaml @@ -0,0 +1,38 @@ +# Shared storage for job outputs and uploaded media. +# Both the orchestrator and frontend (for /outs serving) mount outs-pvc. +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: outs-pvc + namespace: bluez-dubbing +spec: + accessModes: + - ReadWriteMany + resources: + requests: + storage: 50Gi +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: uploads-pvc + namespace: bluez-dubbing +spec: + accessModes: + - ReadWriteMany + resources: + requests: + storage: 20Gi +--- +# Model weights cache — ReadWriteOnce is fine; only one pod writes at a time. +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: model-cache-pvc + namespace: bluez-dubbing +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 100Gi diff --git a/deploy/k8s/secret.yaml b/deploy/k8s/secret.yaml new file mode 100644 index 0000000..e7dcfcc --- /dev/null +++ b/deploy/k8s/secret.yaml @@ -0,0 +1,13 @@ +# Populate before applying: +# kubectl create secret generic bluez-secrets -n bluez-dubbing \ +# --from-literal=HF_TOKEN=hf_... +# +# This file is a reminder of the shape; do not commit real tokens. +apiVersion: v1 +kind: Secret +metadata: + name: bluez-secrets + namespace: bluez-dubbing +type: Opaque +stringData: + HF_TOKEN: "" diff --git a/deploy/k8s/translation.yaml b/deploy/k8s/translation.yaml new file mode 100644 index 0000000..0b4f50e --- /dev/null +++ b/deploy/k8s/translation.yaml @@ -0,0 +1,73 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: translation + namespace: bluez-dubbing +spec: + replicas: 1 + selector: + matchLabels: + app: translation + template: + metadata: + labels: + app: translation + spec: + automountServiceAccountToken: false + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + containers: + - name: translation + image: ghcr.io/codemowers/bluez-dubbing/translation:latest + imagePullPolicy: Always + ports: + - containerPort: 8002 + env: + - name: PYTHONPATH + value: /app/apps/backend:/app + - name: HF_TOKEN + valueFrom: + secretKeyRef: + name: bluez-secrets + key: HF_TOKEN + volumeMounts: + - name: model-cache + mountPath: /root/.cache + resources: + requests: + cpu: "500m" + memory: "2Gi" + limits: + memory: "8Gi" + livenessProbe: + httpGet: + path: /docs + port: 8002 + initialDelaySeconds: 60 + periodSeconds: 30 + failureThreshold: 5 + readinessProbe: + httpGet: + path: /docs + port: 8002 + initialDelaySeconds: 30 + periodSeconds: 10 + volumes: + - name: model-cache + persistentVolumeClaim: + claimName: model-cache-pvc +--- +apiVersion: v1 +kind: Service +metadata: + name: translation + namespace: bluez-dubbing +spec: + selector: + app: translation + ports: + - port: 8002 + targetPort: 8002 diff --git a/deploy/k8s/tts.yaml b/deploy/k8s/tts.yaml new file mode 100644 index 0000000..35dc73b --- /dev/null +++ b/deploy/k8s/tts.yaml @@ -0,0 +1,73 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: tts + namespace: bluez-dubbing +spec: + replicas: 1 + selector: + matchLabels: + app: tts + template: + metadata: + labels: + app: tts + spec: + automountServiceAccountToken: false + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + containers: + - name: tts + image: ghcr.io/codemowers/bluez-dubbing/tts:latest + imagePullPolicy: Always + ports: + - containerPort: 8003 + env: + - name: PYTHONPATH + value: /app/apps/backend:/app + - name: HF_TOKEN + valueFrom: + secretKeyRef: + name: bluez-secrets + key: HF_TOKEN + volumeMounts: + - name: model-cache + mountPath: /root/.cache + resources: + requests: + cpu: "500m" + memory: "2Gi" + limits: + memory: "8Gi" + livenessProbe: + httpGet: + path: /docs + port: 8003 + initialDelaySeconds: 60 + periodSeconds: 30 + failureThreshold: 5 + readinessProbe: + httpGet: + path: /docs + port: 8003 + initialDelaySeconds: 30 + periodSeconds: 10 + volumes: + - name: model-cache + persistentVolumeClaim: + claimName: model-cache-pvc +--- +apiVersion: v1 +kind: Service +metadata: + name: tts + namespace: bluez-dubbing +spec: + selector: + app: tts + ports: + - port: 8003 + targetPort: 8003 diff --git a/docker-compose.override.yml b/docker-compose.override.yml new file mode 100644 index 0000000..2e5134b --- /dev/null +++ b/docker-compose.override.yml @@ -0,0 +1,43 @@ +# Development overrides — bind-mount source and enable hot reload. +# Applied automatically by `docker compose up`. +# Skip with: docker compose -f docker-compose.yml up + +services: + + asr: + volumes: + - .:/app + - model-cache:/root/.cache + command: > + uv run --project apps/backend/services/asr + uvicorn app.main:app --host 0.0.0.0 --port 8001 --reload + + translation: + volumes: + - .:/app + - model-cache:/root/.cache + command: > + uv run --project apps/backend/services/translation + uvicorn app.main:app --host 0.0.0.0 --port 8002 --reload + + tts: + volumes: + - .:/app + - model-cache:/root/.cache + command: > + uv run --project apps/backend/services/tts + uvicorn app.main:app --host 0.0.0.0 --port 8003 --reload + + orchestrator: + volumes: + - .:/app + - model-cache:/root/.cache + - outs:/app/apps/backend/outs + - uploads:/app/apps/backend/uploads + command: > + uv run --project apps/backend/services/orchestrator + uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload + + frontend: + volumes: + - ./apps/frontend:/usr/share/nginx/html:ro diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..63e7004 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,131 @@ +# Production-style local stack. +# All backend services are on an isolated internal network; +# only the frontend (port 5173) and orchestrator (port 8000) reach the host. +# +# Prerequisites: +# cp .env.example .env # then fill in HF_TOKEN +# docker compose up --build + +services: + + asr: + build: + context: . + dockerfile: Dockerfile + target: final-asr + expose: + - "8001" + env_file: .env + environment: + PYTHONPATH: /app/apps/backend:/app + volumes: + - model-cache:/root/.cache + networks: + - internal + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8001/docs')"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 60s + + translation: + build: + context: . + dockerfile: Dockerfile + target: final-translation + expose: + - "8002" + env_file: .env + environment: + PYTHONPATH: /app/apps/backend:/app + volumes: + - model-cache:/root/.cache + networks: + - internal + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8002/docs')"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 60s + + tts: + build: + context: . + dockerfile: Dockerfile + target: final-tts + expose: + - "8003" + env_file: .env + environment: + PYTHONPATH: /app/apps/backend:/app + volumes: + - model-cache:/root/.cache + networks: + - internal + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8003/docs')"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 60s + + orchestrator: + build: + context: . + dockerfile: Dockerfile + target: final-orchestrator + ports: + - "8000:8000" + env_file: .env + environment: + PYTHONPATH: /app/apps/backend:/app + ASR_URL: http://asr:8001/v1/transcribe + TR_URL: http://translation:8002/v1/translate + TTS_URL: http://tts:8003/v1/synthesize + ORCHESTRATOR_ALLOWED_ORIGINS: "*" + volumes: + - model-cache:/root/.cache + - outs:/app/apps/backend/outs + - uploads:/app/apps/backend/uploads + depends_on: + asr: + condition: service_healthy + translation: + condition: service_healthy + tts: + condition: service_healthy + networks: + - internal + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/api/options')"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 120s + + frontend: + build: + context: . + dockerfile: Dockerfile.frontend + ports: + - "5173:80" + depends_on: + - orchestrator + networks: + - internal + restart: unless-stopped + +volumes: + model-cache: + outs: + uploads: + +networks: + internal: + driver: bridge diff --git a/docker/nginx.conf b/docker/nginx.conf new file mode 100644 index 0000000..4fadb99 --- /dev/null +++ b/docker/nginx.conf @@ -0,0 +1,23 @@ +server { + # Run on 8080 so the container can be non-root (ports < 1024 require root). + listen 8080; + server_name _; + + root /usr/share/nginx/html; + index index.html; + + # Proxy /api and /outs to the orchestrator + location ~ ^/(api|outs)/ { + proxy_pass http://orchestrator:8000; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_read_timeout 1800s; + proxy_send_timeout 1800s; + } + + location / { + try_files $uri $uri/ /index.html; + } +}