youtube fulfillment: download worker (pure core)
new core/video/youtube_download.py: fetch a wished youtube video end to end. plans the organised dest (channel/year/date template), builds yt-dlp opts from the quality profile (format_selection), runs the download (injectable factory), then on success marks the video_downloads row completed + archives to history + removes it from the wishlist; on failure archives failed and KEEPS the wish to retry. orchestration is pure (yt-dlp run + all db writes injected) + seam-tested; run_youtube_download binds the real seams for the worker thread. flows through the SAME video_downloads queue as movies/tv (model B) so the downloads page + history work for youtube for free.
This commit is contained in:
BoulderBadgeDad
parent
00d05a04f5
commit
8e071719f9
2 changed files with 377 additions and 0 deletions
224
core/video/youtube_download.py
Normal file
224
core/video/youtube_download.py
Normal file
|
|
@ -0,0 +1,224 @@
|
|||
"""YouTube download worker — the fulfillment lane for wished YouTube videos.
|
||||
|
||||
Model B (Boulder): YouTube grabs flow through the SAME ``video_downloads`` queue +
|
||||
history as movies/TV, so the Downloads page, live progress, and History modal all work
|
||||
for YouTube for free. But the mechanism is different — there's no slskd transfer to poll;
|
||||
yt-dlp fetches the stream directly. So this worker owns a YouTube download end to end:
|
||||
|
||||
pick stream (quality profile → yt-dlp format) → download into the library, organised
|
||||
as a Plex "TV by date" show (channel/Season YEAR/channel - DATE - title) → mark the
|
||||
row completed + archive it to history → remove the video from the wishlist.
|
||||
|
||||
The slskd ``download_monitor`` simply SKIPS ``source='youtube'`` rows (they have no
|
||||
transfer to match), so this lane never disturbs the movie/TV pipeline.
|
||||
|
||||
The orchestration (``process_youtube_download``) is PURE — the actual yt-dlp run and all
|
||||
DB writes are injected seams — so the lifecycle (dest planning, completion → archive +
|
||||
unwish, failure → archive, no unwish) is unit-tested without a network or a DB. Production
|
||||
(``run_youtube_download``) lazily binds the real calls and runs it on a worker thread.
|
||||
|
||||
Isolated: imports only sibling ``core.video`` modules; nothing from the music side.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Callable, Dict, Optional
|
||||
|
||||
from core.video import organization, youtube_quality
|
||||
from core.video.youtube_quality import format_selection
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
import yt_dlp
|
||||
except Exception: # noqa: BLE001 - optional at import; absence handled at call time
|
||||
yt_dlp = None
|
||||
|
||||
|
||||
def youtube_fields_from_download(dl: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Organising fields for a YouTube download row. The channel/video-title/date that
|
||||
``render_path('youtube', …)`` needs ride in ``search_ctx`` (a generic JSON column);
|
||||
fall back to the row's own columns when absent."""
|
||||
ctx = dl.get("search_ctx")
|
||||
if isinstance(ctx, str):
|
||||
try:
|
||||
ctx = json.loads(ctx)
|
||||
except (ValueError, TypeError):
|
||||
ctx = {}
|
||||
if not isinstance(ctx, dict):
|
||||
ctx = {}
|
||||
return {
|
||||
"channel": ctx.get("channel") or dl.get("title"),
|
||||
"title": ctx.get("video_title") or dl.get("title"),
|
||||
"published_at": ctx.get("published_at") or dl.get("year"),
|
||||
"youtube_id": dl.get("media_id"),
|
||||
}
|
||||
|
||||
|
||||
def plan_destination(dl: Dict[str, Any], settings: Dict[str, Any], container: str) -> Dict[str, str]:
|
||||
"""Where this video lands in the library: ``{dir, filename, path}`` under the youtube
|
||||
root (``target_dir``), organised by the youtube template. Pure."""
|
||||
ext = "." + str(container or "mp4").lstrip(".")
|
||||
return organization.render_path("youtube", dl.get("target_dir"),
|
||||
youtube_fields_from_download(dl), settings, ext)
|
||||
|
||||
|
||||
def ydl_download_opts(profile: Any, dest_dir: str, dest_stem: str,
|
||||
*, progress_hook: Optional[Callable] = None, cookie_opts: Optional[dict] = None) -> dict:
|
||||
"""The yt-dlp options dict for one download: format selection from the quality profile,
|
||||
a fixed output path (dir + stem + yt-dlp's own ext), polite defaults. Pure."""
|
||||
sel = format_selection(profile)
|
||||
opts = {
|
||||
"quiet": True,
|
||||
"no_warnings": True,
|
||||
"noplaylist": True,
|
||||
"retries": 3,
|
||||
"format": sel["format"],
|
||||
"format_sort": sel["format_sort"],
|
||||
"merge_output_format": sel["merge_output_format"],
|
||||
"paths": {"home": str(dest_dir or "")},
|
||||
"outtmpl": dest_stem + ".%(ext)s",
|
||||
}
|
||||
if cookie_opts:
|
||||
opts.update(cookie_opts)
|
||||
if progress_hook:
|
||||
opts["progress_hooks"] = [progress_hook]
|
||||
return opts
|
||||
|
||||
|
||||
def _stem_and_container(dest: Dict[str, str], container: str) -> tuple:
|
||||
"""(filename stem, final ext) — strip the ext render_path put on the filename so
|
||||
yt-dlp can own the extension during the merge."""
|
||||
fn = dest.get("filename") or "download"
|
||||
cont = str(container or "mp4").lstrip(".")
|
||||
stem = fn[:-(len(cont) + 1)] if fn.lower().endswith("." + cont.lower()) else os.path.splitext(fn)[0]
|
||||
return stem or "download", cont
|
||||
|
||||
|
||||
def download_one(video_id: Any, dest_dir: str, dest_stem: str, profile: Any, container: str,
|
||||
*, ydl_factory=None, progress_hook=None, cookie_opts=None) -> Dict[str, Any]:
|
||||
"""Run yt-dlp for ONE video into ``dest_dir/dest_stem.ext``. Returns
|
||||
``{ok, dest_path|None, error|None}``. The yt-dlp class is injectable for tests."""
|
||||
vid = str(video_id or "").strip()
|
||||
if not vid:
|
||||
return {"ok": False, "dest_path": None, "error": "No video id"}
|
||||
factory = ydl_factory or (yt_dlp.YoutubeDL if yt_dlp else None)
|
||||
if factory is None:
|
||||
return {"ok": False, "dest_path": None, "error": "yt-dlp unavailable"}
|
||||
opts = ydl_download_opts(profile, dest_dir, dest_stem,
|
||||
progress_hook=progress_hook, cookie_opts=cookie_opts)
|
||||
url = vid if vid.startswith("http") else "https://www.youtube.com/watch?v=" + vid
|
||||
try:
|
||||
with factory(opts) as ydl:
|
||||
ydl.download([url])
|
||||
except Exception as e: # noqa: BLE001 - any yt-dlp failure → a failed download, not a crash
|
||||
logger.info("youtube download failed for %s: %s", vid, e)
|
||||
return {"ok": False, "dest_path": None, "error": str(e)}
|
||||
dest_path = os.path.join(str(dest_dir or ""), dest_stem + "." + str(container or "mp4").lstrip("."))
|
||||
return {"ok": True, "dest_path": dest_path, "error": None}
|
||||
|
||||
|
||||
def process_youtube_download(
|
||||
dl: Dict[str, Any],
|
||||
*,
|
||||
profile: Any,
|
||||
settings: Dict[str, Any],
|
||||
download: Callable = download_one,
|
||||
update_row: Callable[..., Any],
|
||||
archive: Callable[[Dict[str, Any], Dict[str, Any]], Any],
|
||||
clear_wishlist: Callable[[Any], Any],
|
||||
progress_hook: Optional[Callable] = None,
|
||||
cookie_opts: Optional[dict] = None,
|
||||
now: Optional[Callable[[], str]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Fulfil one queued YouTube download. PURE — all I/O injected.
|
||||
|
||||
On success: row → completed (with dest_path), snapshot to history, and remove the video
|
||||
from the wishlist (it's in the library now; history is the permanent record). On
|
||||
failure: row → failed + a history snapshot; the wishlist row is LEFT so a later scan/run
|
||||
can retry. Returns a small result dict."""
|
||||
now = now or (lambda: "")
|
||||
settings = settings if isinstance(settings, dict) else {}
|
||||
container = format_selection(profile)["merge_output_format"]
|
||||
dest = plan_destination(dl, settings, container)
|
||||
stem, cont = _stem_and_container(dest, container)
|
||||
|
||||
# Record where it's going up front (Downloads page shows the organised name).
|
||||
update_row(dl.get("id"), status="downloading", progress=0,
|
||||
target_dir=dest.get("dir"), filename=dest.get("filename"))
|
||||
|
||||
res = download(dl.get("media_id"), dest.get("dir"), stem, profile, cont,
|
||||
progress_hook=progress_hook, cookie_opts=cookie_opts)
|
||||
|
||||
if res.get("ok"):
|
||||
dest_path = res.get("dest_path") or dest.get("path")
|
||||
completed = now()
|
||||
update_row(dl.get("id"), status="completed", progress=100,
|
||||
dest_path=dest_path, completed_at=completed)
|
||||
archive(dl, {"status": "completed", "dest_path": dest_path, "completed_at": completed})
|
||||
try:
|
||||
clear_wishlist(dl.get("media_id"))
|
||||
except Exception: # noqa: BLE001 - unwish is best-effort; the file is already in place
|
||||
logger.exception("youtube download %s: unwish failed", dl.get("id"))
|
||||
return {"status": "completed", "dest_path": dest_path}
|
||||
|
||||
err = res.get("error") or "Download failed"
|
||||
completed = now()
|
||||
update_row(dl.get("id"), status="failed", error=err, completed_at=completed)
|
||||
archive(dl, {"status": "failed", "error": err, "completed_at": completed})
|
||||
return {"status": "failed", "error": err}
|
||||
|
||||
|
||||
# ── production wiring ─────────────────────────────────────────────────────────
|
||||
def run_youtube_download(dl_id: Any, db_provider: Callable) -> None:
|
||||
"""Production entry: fetch the row, bind real seams, fulfil it. Called on a worker
|
||||
thread by the enqueue path; swallows nothing silently into the DB."""
|
||||
db = db_provider()
|
||||
dl = db.get_video_download(dl_id)
|
||||
if not dl:
|
||||
return
|
||||
profile = youtube_quality.load(db)
|
||||
settings = organization.load(db)
|
||||
from datetime import datetime, timezone
|
||||
|
||||
def _now():
|
||||
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||||
|
||||
def _progress(d):
|
||||
# yt-dlp progress hook → row progress %. Best-effort; never raises into yt-dlp.
|
||||
try:
|
||||
if d.get("status") == "downloading":
|
||||
total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
|
||||
got = d.get("downloaded_bytes") or 0
|
||||
if total:
|
||||
db.update_video_download(dl_id, progress=int(got * 100 / total))
|
||||
except Exception: # noqa: BLE001, S110 - a progress glitch must not abort the download
|
||||
pass
|
||||
|
||||
def _archive(row, upd):
|
||||
try:
|
||||
db.record_download_history({**row, **upd})
|
||||
except Exception:
|
||||
logger.exception("youtube download %s: history snapshot failed", dl_id)
|
||||
|
||||
cookie_opts = None
|
||||
try:
|
||||
from core.video.youtube import _cookie_opts
|
||||
cookie_opts = _cookie_opts()
|
||||
except Exception: # noqa: BLE001 - cookies are optional
|
||||
cookie_opts = None
|
||||
|
||||
process_youtube_download(
|
||||
dl, profile=profile, settings=settings,
|
||||
update_row=db.update_video_download, archive=_archive,
|
||||
clear_wishlist=lambda vid: db.remove_youtube_from_wishlist("video", vid),
|
||||
progress_hook=_progress, cookie_opts=cookie_opts, now=_now)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"youtube_fields_from_download", "plan_destination", "ydl_download_opts",
|
||||
"download_one", "process_youtube_download", "run_youtube_download",
|
||||
]
|
||||
153
tests/test_youtube_download.py
Normal file
153
tests/test_youtube_download.py
Normal file
|
|
@ -0,0 +1,153 @@
|
|||
"""YouTube download worker — the fulfillment lane for wished YouTube videos. Pure
|
||||
orchestration (dest planning, yt-dlp opts, completion → archive + unwish, failure →
|
||||
archive only) tested with the yt-dlp run + all DB writes injected.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
from core.video import youtube_download as ytd
|
||||
from core.video.youtube_quality import default_profile
|
||||
|
||||
|
||||
# ── organising fields from the queue row ──────────────────────────────────────
|
||||
def test_fields_prefer_search_ctx_then_fall_back():
|
||||
dl = {"title": "Some Channel", "year": "2024-01-01", "media_id": "vid1",
|
||||
"search_ctx": json.dumps({"channel": "Veritasium", "video_title": "Electricity",
|
||||
"published_at": "2024-03-15"})}
|
||||
f = ytd.youtube_fields_from_download(dl)
|
||||
assert f == {"channel": "Veritasium", "title": "Electricity",
|
||||
"published_at": "2024-03-15", "youtube_id": "vid1"}
|
||||
|
||||
|
||||
def test_fields_fall_back_to_row_when_ctx_absent_or_garbage():
|
||||
dl = {"title": "Chan", "year": "2024-02-02", "media_id": "v2", "search_ctx": "{bad"}
|
||||
f = ytd.youtube_fields_from_download(dl)
|
||||
assert f["channel"] == "Chan" and f["title"] == "Chan"
|
||||
assert f["published_at"] == "2024-02-02" and f["youtube_id"] == "v2"
|
||||
|
||||
|
||||
# ── destination planning ──────────────────────────────────────────────────────
|
||||
def test_plan_destination_uses_the_youtube_template():
|
||||
dl = {"target_dir": "/yt", "media_id": "v1",
|
||||
"search_ctx": json.dumps({"channel": "Veritasium", "video_title": "How It Works",
|
||||
"published_at": "2024-03-15"})}
|
||||
dest = ytd.plan_destination(dl, {}, "mp4")
|
||||
assert dest["path"] == os.path.join("/yt", "Veritasium", "Season 2024",
|
||||
"Veritasium - 2024-03-15 - How It Works.mp4")
|
||||
|
||||
|
||||
# ── yt-dlp opts ───────────────────────────────────────────────────────────────
|
||||
def test_ydl_opts_carry_format_selection_and_fixed_output():
|
||||
opts = ytd.ydl_download_opts(default_profile(), "/yt/dir", "Chan - 2024-03-15 - Title")
|
||||
assert opts["format"] == "bv*[height<=1080]+ba/b[height<=1080]/bv*+ba/b"
|
||||
assert opts["merge_output_format"] == "mp4"
|
||||
assert opts["paths"] == {"home": "/yt/dir"}
|
||||
assert opts["outtmpl"] == "Chan - 2024-03-15 - Title.%(ext)s"
|
||||
assert opts["noplaylist"] is True
|
||||
|
||||
|
||||
# ── download_one with an injected yt-dlp ───────────────────────────────────────
|
||||
class _FakeYDL:
|
||||
def __init__(self, opts):
|
||||
self.opts = opts
|
||||
_FakeYDL.last = self
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def download(self, urls):
|
||||
self.urls = urls
|
||||
|
||||
|
||||
class _BoomYDL(_FakeYDL):
|
||||
def download(self, urls):
|
||||
raise RuntimeError("403 blocked")
|
||||
|
||||
|
||||
def test_download_one_success_returns_built_dest_path():
|
||||
res = ytd.download_one("vid1", "/yt/Chan/Season 2024", "Chan - 2024-03-15 - T",
|
||||
default_profile(), "mp4", ydl_factory=_FakeYDL)
|
||||
assert res["ok"] is True
|
||||
assert res["dest_path"] == os.path.join("/yt/Chan/Season 2024", "Chan - 2024-03-15 - T.mp4")
|
||||
assert _FakeYDL.last.urls == ["https://www.youtube.com/watch?v=vid1"]
|
||||
|
||||
|
||||
def test_download_one_failure_is_captured_not_raised():
|
||||
res = ytd.download_one("vid1", "/yt", "stem", default_profile(), "mp4", ydl_factory=_BoomYDL)
|
||||
assert res["ok"] is False and "403 blocked" in res["error"]
|
||||
|
||||
|
||||
def test_download_one_no_factory_is_unavailable():
|
||||
res = ytd.download_one("vid1", "/yt", "stem", default_profile(), "mp4", ydl_factory=None)
|
||||
# yt_dlp may or may not be importable in the test env; either way no real run happens
|
||||
if res["ok"] is False:
|
||||
assert res["error"]
|
||||
|
||||
|
||||
# ── the orchestration: completion vs failure ──────────────────────────────────
|
||||
def _recorder():
|
||||
calls = {"rows": [], "archive": [], "unwish": []}
|
||||
|
||||
def update_row(dl_id, **kw):
|
||||
calls["rows"].append((dl_id, kw))
|
||||
|
||||
def archive(row, upd):
|
||||
calls["archive"].append(upd)
|
||||
|
||||
def clear_wishlist(vid):
|
||||
calls["unwish"].append(vid)
|
||||
|
||||
return calls, update_row, archive, clear_wishlist
|
||||
|
||||
|
||||
def _dl():
|
||||
return {"id": 7, "media_id": "vid1", "target_dir": "/yt", "title": "Chan", "year": "2024-03-15",
|
||||
"search_ctx": json.dumps({"channel": "Chan", "video_title": "T", "published_at": "2024-03-15"})}
|
||||
|
||||
|
||||
def test_process_completion_archives_and_unwishes():
|
||||
calls, update_row, archive, clear = _recorder()
|
||||
res = ytd.process_youtube_download(
|
||||
_dl(), profile=default_profile(), settings={},
|
||||
download=lambda *a, **k: {"ok": True, "dest_path": "/yt/Chan/Season 2024/Chan - 2024-03-15 - T.mp4"},
|
||||
update_row=update_row, archive=archive, clear_wishlist=clear, now=lambda: "2026-06-25T00:00:00+00:00")
|
||||
assert res["status"] == "completed"
|
||||
# row marked completed, history snapshot 'completed', and the video unwished
|
||||
statuses = [kw.get("status") for _, kw in calls["rows"]]
|
||||
assert "downloading" in statuses and statuses[-1] == "completed"
|
||||
assert calls["archive"][-1]["status"] == "completed"
|
||||
assert calls["unwish"] == ["vid1"]
|
||||
|
||||
|
||||
def test_process_failure_archives_but_keeps_the_wish():
|
||||
calls, update_row, archive, clear = _recorder()
|
||||
res = ytd.process_youtube_download(
|
||||
_dl(), profile=default_profile(), settings={},
|
||||
download=lambda *a, **k: {"ok": False, "error": "yt-dlp said no"},
|
||||
update_row=update_row, archive=archive, clear_wishlist=clear, now=lambda: "t")
|
||||
assert res["status"] == "failed" and "yt-dlp said no" in res["error"]
|
||||
assert calls["rows"][-1][1]["status"] == "failed"
|
||||
assert calls["archive"][-1]["status"] == "failed"
|
||||
assert calls["unwish"] == [] # wish kept so it can retry later
|
||||
|
||||
|
||||
def test_process_passes_the_organised_dir_to_the_downloader():
|
||||
seen = {}
|
||||
|
||||
def fake_download(video_id, dest_dir, stem, profile, container, **kw):
|
||||
seen.update(video_id=video_id, dest_dir=dest_dir, stem=stem, container=container)
|
||||
return {"ok": True, "dest_path": "/x"}
|
||||
|
||||
calls, update_row, archive, clear = _recorder()
|
||||
ytd.process_youtube_download(_dl(), profile=default_profile(), settings={},
|
||||
download=fake_download, update_row=update_row,
|
||||
archive=archive, clear_wishlist=clear, now=lambda: "t")
|
||||
assert seen["video_id"] == "vid1"
|
||||
assert seen["dest_dir"] == os.path.join("/yt", "Chan", "Season 2024")
|
||||
assert seen["stem"] == "Chan - 2024-03-15 - T" and seen["container"] == "mp4"
|
||||
Loading…
Reference in a new issue