236 lines
9.9 KiB
Python
236 lines
9.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"
|
|
|
|
|
|
MAX_CHOICES = int(os.environ.get("MUSICFETCH_MAX_CHOICES", "10"))
|
|
|
|
|
|
class FetchError(RuntimeError):
|
|
"""A failure whose text is written for a person to hear. Anything else
|
|
fails with the job's generic message, because a yt-dlp stack trace is not
|
|
something Siri should read out."""
|
|
|
|
|
|
def started_message(hit) -> str:
|
|
"""What the server is about to do — never a claim that it finished."""
|
|
artist = _primary_artist(hit)
|
|
if hit.source == "lidarr":
|
|
if hit.kind == "artist":
|
|
return f"Looking up {artist} in Lidarr."
|
|
return (f"Found the album '{_title(hit)}' by {artist} in Lidarr. "
|
|
f"Searching indexers now.")
|
|
return f"Found '{_title(hit)}' by {artist} on {_source_label(hit)}. Downloading now."
|
|
|
|
|
|
def _artist_line(state: str, artist: str) -> str:
|
|
if state == "added":
|
|
return f"Added {artist} to Lidarr."
|
|
if state == "existing":
|
|
return f"{artist} is already in Lidarr."
|
|
return ""
|
|
|
|
|
|
def done_message(hit, result: "dict | None" = None) -> str:
|
|
"""Report what really happened. An artist add is not a download, and a
|
|
Lidarr grab is a hand-off to the download client, not a finished file."""
|
|
artist, title = _primary_artist(hit), _title(hit)
|
|
outcome = (result or {}).get("outcome", "")
|
|
|
|
if outcome == "artist_added":
|
|
return (f"Added {artist} to Lidarr. Lidarr now monitors their albums. "
|
|
f"Nothing was downloaded.")
|
|
if outcome == "artist_existing":
|
|
return f"{artist} is already in Lidarr. Nothing was added."
|
|
if outcome == "album_grabbed":
|
|
return grab_message(result or {}, artist=artist, album=title)
|
|
if outcome == "no_release":
|
|
lead = _artist_line(result.get("artist_state", ""), artist)
|
|
return (f"{lead} '{title}' is monitored in Lidarr, but no indexer has a "
|
|
f"release for it right now.").strip()
|
|
if outcome == "youtube_fallback":
|
|
return (f"Lidarr could not get '{title}' by {artist}. "
|
|
f"Downloaded it from YouTube Music instead.")
|
|
return f"Finished downloading '{title}' by {artist}."
|
|
|
|
|
|
def failed_message(hit) -> str:
|
|
artist, title = _primary_artist(hit), _title(hit)
|
|
if hit.source == "lidarr" and hit.kind == "artist":
|
|
return f"Could not add {artist} to Lidarr."
|
|
return f"Failed to fetch '{title}' by {artist}."
|
|
|
|
|
|
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, outcome: str = "youtube") -> dict:
|
|
mf.act_youtube(hit, root, quality, False)
|
|
return {"outcome": outcome, "path": _yt_path(hit, root), "lidarr_album_id": None}
|
|
|
|
|
|
def perform_fetch(chosen, hits: list, quality: str, root: str,
|
|
auto: bool = False, allow_youtube: bool = True) -> dict:
|
|
"""Act on the chosen hit. Returns a result dict; a non-empty "choices" list
|
|
parks the job so the client can pick a release. Raises on unrecoverable
|
|
failure (recorded by the job worker).
|
|
|
|
`auto=True` grabs the best release and never asks. A URL always uses it.
|
|
"""
|
|
if chosen.source == "youtube":
|
|
return _download_youtube(chosen, quality, root)
|
|
|
|
if chosen.kind == "artist":
|
|
# ensure_artist never adds a duplicate.
|
|
res = mf.ensure_artist(chosen.payload["artist"], root, False, False)
|
|
if res["state"] == "failed":
|
|
raise FetchError(f"Could not add {res['name']} to Lidarr.")
|
|
return {"outcome": f"artist_{res['state']}", "artist": res["name"],
|
|
"path": None, "lidarr_album_id": (res["artist"] or {}).get("id")}
|
|
|
|
res = mf.ensure_album(chosen.payload["album"], root, False, False)
|
|
artist, album = res["artist_name"], res["album_title"]
|
|
if res["album"] is None:
|
|
if res["artist_state"] == "failed":
|
|
raise FetchError(f"Could not add {artist} to Lidarr.")
|
|
raise FetchError(f"{artist} is in Lidarr, but it has no album called '{album}'.")
|
|
|
|
base = {"artist": artist, "album": album, "path": None,
|
|
"lidarr_album_id": res["album_id"],
|
|
"artist_state": res["artist_state"], "album_state": res["album_state"]}
|
|
|
|
releases = mf.list_releases(res["album_id"])
|
|
if not releases:
|
|
# Nothing on any indexer -> the top YouTube hit, like the CLI.
|
|
yt = next((h for h in hits if h.source == "youtube"), None) if allow_youtube else None
|
|
if yt is not None:
|
|
return {**base, **_download_youtube(yt, quality, root, outcome="youtube_fallback")}
|
|
return {**base, "outcome": "no_release"}
|
|
|
|
if auto:
|
|
return {**base, **_grab(releases[0], artist, album)}
|
|
|
|
releases = releases[:MAX_CHOICES]
|
|
return {**base, "outcome": "choices",
|
|
"choices": [_choice(i, r) for i, r in enumerate(releases, 1)],
|
|
"pending": releases}
|
|
|
|
|
|
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}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lidarr find + choose (the two-step Siri Shortcuts flow)
|
|
# ---------------------------------------------------------------------------
|
|
def _choice(n: int, release: dict) -> dict:
|
|
"""The client-facing view of one indexer release."""
|
|
quality = ((release.get("quality") or {}).get("quality") or {}).get("name") or ""
|
|
size = release.get("size") or 0
|
|
label = mf.release_label(release)
|
|
return {
|
|
"n": n,
|
|
"label": label,
|
|
# "1. FLAC, 312 MB, ..." — a Siri Shortcut can show this in a list and
|
|
# read the number back off the item the user tapped.
|
|
"pick_label": f"{n}. {label}",
|
|
"title": release.get("title", ""),
|
|
"quality": quality,
|
|
"size_mb": round(float(size) / (1024 * 1024)) if size else None,
|
|
"seeders": release.get("seeders"),
|
|
"indexer": release.get("indexer", ""),
|
|
"rejected": bool(release.get("rejected")),
|
|
}
|
|
|
|
|
|
def choice_message(hit, result: dict) -> str:
|
|
"""Asked when the client must pick a release."""
|
|
artist = result.get("artist", "") or _primary_artist(hit)
|
|
album = result.get("album", "") or _title(hit)
|
|
lead = _artist_line(result.get("artist_state", ""), artist)
|
|
if result.get("album_state") == "monitored":
|
|
lead = f"{lead} Now monitoring '{album}'.".strip()
|
|
n = len(result.get("choices") or [])
|
|
ask = (f"I found {n} release{'s' if n != 1 else ''} for "
|
|
f"'{album}' by {artist}. Which one?")
|
|
return f"{lead} {ask}".strip()
|
|
|
|
|
|
def _grab(release: dict, artist: str, album: str) -> dict:
|
|
if not mf.grab_release(release, False):
|
|
raise FetchError(f"Lidarr refused the release for '{album}' by {artist}.")
|
|
return {"outcome": "album_grabbed", "release": mf.release_label(release),
|
|
"release_title": release.get("title", ""), "artist": artist, "album": album}
|
|
|
|
|
|
def grab_started_message(artist: str, album: str) -> str:
|
|
return f"Sending '{album}' by {artist} to the download client."
|
|
|
|
|
|
def grab_message(result: dict, artist: str = "", album: str = "") -> str:
|
|
"""Names the release when the result carries one. The artist/album
|
|
arguments are a fallback for a result that does not name them."""
|
|
who = result.get("artist") or artist
|
|
what = result.get("album") or album
|
|
base = f"Sent '{what}' by {who} to the download client."
|
|
label = result.get("release", "")
|
|
return f"{base} {label}." if label else base
|
|
|
|
|
|
def perform_grab(release: dict, artist: str, album: str) -> dict:
|
|
"""Push one exact release to the download client (used by POST /lidarr/choose)."""
|
|
return _grab(release, artist, album)
|
|
|
|
|
|
def fail_text(hit, exc: Exception) -> str:
|
|
"""A FetchError already reads like a sentence; anything else does not."""
|
|
return str(exc) if isinstance(exc, FetchError) and str(exc) else failed_message(hit)
|