zebra
f103b6c253
Generalize URL handling beyond YouTube to any yt-dlp-supported site (SoundCloud, Bandcamp, etc), single tracks and playlists/sets/albums. - probe_url(): one yt-dlp --flat-playlist probe classifies playlist vs track and returns per-entry Hits; YouTube playlists still use ytmusicapi. - _track_url(): YouTube tracks keep the music.youtube album-art URL; other platforms download via their native entry URL (no more videoId reconstruction). - Per-source folders: <root>/<artist>/<extractor>/ (soundcloud/bandcamp/youtube) instead of hardcoded youtube; download_single derives source from metadata. - download_hits() downloads pre-probed Hits; API probes once and passes hits into the job closure. Replaces YouTube-only is_playlist_url/expand_playlist. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
99 lines
3.9 KiB
Python
99 lines
3.9 KiB
Python
"""Glue between a chosen Hit and a side-effecting download. Mirrors musicfetch's
|
|
main() dispatch but returns a structured result dict and speakable messages."""
|
|
import os
|
|
|
|
from . import mf
|
|
|
|
|
|
def _source_label(hit) -> str:
|
|
return "YouTube Music" if hit.source == "youtube" else "Lidarr"
|
|
|
|
|
|
def _title(hit) -> str:
|
|
return hit.album if hit.kind == "album" else (hit.title or hit.album or hit.artist)
|
|
|
|
|
|
def _primary_artist(hit) -> str:
|
|
"""First artist only — ignore featured/secondary artists."""
|
|
return (hit.artist.split(",")[0].strip() if hit.artist else "") or "unknown artist"
|
|
|
|
|
|
def started_message(hit) -> str:
|
|
return f"Found '{_title(hit)}' by {_primary_artist(hit)} on {_source_label(hit)}. Downloading now."
|
|
|
|
|
|
def done_message(hit) -> str:
|
|
return f"Finished downloading '{_title(hit)}' by {_primary_artist(hit)}."
|
|
|
|
|
|
def failed_message(hit) -> str:
|
|
return f"Failed to download '{_title(hit)}' by {_primary_artist(hit)}."
|
|
|
|
|
|
def _yt_path(hit, root: str) -> str:
|
|
artist_dir = (hit.artist.split(",")[0].strip() if hit.artist else "") or "Unknown Artist"
|
|
return os.path.join(root, artist_dir, "youtube")
|
|
|
|
|
|
def _download_youtube(hit, quality: str, root: str) -> dict:
|
|
mf.act_youtube(hit, root, quality, False)
|
|
return {"path": _yt_path(hit, root), "lidarr_album_id": None}
|
|
|
|
|
|
def perform_fetch(chosen, hits: list, quality: str, root: str) -> dict:
|
|
"""Run the download for the chosen hit. Returns {"path", "lidarr_album_id"}.
|
|
Raises on unrecoverable failure (recorded by the job worker)."""
|
|
if chosen.source == "youtube":
|
|
return _download_youtube(chosen, quality, root)
|
|
|
|
if chosen.kind == "album":
|
|
handled = mf.act_lidarr_album(chosen, root, False, False)
|
|
if handled:
|
|
return {"path": None, "lidarr_album_id": chosen.payload.get("album", {}).get("id")}
|
|
# No indexer release -> fall through to the top YouTube hit, like the CLI.
|
|
yt = next((h for h in hits if h.source == "youtube"), None)
|
|
if yt is None:
|
|
raise RuntimeError("No Lidarr release and no YouTube fallback available.")
|
|
return _download_youtube(yt, quality, root)
|
|
|
|
# Lidarr artist pick.
|
|
ok = mf.act_lidarr_artist(chosen, root, False, False)
|
|
if not ok:
|
|
raise RuntimeError("Failed to add artist to Lidarr.")
|
|
return {"path": None, "lidarr_album_id": None}
|
|
|
|
|
|
def url_started_message(kind: str, title: str = "") -> str:
|
|
if kind == "playlist":
|
|
return (f"Fetching playlist '{title}'. Downloading tracks now."
|
|
if title else "Fetching playlist. Downloading tracks now.")
|
|
return f"Fetching '{title}'. Downloading now." if title else "Fetching track. Downloading now."
|
|
|
|
|
|
def playlist_done_message(result: dict) -> str:
|
|
ok, total = result.get("ok", 0), result.get("total", 0)
|
|
failed = total - ok
|
|
return f"Downloaded {ok}/{total} tracks" + (f" ({failed} failed)." if failed else ".")
|
|
|
|
|
|
def url_done_message(result: dict) -> str:
|
|
title = result.get("title", "")
|
|
return f"Downloaded '{title}'." if title else "Download complete."
|
|
|
|
|
|
def perform_url_fetch(url: str, kind: str, title: str, hits: list, quality: str, root: str) -> dict:
|
|
"""Download a probed URL (playlist -> batch over pre-probed hits, else single).
|
|
Raises if nothing downloaded so the job is marked failed."""
|
|
if kind == "playlist":
|
|
ok, total = mf.download_hits(hits, root, quality, False)
|
|
if ok == 0:
|
|
raise RuntimeError(f"No tracks downloaded from playlist '{title}'." if title
|
|
else "No tracks downloaded from playlist.")
|
|
return {"kind": "playlist", "title": title, "ok": ok, "total": total,
|
|
"path": None, "lidarr_album_id": None}
|
|
info = mf.download_single(url, root, quality, False)
|
|
if not info.get("ok"):
|
|
raise RuntimeError("Download failed.")
|
|
return {"kind": "track", "title": info.get("title", ""), "artist": info.get("artist", ""),
|
|
"ok": 1, "total": 1, "path": None, "lidarr_album_id": None}
|