oimwiodev/youtube-auto-dub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Show job progress steps in web UI

Browse Source
This commit is contained in:
oimwiodev
2026-05-24 16:25:39 +01:00
parent 75522ede50
commit f1e72f27e2
3 changed files with 111 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@@ -68,7 +68,7 @@ Basic example:
### Gradio Web UI
Gradio provides a local browser UI for starting dub jobs, watching logs, and downloading finished videos:
Gradio provides a local browser UI for starting dub jobs, watching progress, and downloading finished videos:
```powershell
.venv\Scripts\python.exe web_app.py
@@ -80,6 +80,8 @@ The OpenAI-compatible translation endpoint, API key, and model can be changed in
You can also upload a local `.mp4` instead of entering a YouTube URL. Uploaded videos are staged under `.cache/uploads` and processed with the same transcription, translation, dubbing, and render pipeline. Restricted YouTube videos can use the **Upload Cookies File** control instead of typing a local cookies path.
The web UI automatically refreshes job status, progress, steps, and output choices every few seconds while it is open. The manual **Refresh** button is still available.
### Docker
Build and run the Gradio UI in a container:

26
tests/test_web_app.py
View File

@@ -6,6 +6,8 @@ import sys
import web_app
from web_app import (
DubJob,
_job_progress,
_stage_uploaded_cookies,
build_pipeline_command,
create_app,
@@ -123,3 +125,27 @@ def test_stage_uploaded_cookies_rejects_unsupported_extension(tmp_path):
assert "Expected one of" in str(exc)
else:
raise AssertionError("Expected ValueError for unsupported cookie upload")
def test_job_progress_tracks_pipeline_steps(tmp_path):
log_path = tmp_path / "job.log"
log_path.write_text("STEP 1: PREPARING CONTENT\nSTEP 2: SPEECH TRANSCRIPTION\n", encoding="utf-8")
job = DubJob(id="demo", command=[], log_path=log_path, status="running")
progress, steps_html = _job_progress(job)
assert progress == 25
assert "[done]" in steps_html
assert "[active]" in steps_html
assert "Speech transcription" in steps_html
def test_job_progress_marks_succeeded_complete(tmp_path):
log_path = tmp_path / "job.log"
log_path.write_text("", encoding="utf-8")
job = DubJob(id="demo", command=[], log_path=log_path, status="succeeded")
progress, steps_html = _job_progress(job)
assert progress == 100
assert "[todo]" not in steps_html

99
web_app.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
import html
import json
from pathlib import Path
import os
@@ -30,6 +31,16 @@ BASE_DIR = Path(__file__).resolve().parent
LOG_DIR = BASE_DIR / "logs" / "gradio"
SETTINGS_FILE = BASE_DIR / ".cache" / "web_settings.json"
UPLOAD_DIR = BASE_DIR / ".cache" / "uploads"
PIPELINE_STEPS = [
("STEP 1", "Preparing content"),
("STEP 2", "Speech transcription"),
("STEP 3", "Intelligent chunking"),
("STEP 4", "Translation"),
("STEP 5", "Dub audio synthesis"),
("STEP 6", "Subtitle generation"),
("STEP 7", "Audio bed preparation"),
("STEP 8", "Final video rendering"),
]
@dataclass
@@ -237,6 +248,60 @@ def _read_log_tail(log_path: Path, max_chars: int = 20000) -> str:
return text[-max_chars:]
def _job_progress(job: DubJob | None) -> tuple[int, str]:
"""Return a coarse progress percentage and HTML step summary."""
if job is None:
return 0, _render_steps_html(0, "queued")
log_text = _read_log_tail(job.log_path)
current_step = 0
for index, (marker, _) in enumerate(PIPELINE_STEPS, start=1):
if marker in log_text:
current_step = index
if job.status == "succeeded":
return 100, _render_steps_html(len(PIPELINE_STEPS), job.status)
progress = int((current_step / len(PIPELINE_STEPS)) * 100)
if job.status == "running" and progress == 0:
progress = 3
return progress, _render_steps_html(current_step, job.status)
def _render_steps_html(current_step: int, status: str) -> str:
rows = []
failed = status == "failed"
for index, (_, label) in enumerate(PIPELINE_STEPS, start=1):
if failed and index == max(current_step, 1):
state = "failed"
elif index < current_step or status == "succeeded":
state = "done"
elif index == current_step and status in {"queued", "running"}:
state = "active"
else:
state = "todo"
rows.append(
"<li>"
f"<strong>[{html.escape(state)}]</strong> "
f"{index}. {html.escape(label)}"
"</li>"
)
return "<ul>" + "".join(rows) + "</ul>"
def _render_progress_html(progress: int) -> str:
bounded_progress = max(0, min(100, int(progress)))
return (
"<div>"
"<label><strong>Progress</strong></label>"
f"<progress value='{bounded_progress}' max='100' style='width: 100%; height: 24px;'></progress>"
f"<div>{bounded_progress}%</div>"
"</div>"
)
def _run_job(job: DubJob) -> None:
with JOBS_LOCK:
job.status = "running"
@@ -290,7 +355,7 @@ def _start_job(
lmstudio_api_key: str,
lmstudio_model: str,
gpu: bool,
) -> tuple[str, str, str, gr.Dropdown]:
) -> tuple[str, str, str, str, gr.Dropdown]:
saved_settings = load_translation_settings()
base_url = (lmstudio_base_url or "").strip() or saved_settings["base_url"]
api_key = (lmstudio_api_key or "").strip() or saved_settings["api_key"]
@@ -299,12 +364,12 @@ def _start_job(
input_file = _stage_uploaded_mp4(uploaded_mp4)
except (OSError, ValueError) as exc:
message = str(exc) or "Invalid uploaded MP4."
return "", message, message, gr.update(choices=_output_choices())
return "", message, _render_progress_html(0), _render_steps_html(0, "failed"), gr.update(choices=_output_choices())
try:
cookies = _stage_uploaded_cookies(cookies_upload)
except (OSError, ValueError) as exc:
message = str(exc) or "Invalid uploaded cookies file."
return "", message, message, gr.update(choices=_output_choices())
return "", message, _render_progress_html(0), _render_steps_html(0, "failed"), gr.update(choices=_output_choices())
form = {
"url": url,
@@ -324,7 +389,7 @@ def _start_job(
command = build_pipeline_command(form)
except (SystemExit, ValueError) as exc:
message = str(exc) or "Invalid job options."
return "", message, message, gr.update(choices=_output_choices())
return "", message, _render_progress_html(0), _render_steps_html(0, "failed"), gr.update(choices=_output_choices())
LOG_DIR.mkdir(parents=True, exist_ok=True)
job_id = uuid.uuid4().hex[:12]
@@ -344,17 +409,19 @@ def _start_job(
thread = threading.Thread(target=_run_job, args=(job,), daemon=True)
thread.start()
return job.id, _format_job_status(job), _read_log_tail(job.log_path), gr.update(choices=_output_choices())
progress_value, steps_html = _job_progress(job)
return job.id, _format_job_status(job), _render_progress_html(progress_value), steps_html, gr.update(choices=_output_choices())
def _refresh_job(job_id: str) -> tuple[str, str, gr.Dropdown]:
def _refresh_job(job_id: str) -> tuple[str, str, str, gr.Dropdown]:
with JOBS_LOCK:
job = JOBS.get(job_id)
if job is None:
return "Ready", "No job selected.", gr.update(choices=_output_choices())
return "Ready", _render_progress_html(0), _render_steps_html(0, "queued"), gr.update(choices=_output_choices())
return _format_job_status(job), _read_log_tail(job.log_path), gr.update(choices=_output_choices())
progress_value, steps_html = _job_progress(job)
return _format_job_status(job), _render_progress_html(progress_value), steps_html, gr.update(choices=_output_choices())
def _select_output(filename: str | None) -> str | None:
@@ -374,10 +441,11 @@ def create_app() -> gr.Blocks:
gr.Markdown(
"""
# YouTube Auto Dub
Start local dubbing jobs, watch the pipeline log, and collect finished videos.
Start local dubbing jobs, watch progress, and collect finished videos.
"""
)
job_id = gr.State("")
log_timer = gr.Timer(value=2.0, active=True)
with gr.Row():
with gr.Column(scale=5):
@@ -440,12 +508,8 @@ def create_app() -> gr.Blocks:
with gr.Column(scale=7):
status = gr.Textbox(label="Job Status", value="Ready", lines=5, interactive=False)
log = gr.Textbox(
label="Run Log",
value="No jobs yet.",
lines=20,
interactive=False,
)
progress = gr.HTML(value=_render_progress_html(0))
steps = gr.HTML(label="Steps", value=_render_steps_html(0, "queued"))
refresh = gr.Button("Refresh")
with gr.Row():
@@ -473,9 +537,10 @@ def create_app() -> gr.Blocks:
start.click(
_start_job,
inputs=inputs,
outputs=[job_id, status, log, output_choice],
outputs=[job_id, status, progress, steps, output_choice],
)
refresh.click(_refresh_job, inputs=[job_id], outputs=[status, log, output_choice])
refresh.click(_refresh_job, inputs=[job_id], outputs=[status, progress, steps, output_choice])
log_timer.tick(_refresh_job, inputs=[job_id], outputs=[status, progress, steps, output_choice])
output_choice.change(_select_output, inputs=[output_choice], outputs=[output_file])
return demo