debpalashandClaude Opus 4.8
027c08f1ff
Two P1s, both correct, and both the same shape as the bug being fixed:
- The preflight OSError I added ("Can't save the download: …") matched neither
an errno nor a download marker, so classify() returned "" and the user got
NO hint — the exact dead end this PR exists to remove. Reworded to carry
both signals; a test now asserts the class and that the hint names the data
directory.
- classify() covered ENOENT but _with_target_facts' own signature list did
not, so a job folder that vanished after preflight produced a
disk-classified error that never named the folder.
That second one is a drift class, not a one-off: two lists answering "is this
a disk problem?" will diverge again. They now share
failure.is_os_write_refusal(), with a test asserting both consumers agree
across all four errnos.
CHANGELOG entries shortened with refs last (CodeRabbit).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1153 lines
51 KiB
Python
1153 lines
51 KiB
Python
"""
|
|
Dub-pipeline service — Phase 2.4 (ROADMAP.md).
|
|
|
|
Extracts the non-HTTP business logic out of the 889-line `dub_core.py` router
|
|
so the router can stay thin (HTTP concerns only) and this module can be
|
|
imported cleanly by other routers (dub_generate, dub_translate, dub_export)
|
|
and future callers (the Tools page, the headless CLI, tests).
|
|
|
|
What's here
|
|
-----------
|
|
* **Job state** — `_dub_jobs` in-memory dict + `get_job` / `save_job` that
|
|
hydrate from `dub_history.job_data` on cache miss.
|
|
* **Content-hash cache lookup** — `compute_file_hash`, `find_cached_job`.
|
|
* **Safe path resolution** — `safe_job_dir`.
|
|
* **Process lifecycle** — ffmpeg/demucs subprocess tracking + `kill_job_procs`
|
|
so `POST /dub/abort/{id}` can tear down in-flight work (implemented in
|
|
`services.proc_registry`, re-exported here for compatibility).
|
|
* **SSE helpers** — `sse_event`, `prep_event`.
|
|
|
|
What stays in the router
|
|
------------------------
|
|
The route decorators + request-body validation + response shaping. The big
|
|
ingest/transcribe generators still live there for now — they're tightly
|
|
coupled to FastAPI's `StreamingResponse`/async-generator contract, and
|
|
moving them is a follow-up.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from typing import AsyncIterator, Optional
|
|
|
|
import soundfile as sf
|
|
|
|
from core.config import DUB_DIR
|
|
from fastapi import HTTPException
|
|
from services.ffmpeg_utils import find_ffmpeg, find_ffprobe, _get_semaphore, _spawn_with_retry
|
|
from services.model_manager import get_best_device
|
|
# Process lifecycle moved to its own leaf module so ffmpeg_utils can import
|
|
# it at module top (no dub_pipeline ↔ ffmpeg_utils cycle). Re-exported here —
|
|
# dub_core and tests still alias these names through this module.
|
|
from services.proc_registry import ( # noqa: F401 — re-exports
|
|
_active_procs,
|
|
_active_procs_lock,
|
|
has_active_procs,
|
|
kill_job_procs,
|
|
register_proc,
|
|
unregister_proc,
|
|
)
|
|
from core.db import db_conn
|
|
from core import event_bus
|
|
from core import failure
|
|
|
|
logger = logging.getLogger("omnivoice.dub_pipeline")
|
|
|
|
# ── Module-level state ──────────────────────────────────────────────────────
|
|
# These used to live in dub_core.py. The router now re-exports them for
|
|
# backward compat during the transition.
|
|
|
|
_dub_jobs: dict[str, dict] = {}
|
|
_dub_jobs_lock = threading.Lock()
|
|
|
|
_DUB_DIR_REAL = os.path.realpath(DUB_DIR)
|
|
_HASH_BUF_SIZE = 1 << 18 # 256 KB chunks for hashing
|
|
|
|
|
|
# ── Pure helpers ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def compute_file_hash(path: str) -> str:
|
|
"""SHA-256 digest of a file, streamed in 256 KB chunks."""
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
while True:
|
|
chunk = f.read(_HASH_BUF_SIZE)
|
|
if not chunk:
|
|
break
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def safe_job_dir(job_id: str) -> Optional[str]:
|
|
"""Resolve a job directory under DUB_DIR, rejecting traversal."""
|
|
if not job_id or "/" in job_id or "\\" in job_id or job_id in (".", ".."):
|
|
return None
|
|
candidate = os.path.realpath(os.path.join(DUB_DIR, job_id))
|
|
if not candidate.startswith(_DUB_DIR_REAL + os.sep):
|
|
return None
|
|
return candidate
|
|
|
|
|
|
def sse_event(event: str, payload) -> bytes:
|
|
"""Encode one Server-Sent Event frame. UTF-8 bytes, ready to yield."""
|
|
return f"event: {event}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n".encode("utf-8")
|
|
|
|
|
|
def prep_event(event_type: str, **fields) -> str:
|
|
"""Build a `data:` SSE line for the ingest pipeline (plain `data: {...}`)."""
|
|
return f"data: {json.dumps({'type': event_type, **fields})}\n\n"
|
|
|
|
|
|
# ── Content-hash cache lookup ───────────────────────────────────────────────
|
|
|
|
|
|
def find_cached_job(content_hash: str, exclude_job_id: str) -> Optional[dict]:
|
|
"""Find a previous job with the same content hash that has the heavy
|
|
artifacts (vocals/no-vocals/scene cuts) still on disk. Returns a dict
|
|
of paths + metadata the caller can shallow-copy into the new job, or
|
|
None if no usable cache exists.
|
|
"""
|
|
if not content_hash:
|
|
return None
|
|
with db_conn() as conn:
|
|
rows = conn.execute(
|
|
"SELECT id, job_data FROM dub_history WHERE content_hash=? AND id!=? ORDER BY created_at DESC LIMIT 5",
|
|
(content_hash, exclude_job_id),
|
|
).fetchall()
|
|
for row in rows:
|
|
try:
|
|
job = json.loads(row["job_data"])
|
|
except (json.JSONDecodeError, TypeError):
|
|
continue
|
|
cached_dir = safe_job_dir(row["id"])
|
|
if not cached_dir or not os.path.isdir(cached_dir):
|
|
continue
|
|
vocals = job.get("vocals_path") or os.path.join(cached_dir, "vocals.wav")
|
|
if not os.path.isfile(vocals):
|
|
continue
|
|
# Separation-quality gate: stems produced before the HQ-extraction
|
|
# change were separated from the 16 kHz MONO ASR file — a mono,
|
|
# 8 kHz-ceiling music bed. audio_hq.wav in the cached job dir is the
|
|
# marker that its stems came from the full-quality stereo extraction;
|
|
# without it, reusing the cache would silently keep serving the
|
|
# narrow-band mono bed forever for that video. Re-separating once is
|
|
# the better deal.
|
|
if not os.path.isfile(os.path.join(cached_dir, "audio_hq.wav")):
|
|
logger.info(
|
|
"cache candidate %s has pre-HQ (mono/16k-derived) stems — "
|
|
"skipping reuse so separation reruns at full quality", row["id"],
|
|
)
|
|
continue
|
|
return {
|
|
"job_dir": cached_dir,
|
|
"job_id": row["id"],
|
|
"vocals_path": vocals,
|
|
"no_vocals_path": job.get("no_vocals_path"),
|
|
"scene_cuts": job.get("scene_cuts") or [],
|
|
"thumb_path": job.get("thumb_path"),
|
|
"duration": job.get("duration", 0.0),
|
|
}
|
|
return None
|
|
|
|
|
|
# ── Process lifecycle ───────────────────────────────────────────────────────
|
|
# register_proc / unregister_proc / kill_job_procs / has_active_procs live in
|
|
# services.proc_registry (imported + re-exported above).
|
|
|
|
|
|
# ── Job state (in-memory + SQLite fallback) ────────────────────────────────
|
|
|
|
|
|
def get_job(job_id: str) -> Optional[dict]:
|
|
"""Look up a job. Checks the in-memory cache first, then falls back to
|
|
`dub_history.job_data` so saved projects still resolve after restart.
|
|
"""
|
|
with _dub_jobs_lock:
|
|
if job_id in _dub_jobs:
|
|
return _dub_jobs[job_id]
|
|
with db_conn() as conn:
|
|
row = conn.execute("SELECT job_data FROM dub_history WHERE id=?", (job_id,)).fetchone()
|
|
if row and row["job_data"]:
|
|
try:
|
|
job = json.loads(row["job_data"])
|
|
with _dub_jobs_lock:
|
|
_dub_jobs[job_id] = job
|
|
return job
|
|
except json.JSONDecodeError:
|
|
# job_id arrives from request paths — strip newlines so a crafted
|
|
# id can't forge extra log lines (py/log-injection).
|
|
safe_id = str(job_id).replace("\r", "").replace("\n", "")
|
|
logger.exception("Failed to decode dub_history.job_data for %s", safe_id)
|
|
return None
|
|
|
|
|
|
def put_job(job_id: str, job: dict) -> None:
|
|
"""Insert / replace the in-memory job record. Does NOT persist."""
|
|
with _dub_jobs_lock:
|
|
_dub_jobs[job_id] = job
|
|
|
|
|
|
def save_job(job_id: str, job: dict, filename: str = "", duration: float = 0.0, content_hash: str = "") -> None:
|
|
"""Persist dub job state to SQLite so it survives restarts. Uses UPSERT
|
|
on `id` so repeated saves in a session keep the latest snapshot.
|
|
|
|
language / language_code / content_hash only update when the incoming
|
|
value is non-empty: the ingest-time insert runs before the target
|
|
language is known (both columns ""), generation sets them on the job
|
|
dict, and a later save from a job that lost them (e.g. hydrated from an
|
|
old row) must not clobber the healed columns back to "". The frontend
|
|
keys history restore off language_code, so a frozen "" hid finished
|
|
tracks until the user re-picked a language.
|
|
"""
|
|
try:
|
|
segments = job.get("segments") or []
|
|
tracks = list((job.get("dubbed_tracks") or {}).keys())
|
|
with db_conn() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO dub_history
|
|
(id, filename, duration, segments_count, language, language_code, tracks, job_data, content_hash, created_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
filename=excluded.filename,
|
|
duration=excluded.duration,
|
|
segments_count=excluded.segments_count,
|
|
language=CASE WHEN excluded.language != '' THEN excluded.language ELSE dub_history.language END,
|
|
language_code=CASE WHEN excluded.language_code != '' THEN excluded.language_code ELSE dub_history.language_code END,
|
|
tracks=excluded.tracks,
|
|
job_data=excluded.job_data,
|
|
content_hash=CASE WHEN excluded.content_hash != '' THEN excluded.content_hash ELSE dub_history.content_hash END""",
|
|
(job_id, filename or job.get("filename", ""),
|
|
duration or job.get("duration", 0.0),
|
|
len(segments), job.get("language", ""), job.get("language_code", ""),
|
|
json.dumps(tracks), json.dumps(job, default=str), content_hash or "", time.time()),
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to persist dub job %s", job_id)
|
|
return
|
|
event_bus.emit("dub_history", {"action": "saved", "id": job_id})
|
|
|
|
|
|
# ── Ingest pipeline (download → extract → demucs → scene → thumb) ──────────
|
|
|
|
|
|
def run_proc_factory(job_id: str):
|
|
"""Return an async `run_proc` helper bound to this job_id.
|
|
|
|
Spawns subprocesses under the shared ffmpeg semaphore, tracks them so
|
|
`kill_job_procs(job_id)` can terminate them, raises HTTP 504 on timeout.
|
|
"""
|
|
async def run_proc(cmd, timeout: float = 900.0):
|
|
async with _get_semaphore():
|
|
p = await _spawn_with_retry(
|
|
cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
register_proc(job_id, p)
|
|
try:
|
|
try:
|
|
stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=timeout)
|
|
except asyncio.TimeoutError:
|
|
try:
|
|
p.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
await asyncio.wait_for(p.wait(), timeout=5.0)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
raise HTTPException(status_code=504, detail=f"subprocess timed out after {timeout}s")
|
|
return p, stdout, stderr
|
|
finally:
|
|
unregister_proc(job_id, p)
|
|
if p.returncode is None:
|
|
try:
|
|
p.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
await asyncio.wait_for(p.wait(), timeout=5.0)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
return run_proc
|
|
|
|
|
|
async def run_proc_streaming_stderr(
|
|
job_id: str,
|
|
cmd: list[str],
|
|
*,
|
|
timeout: float = 1800.0,
|
|
) -> AsyncIterator[tuple]:
|
|
"""Spawn `cmd` and stream its stderr line-by-line as it runs.
|
|
|
|
Yields:
|
|
('stderr', line: str) — once per logical line (split on \\r or \\n,
|
|
so tqdm-style progress bars that overwrite the same line via
|
|
carriage return surface as separate lines)
|
|
('done', returncode: int, full_stderr: bytes) — exactly once when
|
|
the subprocess exits; this is always the final value.
|
|
|
|
Honors the same semaphore + register_proc/kill plumbing as run_proc.
|
|
"""
|
|
async with _get_semaphore():
|
|
p = await _spawn_with_retry(
|
|
cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
register_proc(job_id, p)
|
|
stderr_parts: list[bytes] = []
|
|
rc: int = -1
|
|
try:
|
|
if getattr(p, "uses_sync_pipes", False):
|
|
# Fallback loops (the Windows SelectorEventLoop uvicorn forces
|
|
# under --reload) hand back a thread-backed proc whose .stderr is
|
|
# a plain SYNC pipe, not an asyncio StreamReader — `await
|
|
# p.stderr.read()` there raises "a coroutine or an awaitable is
|
|
# required" and crashed the demucs step. We can't stream that
|
|
# pipe incrementally without leaking blocked executor threads on
|
|
# every 1s poll, so run to completion via the wrapper's async
|
|
# communicate() and replay stderr as the same line events. No
|
|
# live progress on that degraded loop, but the subprocess still
|
|
# runs and the emitted event sequence is identical. The native
|
|
# async path (Proactor/posix — every release build) is the
|
|
# unchanged `else` below.
|
|
try:
|
|
_out, err_bytes = await asyncio.wait_for(
|
|
p.communicate(), timeout=timeout
|
|
)
|
|
except asyncio.TimeoutError:
|
|
try:
|
|
p.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
raise HTTPException(
|
|
status_code=504,
|
|
detail=f"subprocess timed out after {timeout}s",
|
|
)
|
|
err_bytes = err_bytes or b""
|
|
stderr_parts.append(err_bytes)
|
|
for _line in re.split(rb"[\r\n]", err_bytes):
|
|
_text = _line.decode(errors="replace")
|
|
if _text.strip():
|
|
yield ("stderr", _text)
|
|
else:
|
|
buf = b""
|
|
start = time.monotonic()
|
|
while True:
|
|
if time.monotonic() - start > timeout:
|
|
try:
|
|
p.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
raise HTTPException(
|
|
status_code=504,
|
|
detail=f"subprocess timed out after {timeout}s",
|
|
)
|
|
try:
|
|
chunk = await asyncio.wait_for(p.stderr.read(256), timeout=1.0)
|
|
except asyncio.TimeoutError:
|
|
if p.returncode is not None:
|
|
break
|
|
continue
|
|
if not chunk:
|
|
break
|
|
stderr_parts.append(chunk)
|
|
buf += chunk
|
|
while True:
|
|
idx_r = buf.find(b"\r")
|
|
idx_n = buf.find(b"\n")
|
|
if idx_r < 0 and idx_n < 0:
|
|
break
|
|
if idx_r < 0:
|
|
idx = idx_n
|
|
elif idx_n < 0:
|
|
idx = idx_r
|
|
else:
|
|
idx = min(idx_r, idx_n)
|
|
line = buf[:idx].decode(errors="replace")
|
|
buf = buf[idx + 1:]
|
|
if line.strip():
|
|
yield ("stderr", line)
|
|
rc = await p.wait()
|
|
finally:
|
|
unregister_proc(job_id, p)
|
|
if p.returncode is None:
|
|
try:
|
|
p.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
await asyncio.wait_for(p.wait(), timeout=5.0)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
try:
|
|
p.stdout.close()
|
|
except Exception:
|
|
pass
|
|
yield ("done", rc, b"".join(stderr_parts))
|
|
|
|
|
|
_BROWSER_VIDEO_CODECS = {"h264", "avc1"}
|
|
_BROWSER_AUDIO_CODECS = {"aac", "mp4a"}
|
|
|
|
|
|
def _probe_codecs(path: str) -> tuple[str, str]:
|
|
"""Return (video_codec, audio_codec) lowercased, or ('','') on probe error."""
|
|
ffprobe = find_ffprobe()
|
|
if not ffprobe:
|
|
return ("", "")
|
|
try:
|
|
out = subprocess.run(
|
|
[ffprobe, "-v", "error", "-select_streams", "v:0",
|
|
"-show_entries", "stream=codec_name", "-of", "csv=p=0", path],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
vcodec = (out.stdout or "").strip().lower()
|
|
out = subprocess.run(
|
|
[ffprobe, "-v", "error", "-select_streams", "a:0",
|
|
"-show_entries", "stream=codec_name", "-of", "csv=p=0", path],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
acodec = (out.stdout or "").strip().lower()
|
|
return vcodec, acodec
|
|
except Exception:
|
|
return ("", "")
|
|
|
|
|
|
def _ensure_browser_playable_mp4(video_path: str) -> str:
|
|
"""Guarantee `video_path` is an mp4 with h264 video + aac audio.
|
|
|
|
Three classes of fix the in-app `<video>` element needs:
|
|
1. `.webm`/`.mkv` containers — WKWebView refuses them outright.
|
|
2. `.mp4` with VP9 or AV1 video — WKWebView can't decode either
|
|
inside an mp4 container (Safari ships VP9 only in WebM).
|
|
3. `.mp4` with Opus audio — same story; mp4-with-opus is rare and
|
|
poorly supported across webviews.
|
|
|
|
Strategy: probe codecs; if extension is mp4 AND both codecs are in
|
|
the browser-safe set, leave alone. Otherwise transcode to h264+aac.
|
|
Returns the (possibly new) file path.
|
|
"""
|
|
ffmpeg_bin = find_ffmpeg()
|
|
is_mp4 = video_path.lower().endswith(".mp4")
|
|
if is_mp4:
|
|
vcodec, acodec = _probe_codecs(video_path)
|
|
if vcodec in _BROWSER_VIDEO_CODECS and acodec in _BROWSER_AUDIO_CODECS:
|
|
return video_path # already safe — fast path, no rewrite
|
|
# Need to rewrite. Try stream-copy first when the container is the
|
|
# only problem; fall back to full transcode for codec mismatches.
|
|
target = os.path.splitext(video_path)[0] + ".mp4"
|
|
if target == video_path:
|
|
target = os.path.splitext(video_path)[0] + ".browser.mp4"
|
|
# Stream-copy attempt — works when codecs are already h264+aac but
|
|
# the container is wrong (rare with our format selector but cheap to
|
|
# try). Skip straight to transcode if we already know codecs are bad.
|
|
rc = 1
|
|
if is_mp4:
|
|
# Codecs were probed and known-bad; skip the copy attempt.
|
|
pass
|
|
else:
|
|
rc = subprocess.run(
|
|
[ffmpeg_bin, "-y", "-i", video_path,
|
|
"-c:v", "copy", "-c:a", "copy",
|
|
"-movflags", "+faststart", target],
|
|
capture_output=True,
|
|
).returncode
|
|
if rc != 0 or not os.path.exists(target):
|
|
rc = 1
|
|
if rc != 0:
|
|
# Full transcode — h264 baseline-ish + aac is the safe combo.
|
|
rc = subprocess.run(
|
|
[ffmpeg_bin, "-y", "-i", video_path,
|
|
"-c:v", "libx264", "-preset", "veryfast", "-crf", "23",
|
|
"-pix_fmt", "yuv420p",
|
|
"-c:a", "aac", "-b:a", "192k",
|
|
"-movflags", "+faststart", target],
|
|
capture_output=True,
|
|
).returncode
|
|
if rc == 0 and os.path.exists(target) and target != video_path:
|
|
try:
|
|
os.remove(video_path)
|
|
except OSError:
|
|
pass
|
|
return target
|
|
if rc != 0:
|
|
logger.warning(
|
|
"Could not transcode %s to browser-playable mp4 — the in-app "
|
|
"video player may render this file as a black box.",
|
|
video_path,
|
|
)
|
|
return video_path
|
|
|
|
|
|
# Bounded retry for transient download failures (#579/#598). yt-dlp's own
|
|
# `retries`/`fragment_retries` cover per-fragment HTTP flakes, but a broken
|
|
# pipe ([Errno 32]) raised while the write side of a pipe closes mid-stream
|
|
# (a killed ffmpeg merge child, a CDN reset during muxing) aborts the whole
|
|
# `extract_info` call and is NOT covered by them — so a single transient blip
|
|
# failed the entire ingest with a raw "Broken pipe". We add a small download-
|
|
# level retry on top, cleaning up the partial download between attempts so a
|
|
# half-written `original.*` can't poison the next try.
|
|
_YT_DOWNLOAD_RETRIES = 2 # total attempts = 1 + retries = 3
|
|
|
|
|
|
def _with_target_facts(exc: BaseException, job_dir: str) -> BaseException:
|
|
"""``exc`` with the download destination described, when the failure looks
|
|
like the OS refusing a file operation (#1225).
|
|
|
|
Returns ``exc`` untouched for network/format failures — their message is
|
|
already about the remote side, and appending disk facts would just be
|
|
noise. Never raises."""
|
|
try:
|
|
# Shared with failure.classify() so the "is this a disk problem?"
|
|
# answer can't differ between the class we assign and whether we
|
|
# bother naming the folder (#1225 review).
|
|
if not failure.is_os_write_refusal(str(exc)):
|
|
return exc
|
|
facts = failure.describe_path_target(os.path.join(job_dir, "original.mp4"))
|
|
if not facts:
|
|
return exc
|
|
msg = (
|
|
f"{exc} — saving to {job_dir} ({facts}). The OS refused the write, "
|
|
f"so retrying the same link won't help: check the drive isn't full, "
|
|
f"the folder is writable, and antivirus or a cloud-sync client "
|
|
f"(OneDrive, Dropbox) isn't locking it."
|
|
)
|
|
try:
|
|
return type(exc)(msg)
|
|
except Exception:
|
|
# Not every exception class takes a plain message (soundfile's
|
|
# LibsndfileError wants an int code). Keep the text, drop the type.
|
|
return RuntimeError(msg)
|
|
except Exception:
|
|
return exc
|
|
|
|
|
|
def _is_transient_download_error(exc: BaseException) -> bool:
|
|
"""True when a download failure is worth retrying (broken pipe / net drop).
|
|
|
|
Reuses the single failure taxonomy (`VIDEO_DOWNLOAD_NETWORK`) rather than a
|
|
parallel keyword list, so "what counts as transient" stays single-sourced
|
|
with the error-hint classification. ``BrokenPipeError``/``ConnectionError``
|
|
are matched by class too, since a bare instance may be wrapped or re-raised
|
|
with a stripped message that no longer contains "broken pipe".
|
|
"""
|
|
if isinstance(exc, (BrokenPipeError, ConnectionError)):
|
|
return True
|
|
return failure.classify(str(exc)) == "VIDEO_DOWNLOAD_NETWORK"
|
|
|
|
|
|
# YouTube serves some videos' high-quality formats signature-protected to the
|
|
# default player client, so the media download 403s even though extraction
|
|
# worked. Forcing an alternate client commonly bypasses it; on a 403 we escalate
|
|
# through these (in order) before giving up (#625).
|
|
_YT_PLAYER_CLIENTS = ["tv", "android", "web_safari"]
|
|
|
|
|
|
def _is_forbidden_download_error(exc: BaseException) -> bool:
|
|
"""True for an HTTP 403 — not transient (the same client keeps 403ing), but
|
|
often fixable by switching the YouTube player client."""
|
|
s = str(exc)
|
|
return "403" in s or "Forbidden" in s
|
|
|
|
|
|
def _cleanup_partial_download(job_dir: str) -> None:
|
|
"""Remove any half-written `original.*` files before a retry.
|
|
|
|
A partial download left on disk would otherwise be picked up as a "finished"
|
|
file by the post-download codec probe, or collide with the next attempt's
|
|
output. Best-effort — never raises on the failure path.
|
|
"""
|
|
import glob
|
|
for stale in glob.glob(os.path.join(job_dir, "original.*")):
|
|
try:
|
|
os.remove(stale)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def yt_download_sync(
|
|
url: str,
|
|
job_dir: str,
|
|
*,
|
|
fetch_subs: bool = False,
|
|
sub_langs: list[str] | None = None,
|
|
progress_hook=None,
|
|
) -> tuple[str, str, list[str]]:
|
|
"""Blocking yt-dlp download into `job_dir`.
|
|
|
|
Returns (video_path, title, downloaded_sub_files).
|
|
|
|
Captions are downloaded in a separate, best-effort pass after the video
|
|
so subtitle failures (rate-limits, missing tracks) can never derail the
|
|
actual ingest. Defaults skip YouTube's auto-translated set — requesting
|
|
"all" expands into ~100 per-language variants per video and reliably
|
|
trips HTTP 429, and the downstream Translate step handles target
|
|
languages more reliably than YouTube's machine translations anyway.
|
|
Pass an explicit `sub_langs` list to override the default selection.
|
|
"""
|
|
import glob
|
|
import yt_dlp
|
|
outtmpl = os.path.join(job_dir, "original.%(ext)s")
|
|
# #1225: yt-dlp surfaces an OS write rejection as a bare
|
|
# "Unable to download video: [Errno 22] Invalid argument" — no path, no
|
|
# reason, and three manual retries all fail identically because nothing
|
|
# about it is transient. Fail here instead, naming the directory, when we
|
|
# can already see it won't work.
|
|
_target_facts = failure.describe_path_target(outtmpl)
|
|
if "not writable" in _target_facts or "does not exist" in _target_facts:
|
|
# Worded so classify() places it in the download path: it must carry
|
|
# both an OS-refusal signature and download context, or the user gets
|
|
# no hint at all — the failure this PR exists to fix (#1225 review).
|
|
raise OSError(
|
|
f"Unable to download video: unable to open for writing in "
|
|
f"{job_dir} ({_target_facts}). The video downloads into this job "
|
|
f"folder under your OmniVoice data directory — check it exists, is "
|
|
f"writable, and isn't locked by antivirus or a cloud-sync client."
|
|
)
|
|
ydl_opts: dict = {
|
|
"outtmpl": outtmpl,
|
|
# Prefer h264+aac streams so the merged mp4 is natively decodable
|
|
# by WKWebView/WebView2/Chromium. YouTube serves VP9+Opus as the
|
|
# default high-quality combo, which yt-dlp will happily mux into
|
|
# an mp4 container — but Safari/WKWebView refuses to decode VP9
|
|
# inside mp4, leaving a black <video> in the dub editor. Fall
|
|
# back to any combo only when no h264/aac variant exists, then
|
|
# the post-download codec probe below will transcode it.
|
|
"format": (
|
|
"bv*[vcodec^=avc1][ext=mp4]+ba[acodec^=mp4a][ext=m4a]/"
|
|
"bv*[vcodec^=avc1]+ba[acodec^=mp4a]/"
|
|
"b[vcodec^=avc1][acodec^=mp4a]/"
|
|
"bv*[vcodec^=avc1]+ba/"
|
|
"b[vcodec^=avc1]/"
|
|
"bv*+ba/b"
|
|
),
|
|
"merge_output_format": "mp4",
|
|
"noplaylist": True,
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
"restrictfilenames": True,
|
|
# Don't stamp the downloaded file's mtime with the video's upload date
|
|
# (#642): on Windows an out-of-range/invalid timestamp makes the os.utime
|
|
# call raise `[Errno 22] Invalid argument`, failing the whole ingest. We
|
|
# download to a throwaway `original.*` and never use its mtime, so skip
|
|
# it entirely (equivalent to yt-dlp's --no-mtime).
|
|
"updatetime": False,
|
|
"socket_timeout": 30,
|
|
# Resilience against YouTube CDN flakes: a single empty fragment
|
|
# (commonly the very last one — "Did not get any data blocks")
|
|
# used to fail the whole ingest at 99% complete. Retry each
|
|
# fragment generously and tolerate one that never returns data;
|
|
# missing the last <0.5% of audio is acceptable for transcription.
|
|
"fragment_retries": 10,
|
|
"retries": 10,
|
|
"extractor_retries": 5,
|
|
"skip_unavailable_fragments": True,
|
|
}
|
|
# #712: the format selector above pulls separate video+audio streams, so
|
|
# yt-dlp muxes them via ffmpeg (merge_output_format=mp4). yt-dlp only looks
|
|
# for ffmpeg on PATH and aborts with "you have requested merging of multiple
|
|
# formats but ffmpeg is not installed" — but OmniVoice's ffmpeg is often a
|
|
# bundled Tauri sidecar / imageio-ffmpeg binary that isn't on PATH (common on
|
|
# Windows). Point yt-dlp at the exact ffmpeg we resolve so the merge works.
|
|
_ffmpeg_bin = find_ffmpeg()
|
|
if _ffmpeg_bin:
|
|
ydl_opts["ffmpeg_location"] = _ffmpeg_bin
|
|
if progress_hook is not None:
|
|
ydl_opts["progress_hooks"] = [progress_hook]
|
|
|
|
# Download with a bounded retry on transient/broken-pipe-class failures
|
|
# (#579/#598). A broken pipe mid-mux isn't recoverable inside yt-dlp's own
|
|
# fragment retries, but a fresh `extract_info` usually succeeds. Between
|
|
# attempts we wipe the partial `original.*` so a half-written file can't be
|
|
# mistaken for a finished download.
|
|
info = None
|
|
path = None
|
|
transient_used = 0
|
|
client_idx = 0
|
|
while True:
|
|
try:
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
info = ydl.extract_info(url, download=True)
|
|
path = ydl.prepare_filename(info)
|
|
break
|
|
except Exception as exc:
|
|
_cleanup_partial_download(job_dir)
|
|
# 403 Forbidden: not transient — escalate the YouTube player client,
|
|
# which commonly bypasses a signature-protected format set (#625).
|
|
if _is_forbidden_download_error(exc) and client_idx < len(_YT_PLAYER_CLIENTS):
|
|
client = _YT_PLAYER_CLIENTS[client_idx]
|
|
client_idx += 1
|
|
ydl_opts = {**ydl_opts, "extractor_args": {"youtube": {"player_client": [client]}}}
|
|
logger.warning(
|
|
"Download 403 for %s — retrying with player_client=%s (#625)", url, client,
|
|
)
|
|
continue
|
|
# Transient/broken-pipe: a fresh extract_info usually succeeds
|
|
# (#579/#598). A 403 never counts here — it's escalated above.
|
|
if (transient_used < _YT_DOWNLOAD_RETRIES
|
|
and _is_transient_download_error(exc)
|
|
and not _is_forbidden_download_error(exc)):
|
|
transient_used += 1
|
|
logger.warning(
|
|
"Transient download failure for %s (attempt %d/%d): %s — retrying",
|
|
url, transient_used, _YT_DOWNLOAD_RETRIES, exc,
|
|
)
|
|
time.sleep(2 * transient_used) # brief, increasing backoff
|
|
continue
|
|
# #1225: an OS-level rejection (errno 22 / EACCES / ENOSPC) tells
|
|
# the user nothing on its own. Attach what we can observe about
|
|
# the destination so the message identifies a full drive, a
|
|
# removed folder, or an antivirus/cloud-sync lock. Wording keeps
|
|
# the yt-dlp text so classify() still sees the download context.
|
|
described = _with_target_facts(exc, job_dir)
|
|
if described is exc:
|
|
raise
|
|
raise described from exc
|
|
root, _ = os.path.splitext(path)
|
|
mp4 = root + ".mp4"
|
|
if os.path.exists(mp4):
|
|
video_path = mp4
|
|
else:
|
|
video_path = path
|
|
# Browser-playability guard: WKWebView (Tauri on macOS) refuses to
|
|
# decode VP9/AV1 video and Opus audio even when they're wrapped in an
|
|
# mp4 container, and refuses .webm/.mkv outright. We probe the actual
|
|
# codecs and transcode only when needed — `<video>` rendering is
|
|
# what we care about, not what extension yt-dlp ended up writing.
|
|
video_path = _ensure_browser_playable_mp4(video_path)
|
|
title = info.get("title") or os.path.basename(video_path)
|
|
|
|
sub_files: list[str] = []
|
|
if fetch_subs:
|
|
if sub_langs:
|
|
langs = list(sub_langs)
|
|
else:
|
|
orig = (info.get("language") or "").strip()
|
|
manual = list((info.get("subtitles") or {}).keys())
|
|
langs = sorted({*manual, *([orig] if orig else [])})
|
|
if not langs:
|
|
logger.info("No captions available on %s (skipping subtitle pass)", url)
|
|
else:
|
|
sub_opts = {
|
|
**ydl_opts,
|
|
"skip_download": True,
|
|
"writesubtitles": True,
|
|
"writeautomaticsub": True,
|
|
"subtitleslangs": langs,
|
|
"subtitlesformat": "vtt",
|
|
"extractor_args": {"youtube": {"skip": ["translated_subs"]}},
|
|
"ignoreerrors": True,
|
|
"extractor_retries": 5,
|
|
"sleep_interval_subtitles": 1,
|
|
}
|
|
try:
|
|
with yt_dlp.YoutubeDL(sub_opts) as ydl_sub:
|
|
ydl_sub.extract_info(url, download=True)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Subtitle download failed for %s (continuing with video): %s",
|
|
url, e,
|
|
)
|
|
base = os.path.splitext(video_path)[0]
|
|
sub_files = sorted(glob.glob(base + ".*.vtt"))
|
|
return video_path, title, sub_files
|
|
|
|
|
|
def parse_vtt_segments(vtt_path: str) -> list[dict]:
|
|
"""Very small WEBVTT parser → list of {start, end, text}.
|
|
|
|
Purpose: turn yt-dlp's downloaded caption track into the same segment
|
|
shape the dub pipeline's transcript step produces, so the UI can seed
|
|
its editor from YouTube captions instead of running Whisper. We don't
|
|
care about positioning cues or styling — just the timed text.
|
|
"""
|
|
try:
|
|
with open(vtt_path, "r", encoding="utf-8") as f:
|
|
raw = f.read()
|
|
except OSError:
|
|
return []
|
|
|
|
def _ts(s: str) -> float:
|
|
# Format: HH:MM:SS.mmm or MM:SS.mmm
|
|
parts = s.strip().split(":")
|
|
try:
|
|
parts = [float(p.replace(",", ".")) for p in parts]
|
|
except ValueError:
|
|
return 0.0
|
|
if len(parts) == 3:
|
|
h, m, sec = parts
|
|
elif len(parts) == 2:
|
|
h, m, sec = 0.0, parts[0], parts[1]
|
|
else:
|
|
return 0.0
|
|
return h * 3600.0 + m * 60.0 + sec
|
|
|
|
segments: list[dict] = []
|
|
blocks = raw.replace("\r\n", "\n").split("\n\n")
|
|
for block in blocks:
|
|
lines = [ln for ln in block.split("\n") if ln.strip() and not ln.startswith("WEBVTT") and not ln.startswith("NOTE")]
|
|
if not lines:
|
|
continue
|
|
# Skip numeric cue ID line if present
|
|
if "-->" not in lines[0] and len(lines) > 1:
|
|
lines = lines[1:]
|
|
if not lines or "-->" not in lines[0]:
|
|
continue
|
|
ts_line = lines[0]
|
|
try:
|
|
left, right = ts_line.split("-->")
|
|
# Drop any settings after the right timestamp ("00:01.000 align:left line:0%")
|
|
right = right.strip().split(" ")[0]
|
|
start = _ts(left)
|
|
end = _ts(right)
|
|
except Exception:
|
|
continue
|
|
text = " ".join(ln.strip() for ln in lines[1:]).strip()
|
|
# Strip inline styling like <c.colorE5E5E5>foo</c> or <00:00:01.200>
|
|
text = re.sub(r"<[^>]+>", "", text)
|
|
if text:
|
|
segments.append({"start": start, "end": end, "text": text})
|
|
return segments
|
|
|
|
|
|
async def ingest_pipeline(
|
|
job_id: str,
|
|
job_dir: str,
|
|
source: dict,
|
|
filename_hint: Optional[str] = None,
|
|
) -> AsyncIterator[str]:
|
|
"""Async generator: emit SSE events per processing stage.
|
|
|
|
Stages: download_start, download_done, extract_start, extract_done,
|
|
demucs_start, demucs_done, scene_start, scene_done, ready, error, cancelled.
|
|
"""
|
|
youtube_subs_by_lang: dict[str, list[dict]] = {}
|
|
# Audio-only jobs (#119) skip scene detection + thumbnailing below; the
|
|
# transcribe → translate → TTS core is identical.
|
|
input_type = (source.get("input_type") or "video").lower()
|
|
try:
|
|
if source.get("kind") == "url":
|
|
url = source["url"]
|
|
fetch_subs = bool(source.get("fetch_subs"))
|
|
sub_langs = source.get("sub_langs") or None
|
|
yield prep_event("download_start", url=url)
|
|
# Bridge yt-dlp's per-fragment progress callback (fires inside
|
|
# the worker thread) into the async generator via a threadsafe
|
|
# queue so the UI can render a real download bar.
|
|
loop = asyncio.get_running_loop()
|
|
dl_queue: asyncio.Queue = asyncio.Queue()
|
|
_last_pct = -1
|
|
|
|
def _yt_progress(d: dict) -> None:
|
|
nonlocal _last_pct
|
|
status = d.get("status")
|
|
if status == "downloading":
|
|
total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
|
|
cur = d.get("downloaded_bytes") or 0
|
|
if total > 0:
|
|
pct = max(0, min(100, int(cur * 100 / total)))
|
|
if pct != _last_pct:
|
|
_last_pct = pct
|
|
payload = {
|
|
"percent": pct,
|
|
"speed_bps": d.get("speed") or 0,
|
|
"eta_s": d.get("eta"),
|
|
}
|
|
loop.call_soon_threadsafe(dl_queue.put_nowait, payload)
|
|
|
|
dl_task = asyncio.create_task(asyncio.to_thread(
|
|
yt_download_sync, url, job_dir,
|
|
fetch_subs=fetch_subs, sub_langs=sub_langs,
|
|
progress_hook=_yt_progress,
|
|
))
|
|
try:
|
|
while not dl_task.done():
|
|
try:
|
|
payload = await asyncio.wait_for(dl_queue.get(), timeout=0.5)
|
|
yield prep_event("download_progress", **payload)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
# Drain any final queued events.
|
|
while not dl_queue.empty():
|
|
payload = dl_queue.get_nowait()
|
|
yield prep_event("download_progress", **payload)
|
|
video_path, title, sub_files = await dl_task
|
|
except Exception as e:
|
|
logger.exception("Download failed for job %s", job_id)
|
|
if not dl_task.done():
|
|
dl_task.cancel()
|
|
yield prep_event("error", **failure.build_failure(e, stage="download"))
|
|
shutil.rmtree(job_dir, ignore_errors=True)
|
|
return
|
|
filename = title or os.path.basename(video_path)
|
|
try:
|
|
size = os.path.getsize(video_path)
|
|
except OSError:
|
|
size = 0
|
|
# Parse each downloaded .<lang>.vtt into a segment list we can
|
|
# stash on the job. The router will merge this into the job dict
|
|
# after ingest completes so /dub/transcribe-stream (or a future
|
|
# "use-youtube-subs" toggle) can seed segments from it.
|
|
for sf_path in sub_files:
|
|
base = os.path.splitext(os.path.basename(sf_path))[0]
|
|
# "original.en" → lang "en"; fallback to the trailing token.
|
|
lang_tag = base.rsplit(".", 1)[-1] if "." in base else "und"
|
|
segs = parse_vtt_segments(sf_path)
|
|
if segs:
|
|
youtube_subs_by_lang[lang_tag] = segs
|
|
yield prep_event(
|
|
"download_done",
|
|
title=title, size=size, filename=filename,
|
|
youtube_subs=sorted(youtube_subs_by_lang.keys()),
|
|
)
|
|
else:
|
|
video_path = source["path"]
|
|
filename = filename_hint or os.path.basename(video_path)
|
|
|
|
audio_path = os.path.join(job_dir, "audio.wav")
|
|
ffmpeg = find_ffmpeg()
|
|
run_proc = run_proc_factory(job_id)
|
|
|
|
yield prep_event("extract_start")
|
|
try:
|
|
p, _, stderr = await run_proc([
|
|
ffmpeg, "-i", video_path, "-vn", "-acodec", "pcm_s16le",
|
|
"-ar", "16000", "-ac", "1", audio_path, "-y",
|
|
])
|
|
if p.returncode != 0:
|
|
msg = (stderr.decode(errors="replace") or f"ffmpeg returned exit code {p.returncode}").strip()[:500]
|
|
raise Exception(msg)
|
|
# Second, FULL-QUALITY extraction for source separation. audio.wav
|
|
# is deliberately 16 kHz mono — that's what ASR wants — but Demucs
|
|
# used to separate that same file, so the music bed inherited mono
|
|
# (stereo image destroyed: L/R correlation 1.000 vs the original's
|
|
# 0.754, measured) and an 8 kHz ceiling (nothing real above half
|
|
# the ASR rate — the bed's "muffled" sound at its source). Demucs
|
|
# resamples to 44.1 kHz internally either way, so separating the
|
|
# stereo original costs about the same and returns a true-stereo,
|
|
# full-band bed. Best-effort: on failure Demucs falls back to the
|
|
# ASR file, which is exactly the old behavior.
|
|
audio_hq_path = os.path.join(job_dir, "audio_hq.wav")
|
|
try:
|
|
p_hq, _, stderr_hq = await run_proc([
|
|
ffmpeg, "-i", video_path, "-vn", "-acodec", "pcm_s16le",
|
|
"-ar", "44100", "-ac", "2", audio_hq_path, "-y",
|
|
])
|
|
if p_hq.returncode != 0 or not os.path.exists(audio_hq_path):
|
|
logger.warning(
|
|
"HQ audio extraction failed (rc=%s) — separation falls "
|
|
"back to the 16k mono ASR file", p_hq.returncode,
|
|
)
|
|
audio_hq_path = None
|
|
except Exception as e_hq: # noqa: BLE001 — quality upgrade, never fatal
|
|
logger.warning("HQ audio extraction errored (%s) — falling back", e_hq)
|
|
audio_hq_path = None
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as e:
|
|
logger.exception("Extract failed for job %s", job_id)
|
|
yield prep_event("error", **failure.build_failure(e, stage="extract"))
|
|
return
|
|
|
|
try:
|
|
dur = float(sf.info(audio_path).frames) / float(sf.info(audio_path).samplerate)
|
|
except Exception:
|
|
dur = 0.0
|
|
|
|
# Content-hash cache: reuse artifacts from previous matching jobs.
|
|
content_hash = await asyncio.to_thread(compute_file_hash, audio_path)
|
|
cached = find_cached_job(content_hash, job_id)
|
|
if cached:
|
|
logger.info("Cache hit for job %s (hash %s) → reusing artifacts from %s",
|
|
job_id, content_hash[:12], cached["job_id"])
|
|
vocals_path = os.path.join(job_dir, "vocals.wav")
|
|
no_vocals_path = os.path.join(job_dir, "no_vocals.wav")
|
|
thumb_path = os.path.join(job_dir, "thumb.jpg")
|
|
|
|
if cached["vocals_path"] and os.path.isfile(cached["vocals_path"]):
|
|
shutil.copy2(cached["vocals_path"], vocals_path)
|
|
else:
|
|
vocals_path = audio_path
|
|
if cached["no_vocals_path"] and os.path.isfile(cached["no_vocals_path"]):
|
|
shutil.copy2(cached["no_vocals_path"], no_vocals_path)
|
|
else:
|
|
no_vocals_path = None
|
|
if cached["thumb_path"] and os.path.isfile(cached["thumb_path"]):
|
|
shutil.copy2(cached["thumb_path"], thumb_path)
|
|
else:
|
|
thumb_path = None
|
|
|
|
scene_cuts = cached["scene_cuts"] or []
|
|
|
|
full_job = {
|
|
"video_path": video_path,
|
|
"audio_path": audio_path,
|
|
"vocals_path": vocals_path,
|
|
"no_vocals_path": no_vocals_path,
|
|
"thumb_path": thumb_path if thumb_path and os.path.exists(thumb_path) else None,
|
|
"duration": dur,
|
|
"filename": filename,
|
|
"segments": None,
|
|
"dubbed_tracks": {},
|
|
"scene_cuts": scene_cuts,
|
|
"youtube_subs": youtube_subs_by_lang or None,
|
|
"input_type": input_type,
|
|
}
|
|
put_job(job_id, full_job)
|
|
save_job(job_id, full_job, filename, dur, content_hash)
|
|
yield prep_event("extract_done", job_id=job_id, duration=round(dur, 2), filename=filename)
|
|
yield prep_event("cached",
|
|
has_bg=bool(no_vocals_path and os.path.exists(no_vocals_path)),
|
|
scene_count=len(scene_cuts))
|
|
yield prep_event("ready", job_id=job_id, duration=round(dur, 2), filename=filename)
|
|
|
|
else:
|
|
# Full pipeline: demucs → scene → thumbnail.
|
|
partial = {
|
|
"video_path": video_path,
|
|
"audio_path": audio_path,
|
|
"vocals_path": audio_path, # fallback until demucs completes
|
|
"no_vocals_path": None,
|
|
"thumb_path": None,
|
|
"duration": dur,
|
|
"filename": filename,
|
|
"segments": None,
|
|
"dubbed_tracks": {},
|
|
"scene_cuts": [],
|
|
"youtube_subs": youtube_subs_by_lang or None,
|
|
"input_type": input_type,
|
|
}
|
|
put_job(job_id, partial)
|
|
save_job(job_id, partial, filename, dur, content_hash)
|
|
yield prep_event("extract_done", job_id=job_id, duration=round(dur, 2), filename=filename)
|
|
|
|
vocals_path = os.path.join(job_dir, "vocals.wav")
|
|
no_vocals_path = os.path.join(job_dir, "no_vocals.wav")
|
|
scene_cuts: list = []
|
|
|
|
yield prep_event("demucs_start")
|
|
try:
|
|
demucs_cmd = [sys.executable, "-m", "demucs.separate",
|
|
"--two-stems", "vocals", "-n", "htdemucs", "-d", get_best_device(),
|
|
audio_hq_path or audio_path, "-o", job_dir]
|
|
rc = -1
|
|
stderr_full = b""
|
|
last_pct = -1
|
|
# demucs writes a tqdm progress bar to stderr as
|
|
# " 42%|████ | …" — surface each new integer percent
|
|
# to the UI so the user sees the bar instead of a static
|
|
# spinner during the multi-minute separation step.
|
|
async for evt in run_proc_streaming_stderr(job_id, demucs_cmd, timeout=1800.0):
|
|
if evt[0] == "stderr":
|
|
m = re.search(r"(\d{1,3})%", evt[1])
|
|
if m:
|
|
pct = max(0, min(100, int(m.group(1))))
|
|
if pct != last_pct:
|
|
last_pct = pct
|
|
yield prep_event("demucs_progress", percent=pct)
|
|
elif evt[0] == "done":
|
|
rc, stderr_full = evt[1], evt[2]
|
|
if rc != 0:
|
|
raise Exception(stderr_full.decode(errors="replace")[:500])
|
|
# Stems land under the INPUT's basename ("audio_hq" when the
|
|
# full-quality extraction succeeded, "audio" on its fallback).
|
|
demucs_out = os.path.join(
|
|
job_dir, "htdemucs",
|
|
os.path.splitext(os.path.basename(audio_hq_path or audio_path))[0],
|
|
)
|
|
if os.path.exists(os.path.join(demucs_out, "vocals.wav")):
|
|
shutil.move(os.path.join(demucs_out, "vocals.wav"), vocals_path)
|
|
shutil.move(os.path.join(demucs_out, "no_vocals.wav"), no_vocals_path)
|
|
shutil.rmtree(os.path.join(job_dir, "htdemucs"), ignore_errors=True)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as e:
|
|
logger.warning("Demucs failed for %s, falling back to mixed audio: %s", job_id, e)
|
|
# plan-04: surface the degradation (job continues with mixed audio).
|
|
yield prep_event("warning", **failure.build_failure(e, stage="demucs", include_diagnostic=False))
|
|
vocals_path = audio_path
|
|
no_vocals_path = None
|
|
yield prep_event("demucs_done",
|
|
has_bg=bool(no_vocals_path and os.path.exists(no_vocals_path)))
|
|
|
|
# Audio-only jobs (#119) have no video to scan or thumbnail. Skip
|
|
# both ffmpeg passes but still emit scene_done (count=0) so the
|
|
# prep SSE contract the frontend waits on is unchanged.
|
|
thumb_path = os.path.join(job_dir, "thumb.jpg")
|
|
if input_type == "audio":
|
|
# No scenes/thumbnail in audio — emit the start/done pair anyway
|
|
# so the prep SSE stage sequence stays symmetric with the video
|
|
# path (the frontend's stage tracker expects both).
|
|
yield prep_event("scene_start")
|
|
yield prep_event("scene_done", count=0)
|
|
thumb_path = None
|
|
else:
|
|
yield prep_event("scene_start")
|
|
try:
|
|
p, _, stderr_scene = await run_proc([
|
|
ffmpeg, "-i", video_path, "-filter:v",
|
|
"select='gt(scene,0.3)',showinfo", "-f", "null", "-",
|
|
], timeout=600.0)
|
|
matches = re.finditer(r"pts_time:([\d\.]+)", stderr_scene.decode(errors="replace"))
|
|
scene_cuts = [float(m.group(1)) for m in matches]
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as e:
|
|
logger.warning("Scene detection failed for %s: %s", job_id, e)
|
|
yield prep_event("warning", **failure.build_failure(e, stage="scene", include_diagnostic=False))
|
|
yield prep_event("scene_done", count=len(scene_cuts))
|
|
|
|
offset = max(0.5, min(1.5, dur * 0.1)) if dur else 1.0
|
|
try:
|
|
await run_proc([
|
|
ffmpeg, "-y", "-ss", f"{offset:.2f}", "-i", video_path,
|
|
"-vframes", "1", "-vf", "scale=320:-2",
|
|
"-q:v", "4", thumb_path,
|
|
], timeout=30.0)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as e:
|
|
logger.warning("Thumbnail extraction failed for %s: %s", job_id, e)
|
|
yield prep_event("warning", **failure.build_failure(e, stage="thumbnail", include_diagnostic=False))
|
|
|
|
_dub_jobs[job_id].update({
|
|
"vocals_path": vocals_path,
|
|
"no_vocals_path": no_vocals_path,
|
|
"thumb_path": thumb_path if (thumb_path and os.path.exists(thumb_path)) else None,
|
|
"scene_cuts": scene_cuts,
|
|
})
|
|
save_job(job_id, _dub_jobs[job_id], filename, dur, content_hash)
|
|
yield prep_event("ready", job_id=job_id, duration=round(dur, 2), filename=filename)
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info("Dub prep cancelled for job %s; killing subprocesses and cleaning up", job_id)
|
|
kill_job_procs(job_id)
|
|
try:
|
|
shutil.rmtree(job_dir, ignore_errors=True)
|
|
finally:
|
|
_dub_jobs.pop(job_id, None)
|
|
yield prep_event("cancelled")
|
|
raise
|
|
except Exception as e:
|
|
# plan-04 (#131): no unhandled ingest failure may be silent. Log the
|
|
# real traceback and surface a structured, non-empty reason with stage
|
|
# context instead of letting it bubble up as a bare task error.
|
|
logger.exception("Ingest pipeline failed for job %s", job_id)
|
|
yield prep_event("error", **failure.build_failure(e, stage="ingest"))
|
|
return
|
|
finally:
|
|
with _active_procs_lock:
|
|
_active_procs.pop(job_id, None)
|