From 9fd4c8585b56a0de7047d919cda460345e1a3fa0 Mon Sep 17 00:00:00 2001 From: zebra Date: Mon, 8 Jun 2026 19:30:47 -0700 Subject: [PATCH] Rewrite musicfetch as v2: dual-source search with interactive picker Accept free-form queries (artist/album/title and combos) instead of strict "Artist - Track". Search Lidarr and YouTube Music concurrently, present a unified rich picker with bold keyword matching, and act on the chosen hit. - Normalize results via a Hit dataclass across both sources - Lidarr: /api/v1/search (album+artist) with album/artist lookup fallback - YouTube: ytmusicapi for accurate metadata + music.youtube.com URLs, yt-dlp scrape fallback; tag overrides from the chosen hit - Lidarr album pick adds artist+album monitored, runs interactive release search, and falls through to top YouTube hit when no indexer release exists - argparse CLI: -n/--noninteractive, -s/--ytsearch, -d/--dry-run, -q/--quality, --limit, --lidarr-only/--yt-only, -o/--root, --search-all - Config via LIDARR_URL / LIDARR_API_KEY / MUSICFETCH_ROOT env vars - Update README; add .gitignore for __pycache__ Co-Authored-By: Claude Opus 4.8 --- .gitignore | 2 + README.md | 147 +++++---- musicfetch | 888 ++++++++++++++++++++++++++++++++++------------------- 3 files changed, 661 insertions(+), 376 deletions(-) create mode 100644 .gitignore mode change 100644 => 100755 musicfetch diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7a60b85 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/README.md b/README.md index c320068..a9c467f 100644 --- a/README.md +++ b/README.md @@ -1,132 +1,149 @@ # 🎵 MusicFetch -**MusicFetch** is a smart command-line utility that fetches music by querying Lidarr (a music collection manager) or, if no match is found or a timeout occurs, falls back to downloading using `yt-dlp`. It supports input in the form of either: +**MusicFetch** is a smart command-line utility that finds music by searching +**Lidarr** (your music collection manager) and **YouTube Music** at the same +time, shows you the top hits in an interactive picker, and downloads/queues +whatever you choose. It accepts: -- `"Artist - Track"` (e.g. `"Daft Punk - Harder Better Faster Stronger"`) -- A full YouTube URL (e.g. `"https://www.youtube.com/watch?v=dQw4w9WgXcQ"`) +- A **free-form query**: an artist, an album, a track title, or combos like + `"Artist - Title"` or `"Artist - Album"` (e.g. `"ODESZA - Bloom"`, `"Daft Punk"`, `"Discovery"`). +- A **URL** (e.g. `"https://music.youtube.com/watch?v=..."` or a regular YouTube URL). -The downloaded track is organized into your Lidarr media folder or a YouTube subfolder if it's a fallback. +Lidarr is tried first by default. If you pick a Lidarr album but **no indexer +release is available**, MusicFetch automatically falls through to the top +YouTube hit. YouTube downloads prefer **YouTube Music URLs** so album art and +tags are correct. --- ## 🚀 Features -- Searches for and adds artists to Lidarr. -- Automatically triggers album searches in Lidarr. -- Falls back to downloading via `yt-dlp` if no match is found. -- Supports metadata extraction from YouTube URLs. -- Handles Lidarr timeouts gracefully. +- One unified picker showing the top hits from **Lidarr and YouTube together**, with matching keywords **bolded**. +- Lidarr-first flow: pick an album → adds artist+album (monitored) → interactive indexer search → falls through to YouTube only if no release is found. +- Accurate YouTube metadata via `ytmusicapi` (real artist / album / year / album art), with `yt-dlp` scraping as a fallback. +- Explicit tag overrides on download so files are tagged from the chosen hit, not from scraped titles. +- Non-interactive, YouTube-first, dry-run, quality, limit, and source-restriction flags. --- ## 📦 Dependencies & Installation -### 🐍 Python Dependencies +### 🐍 Python -- Python 3.6+ +- Python 3.10+ - `requests` - -Install Python dependencies: +- `ytmusicapi` (recommended — accurate YouTube Music metadata) +- `rich` (recommended — nicer picker table + bold keyword matching) ```bash -pip install requests +pip install requests ytmusicapi rich ``` +> **Note:** if you hit `ModuleNotFoundError: No module named 'idna'` from +> `requests`, repair it with: +> ```bash +> pip install --force-reinstall idna requests +> ``` + +`ytmusicapi` and `rich` are optional — without them MusicFetch falls back to +`yt-dlp` search scraping and a plain ANSI picker. + ### 📼 External Tools -- `yt-dlp`: YouTube downloader for audio/video. - -Install via pip (recommended): +- `yt-dlp` (audio download/extraction) and `ffmpeg` (for `-x` extraction / embedding). ```bash pip install -U yt-dlp -``` - -Or install system-wide (Debian/Ubuntu): - -```bash -sudo apt install yt-dlp +sudo apt install ffmpeg # or your distro's equivalent ``` --- ## ⚙️ Configuration -### Lidarr Setup +Set via environment variables: -- Ensure Lidarr is running and accessible (e.g., [http://localhost:8686](http://localhost:8686)). -- Create or retrieve your API key from Lidarr's settings. - -Set your API key in the terminal session: +| Variable | Default | Purpose | +|-------------------|-------------------------|----------------------------------| +| `LIDARR_API_KEY` | *(required for Lidarr)* | Lidarr API key. | +| `LIDARR_URL` | `http://localhost:8686` | Lidarr base URL. | +| `MUSICFETCH_ROOT` | `/media/music` | Default output root folder. | ```bash export LIDARR_API_KEY="your-lidarr-api-key" ``` -Or run inline: - -```bash -LIDARR_API_KEY="your-lidarr-api-key" ./musicfetch "Artist - Track" -``` - --- ## 🧑‍💻 Usage -### 🔉 Download by Search Term - ```bash -./musicfetch "Artist - Track" +./musicfetch [OPTIONS] QUERY... ``` -Example: +### Options + +| Flag | Description | +|------|-------------| +| `-n`, `--noninteractive` | Auto-pick the top hit (no prompt). | +| `-s`, `--ytsearch` | Search/select YouTube first instead of Lidarr first. | +| `-d`, `--dry-run` | Show every action without executing it. | +| `-q`, `--quality {best,320,m4a,opus,flac}` | Audio quality/format (default `best`). | +| `--limit N` | Hits per source (default 10). | +| `--lidarr-only` | Skip YouTube. | +| `--yt-only` | Skip Lidarr. | +| `-o`, `--root PATH` | Output root folder (default `/media/music`). | +| `--search-all` | Search all albums when adding an artist to Lidarr. | +| `--debug` | Verbose output. | + +### Examples ```bash +# Interactive: combined Lidarr + YouTube picker ./musicfetch "ODESZA - Bloom" + +# Just an artist / just an album / just a title all work +./musicfetch "Daft Punk" +./musicfetch "Discovery" + +# YouTube first, auto-pick top hit +./musicfetch -s -n "Daft Punk - Harder Better Faster Stronger" + +# Dry run — see what would happen, change nothing +./musicfetch -d "ODESZA - Bloom" + +# YouTube only, lossless preferred +./musicfetch --yt-only -q flac "Bonobo - Kerala" + +# Download by URL (YouTube Music URL preferred for correct art) +./musicfetch "https://music.youtube.com/watch?v=xxxxxxxxxxx" ``` -If Lidarr finds the artist and album, it will trigger a download in Lidarr. If not, the song will be downloaded via yt-dlp using a YouTube search. - -### 📺 Download by URL - -```bash -./musicfetch "https://www.youtube.com/watch?v=xxxxxxxxxxx" -``` - -The script extracts metadata from the video and organizes the file under the artist's folder. - ### 📁 Output Structure -Music is saved in: - ```text -/base_dir/ +/ ├── Artist Name/ -│ ├── Album Name/ (if found via Lidarr) -│ └── youtube/ (if fallback used) +│ ├── Album Name/ (managed by Lidarr) +│ └── youtube/ (yt-dlp downloads / fallbacks) ``` --- ## ❓ Troubleshooting -- **No results from Lidarr:** - - Check to make sure your Lidarr installation is reachable and the artist exists or can be found in Lidarr's metadata sources. -- **yt-dlp errors:** - - Try updating it: - - ```bash - yt-dlp -U - ``` - -- **Permission denied or file not found?** - - Ensure `/media/music` exists and is writable. +- **No Lidarr hits / "LIDARR_API_KEY not set":** export your key and confirm `LIDARR_URL` is reachable. +- **Wrong album art from YouTube:** install `ytmusicapi` so MusicFetch can resolve proper YouTube Music URLs and metadata. +- **`yt-dlp` errors:** update with `yt-dlp -U`; ensure `ffmpeg` is installed for extraction/embedding. +- **`idna` import error:** `pip install --force-reinstall idna requests`. +- **Permission denied writing files:** ensure the output root exists and is writable (`-o`/`--root` or `MUSICFETCH_ROOT`). --- ## 🛠️ Contributing -PRs are welcome! Please make your changes easy to follow. This script is designed to act as middleware, not a replacement or plugin for Lidarr. Changes must also be compatible with Bash-based workflows. +PRs welcome. This script is middleware around Lidarr + yt-dlp, not a Lidarr +replacement. Keep it a single bash-friendly executable. --- diff --git a/musicfetch b/musicfetch old mode 100644 new mode 100755 index b8a8f13..e8cae20 --- a/musicfetch +++ b/musicfetch @@ -1,351 +1,617 @@ #!/usr/bin/env python3 +"""MusicFetch v2 — fetch music via Lidarr (preferred) or YouTube Music (yt-dlp). + +Accepts a free-form query ("artist", "title", "album", or combos like +"artist - title" / "artist - album") or a URL. Searches Lidarr and YouTube +Music concurrently, shows the top hits in an interactive picker, and acts on +the chosen hit. See README.md for full docs. +""" +import argparse import json -import sys import os import re import subprocess +import sys +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from typing import Optional + import requests -from requests.exceptions import Timeout +from requests.exceptions import RequestException, Timeout + +# Optional deps — degrade gracefully if missing. +try: + from ytmusicapi import YTMusic +except ImportError: + YTMusic = None + +try: + from rich.console import Console + from rich.table import Table + from rich.text import Text + _console = Console() +except ImportError: + Console = None + _console = None # === CONFIGURATION === -LIDARR_URL = "http://localhost:8686" # Your Lidarr base URL -API_KEY = os.environ.get("LIDARR_API_KEY", "") # Your Lidarr API key -ROOT_FOLDER = "/media/music" +LIDARR_URL = os.environ.get("LIDARR_URL", "http://localhost:8686").rstrip("/") +API_KEY = os.environ.get("LIDARR_API_KEY", "") +DEFAULT_ROOT = os.environ.get("MUSICFETCH_ROOT", "/media/music") -headers = { - "X-Api-Key": API_KEY, - "Content-Type": "application/json" -} +HEADERS = {"X-Api-Key": API_KEY, "Content-Type": "application/json"} -def is_url(string): - return re.match(r'https?://', string) +# Runtime flags, populated in main(). +DEBUG = False -def run_yt_dlp_get_metadata(url): +# Quality choices for --quality. +QUALITY_CHOICES = ["best", "320", "m4a", "opus", "flac"] + + +def dbg(*a): + if DEBUG: + print("[DEBUG]", *a) + + +def err(*a): + print("[ERROR]", *a, file=sys.stderr) + + +# --------------------------------------------------------------------------- +# Hit model +# --------------------------------------------------------------------------- +@dataclass +class Hit: + source: str # "lidarr" | "youtube" + kind: str # "artist" | "album" | "track" + title: str = "" # track/album title (display) + artist: str = "" + album: str = "" + year: str = "" + thumbnail: str = "" + payload: dict = field(default_factory=dict) # raw data needed to act + + @property + def display_title(self) -> str: + return self.title or self.album or self.artist + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- +def is_url(s: str) -> bool: + return bool(re.match(r"https?://", s)) + + +def lidarr_get(path, params=None, timeout=15): + resp = requests.get(f"{LIDARR_URL}{path}", headers=HEADERS, params=params, timeout=timeout) + resp.raise_for_status() + return resp.json() + + +def lidarr_post(path, payload, timeout=15): + resp = requests.post(f"{LIDARR_URL}{path}", headers=HEADERS, json=payload, timeout=timeout) + resp.raise_for_status() + return resp.json() if resp.content else {} + + +# --------------------------------------------------------------------------- +# Lidarr search +# --------------------------------------------------------------------------- +def _year_from_album(album: dict) -> str: + rd = album.get("releaseDate") or album.get("firstReleaseDate") or "" + return rd[:4] if rd else "" + + +def _album_to_hit(album: dict) -> Hit: + artist = (album.get("artist") or {}).get("artistName") or album.get("artistName") or "" + return Hit( + source="lidarr", + kind="album", + title=album.get("title", ""), + artist=artist, + album=album.get("title", ""), + year=_year_from_album(album), + payload={"album": album}, + ) + + +def _artist_to_hit(artist: dict) -> Hit: + return Hit( + source="lidarr", + kind="artist", + title=artist.get("artistName") or artist.get("title", ""), + artist=artist.get("artistName") or artist.get("title", ""), + payload={"artist": artist}, + ) + + +def lidarr_search(query: str, limit: int) -> list[Hit]: + """Universal search via /api/v1/search; fall back to album+artist lookup.""" + if not API_KEY: + err("LIDARR_API_KEY not set — skipping Lidarr search.") + return [] + hits: list[Hit] = [] + try: + results = lidarr_get("/api/v1/search", params={"term": query}) + for item in results: + # /search returns objects with 'foreignId' and either 'album' or 'artist'. + if item.get("album"): + hits.append(_album_to_hit(item["album"])) + elif item.get("artist"): + hits.append(_artist_to_hit(item["artist"])) + if hits: + return hits[:limit] + dbg("/api/v1/search returned nothing useful; trying lookup endpoints.") + except Timeout: + err("Lidarr universal search timed out.") + except RequestException as e: + dbg(f"/api/v1/search unavailable ({e}); falling back to lookup endpoints.") + + # Fallback: album lookup then artist lookup. + try: + for album in lidarr_get("/api/v1/album/lookup", params={"term": query}): + hits.append(_album_to_hit(album)) + except RequestException as e: + dbg(f"album/lookup failed: {e}") + try: + for artist in lidarr_get("/api/v1/artist/lookup", params={"term": query}): + hits.append(_artist_to_hit(artist)) + except RequestException as e: + dbg(f"artist/lookup failed: {e}") + return hits[:limit] + + +# --------------------------------------------------------------------------- +# YouTube search (ytmusicapi preferred, yt-dlp scrape fallback) +# --------------------------------------------------------------------------- +def _ytm_thumb(item: dict) -> str: + thumbs = item.get("thumbnails") or [] + return thumbs[-1]["url"] if thumbs else "" + + +def _ytm_artists(item: dict) -> str: + arts = item.get("artists") or [] + return ", ".join(a.get("name", "") for a in arts if a.get("name")) + + +def youtube_search(query: str, limit: int) -> list[Hit]: + if YTMusic is not None: + try: + return _ytmusic_search(query, limit) + except Exception as e: # ytmusicapi raises broadly + dbg(f"ytmusicapi search failed ({e}); falling back to yt-dlp scrape.") + return _ytdlp_search(query, limit) + + +def _ytmusic_search(query: str, limit: int) -> list[Hit]: + yt = YTMusic() + hits: list[Hit] = [] + # Songs give us videoId + album + artist; that's the best download target. + for item in yt.search(query, filter="songs", limit=limit): + vid = item.get("videoId") + if not vid: + continue + album = (item.get("album") or {}).get("name", "") if isinstance(item.get("album"), dict) else (item.get("album") or "") + hits.append(Hit( + source="youtube", + kind="track", + title=item.get("title", ""), + artist=_ytm_artists(item), + album=album, + year=str(item.get("year") or ""), + thumbnail=_ytm_thumb(item), + payload={"videoId": vid}, + )) + if len(hits) >= limit: + break + return hits + + +def _ytdlp_search(query: str, limit: int) -> list[Hit]: + try: + result = subprocess.run( + ["yt-dlp", "--flat-playlist", "-J", f"ytsearch{limit}:{query}"], + capture_output=True, text=True, check=True, + ) + data = json.loads(result.stdout) + except (subprocess.CalledProcessError, json.JSONDecodeError) as e: + err(f"yt-dlp search failed: {e}") + return [] + hits: list[Hit] = [] + for entry in data.get("entries", []): + vid = entry.get("id") + if not vid: + continue + hits.append(Hit( + source="youtube", + kind="track", + title=entry.get("title", ""), + artist=entry.get("uploader") or entry.get("channel") or "", + year="", + thumbnail="", + payload={"videoId": vid}, + )) + return hits[:limit] + + +# --------------------------------------------------------------------------- +# Picker / rendering +# --------------------------------------------------------------------------- +def _keywords(query: str) -> list[str]: + return [w for w in re.split(r"[\s\-]+", query.lower()) if len(w) > 1] + + +def _ansi_bold_match(s: str, keywords: list[str]) -> str: + if not s: + return "" + out = s + for kw in keywords: + out = re.sub(f"({re.escape(kw)})", "\033[1m\\1\033[0m", out, flags=re.IGNORECASE) + return out + + +def _rich_match(s: str, keywords: list[str]): + text = Text(s or "") + low = (s or "").lower() + for kw in keywords: + start = 0 + while True: + idx = low.find(kw, start) + if idx == -1: + break + text.stylize("bold", idx, idx + len(kw)) + start = idx + len(kw) + return text + + +def render_picker(hits: list[Hit], query: str, yt_first: bool) -> None: + keywords = _keywords(query) + + if _console is not None: + table = Table(show_lines=False, expand=False) + table.add_column("#", justify="right", style="cyan") + table.add_column("Src") + table.add_column("Artist") + table.add_column("Album / Title") + table.add_column("Year") + table.add_column("Type") + for i, h in enumerate(hits, 1): + src = "[green]LID[/]" if h.source == "lidarr" else "[red]YT[/]" + at = h.album if h.kind == "album" else h.display_title + table.add_row( + str(i), src, + _rich_match(h.artist, keywords), + _rich_match(at, keywords), + h.year, h.kind, + ) + _console.print(table) + else: + for i, h in enumerate(hits, 1): + src = "LID" if h.source == "lidarr" else "YT " + at = h.album if h.kind == "album" else h.display_title + print(f"{i:>3} {src} {_ansi_bold_match(h.artist, keywords):<30} " + f"{_ansi_bold_match(at, keywords):<40} {h.year:<6} {h.kind}") + + +def pick(hits: list[Hit], query: str, noninteractive: bool, yt_first: bool) -> Optional[Hit]: + if not hits: + return None + if noninteractive: + primary = "youtube" if yt_first else "lidarr" + for h in hits: + if h.source == primary: + return h + return hits[0] + + render_picker(hits, query, yt_first) + while True: + try: + raw = input("Pick a number (q to quit): ").strip() + except (EOFError, KeyboardInterrupt): + print() + return None + if raw.lower() in ("q", "quit", ""): + return None + if raw.isdigit() and 1 <= int(raw) <= len(hits): + return hits[int(raw) - 1] + print("Invalid choice.") + + +# --------------------------------------------------------------------------- +# Lidarr actions +# --------------------------------------------------------------------------- +def get_existing_artist(name: str) -> Optional[dict]: + try: + for artist in lidarr_get("/api/v1/artist", timeout=10): + if artist.get("artistName", "").lower() == name.lower(): + return artist + except RequestException as e: + dbg(f"existing artist check failed: {e}") + return None + + +def get_default_metadata_profile_id() -> int: + try: + profiles = lidarr_get("/api/v1/metadataprofile", timeout=10) + if profiles: + return profiles[0]["id"] + except RequestException as e: + dbg(f"metadataprofile fetch failed: {e}") + return 1 + + +def add_artist(meta: dict, root: str, search_all: bool, dry_run: bool) -> Optional[dict]: + foreign_id = meta.get("foreignArtistId") or meta.get("id") + name = meta.get("artistName") or meta.get("title") + if not foreign_id or not name: + err("Missing foreignArtistId/artistName; cannot add artist.") + return None + payload = { + "foreignArtistId": foreign_id, + "artistName": name, + "qualityProfileId": 1, + "metadataProfileId": get_default_metadata_profile_id(), + "rootFolderPath": root, + "monitored": True, + "addOptions": {"searchForMissingAlbums": search_all, "monitor": "all"}, + } + if dry_run: + print(f"[dry-run] POST /api/v1/artist {json.dumps(payload)}") + return {"id": -1, "artistName": name, **payload} + try: + return lidarr_post("/api/v1/artist", payload) + except RequestException as e: + err(f"add_artist failed: {e}") + return None + + +def ensure_album_in_library(album: dict, root: str, search_all: bool, dry_run: bool) -> Optional[dict]: + """Return a library album dict (with numeric id). Adds artist if needed.""" + # Already in library? + if album.get("id") and isinstance(album.get("id"), int) and album.get("id") > 0 and not album.get("foreignAlbumId", "").startswith("lookup"): + # Heuristic: lookup results carry a 0/None id; library albums carry real ids. + if album.get("artistId"): + return album + + artist_obj = album.get("artist") or {} + artist_name = artist_obj.get("artistName") or album.get("artistName") or "" + existing = get_existing_artist(artist_name) if artist_name else None + if not existing: + print(f"Adding artist '{artist_name}' to Lidarr...") + existing = add_artist(artist_obj or {"artistName": artist_name, + "foreignArtistId": artist_obj.get("foreignArtistId")}, + root, search_all, dry_run) + if not existing: + return None + + if dry_run: + print(f"[dry-run] would resolve album '{album.get('title')}' under artist id {existing.get('id')}") + return {**album, "id": album.get("id") or -1, "artistId": existing.get("id")} + + # Find the album in the (now-present) artist's albums by title match. + try: + albums = lidarr_get("/api/v1/album", params={"artistId": existing["id"]}, timeout=15) + for a in albums: + if a.get("title", "").lower() == album.get("title", "").lower(): + return a + if albums: + return albums[0] + except RequestException as e: + dbg(f"album list fetch failed: {e}") + return None + + +def release_available(album_id: int) -> bool: + """Interactive search: does any indexer have a release for this album?""" + try: + releases = lidarr_get("/api/v1/release", params={"albumId": album_id}, timeout=90) + dbg(f"interactive search returned {len(releases)} releases for album {album_id}") + return len(releases) > 0 + except RequestException as e: + dbg(f"release search failed: {e}") + return False + + +def trigger_album_search(album_id: int, dry_run: bool): + if dry_run: + print(f"[dry-run] POST /api/v1/command AlbumSearch albumIds=[{album_id}]") + return + lidarr_post("/api/v1/command", {"name": "AlbumSearch", "albumIds": [album_id]}) + + +def act_lidarr_album(hit: Hit, root: str, search_all: bool, dry_run: bool) -> bool: + """Returns True if Lidarr handled it; False to fall through to YouTube.""" + album = hit.payload["album"] + lib_album = ensure_album_in_library(album, root, search_all, dry_run) + if not lib_album: + err("Could not resolve album in Lidarr.") + return False + album_id = lib_album.get("id") + if dry_run: + print(f"[dry-run] would interactive-search album id {album_id}; " + f"if no release found, fall through to YouTube.") + trigger_album_search(album_id, dry_run) + return True + + if isinstance(album_id, int) and album_id > 0 and release_available(album_id): + print(f"Indexer release available — triggering Lidarr grab for '{hit.album}'.") + trigger_album_search(album_id, dry_run) + return True + print("No indexer release found in Lidarr — falling through to YouTube.") + return False + + +def act_lidarr_artist(hit: Hit, root: str, search_all: bool, dry_run: bool) -> bool: + artist = hit.payload["artist"] + print(f"Adding artist '{hit.artist}' to Lidarr...") + result = add_artist(artist, root, search_all, dry_run) + return result is not None + + +# --------------------------------------------------------------------------- +# YouTube download +# --------------------------------------------------------------------------- +def _quality_args(quality: str) -> list[str]: + if quality == "best": + # bestaudio, prefer mp3 320 only if extraction needs a container. + return ["-f", "bestaudio/best", "-x", "--audio-quality", "0"] + if quality == "320": + return ["-f", "bestaudio/best", "-x", "--audio-format", "mp3", "--audio-quality", "0"] + if quality in ("m4a", "opus", "flac"): + return ["-f", "bestaudio/best", "-x", "--audio-format", quality, "--audio-quality", "0"] + return ["-f", "bestaudio/best", "-x"] + + +def yt_download(url_or_query: str, target_folder: str, quality: str, dry_run: bool, + hit: Optional[Hit] = None): + cmd = ["yt-dlp", + *_quality_args(quality), + "--embed-metadata", + "--embed-thumbnail", + "--no-playlist", + "-P", target_folder] + # Override tags from the chosen hit so they don't rely on scraped titles. + if hit: + if hit.artist: + cmd += ["--replace-in-metadata", "artist", ".*", hit.artist] + if hit.album: + cmd += ["--parse-metadata", f"{hit.album}:%(album)s"] + if hit.title: + cmd += ["--parse-metadata", f"{hit.title}:%(title)s"] + if hit.year: + cmd += ["--parse-metadata", f"{hit.year}:%(release_year)s"] + cmd.append(url_or_query) + + if dry_run: + print(f"[dry-run] mkdir -p {target_folder}") + print(f"[dry-run] {' '.join(cmd)}") + return + os.makedirs(target_folder, exist_ok=True) + print(f"Downloading via yt-dlp -> {target_folder}") + subprocess.run(cmd) + + +def act_youtube(hit: Hit, root: str, quality: str, dry_run: bool): + vid = hit.payload.get("videoId") + # Prefer YouTube Music URL for correct album art / topic metadata. + url = f"https://music.youtube.com/watch?v={vid}" if vid else f"ytsearch1:{hit.artist} {hit.title}" + artist_dir = hit.artist.split(",")[0].strip() or "Unknown Artist" + target = os.path.join(root, artist_dir, "youtube") + yt_download(url, target, quality, dry_run, hit=hit) + + +# --------------------------------------------------------------------------- +# URL path +# --------------------------------------------------------------------------- +def run_yt_dlp_get_metadata(url: str) -> Optional[dict]: try: result = subprocess.run( ["yt-dlp", "-j", "--no-playlist", url], - capture_output=True, - text=True, - check=True + capture_output=True, text=True, check=True, ) - metadata = json.loads(result.stdout) - return metadata # Return full metadata dict here - except subprocess.CalledProcessError as e: - print(f"yt-dlp subprocess error: {e}") - print(f"stderr: {e.stderr}") - except json.JSONDecodeError as e: - print(f"JSON decoding error: {e}") - print(f"Output was: {result.stdout}") - except Exception as e: - print(f"Failed to extract metadata from URL: {e}") - return None - -def get_artist_from_metadata(metadata): - # Try different possible keys for artist/creator in metadata - artist = None - for key in ["artist", "creator", "uploader", "channel"]: - artist = metadata.get(key) - if artist: - break - # fallback to title parsing if nothing found - if not artist and "title" in metadata: - # Try to parse artist from title "Artist - Track" - parts = metadata["title"].split(" - ", 1) - if len(parts) == 2: - artist = parts[0].strip() - return artist or "Unknown Artist" - -def yt_dlp_download(url_or_query, target_folder): - print(f"Downloading via yt-dlp to: {target_folder}") - os.makedirs(target_folder, exist_ok=True) - subprocess.run([ - "yt-dlp", - "--embed-metadata", - "--parse-metadata", "%(title)s:%(album)s", - "--embed-thumbnail", - "-f", "bestaudio", - "-x", - "-P", target_folder, - url_or_query - ]) - -def search_artist(name, timeout_seconds=15): - try: - resp = requests.get( - f"{LIDARR_URL}/api/v1/artist/lookup", - headers=headers, - params={"term": name}, - timeout=timeout_seconds - ) - resp.raise_for_status() - results = resp.json() - if results: - artist = results[0] - print(f"Found artist: {artist.get('artistName') or artist.get('title')}") - return artist - return None - except Timeout: - print(f"Lidarr artist search timed out after {timeout_seconds} seconds.") - return None - except requests.RequestException as e: - print(f"Lidarr artist search failed: {e}") + return json.loads(result.stdout) + except (subprocess.CalledProcessError, json.JSONDecodeError) as e: + err(f"yt-dlp metadata extraction failed: {e}") return None -def get_existing_artist(name): - print(f"--> Checking if artist '{name}' already exists in Lidarr...") - try: - if DEBUG: - print("...Sending request to /api/v1/artist") - resp = requests.get(f"{LIDARR_URL}/api/v1/artist", headers=headers, timeout=10) - if DEBUG: - print("...Got response from Lidarr") - resp.raise_for_status() - for artist in resp.json(): - if artist["artistName"].lower() == name.lower(): - print("...Artist match found!") - return artist - print("...Artist not found in existing list.") - except requests.exceptions.Timeout: - print("!!! Timeout during existing artist check.") - except requests.exceptions.RequestException as e: - if DEBUG: - print(f"[DEBUG] Exception during existing artist check: {e}") - return None -def get_default_metadata_profile_id(): - try: - resp = requests.get(f"{LIDARR_URL}/api/v1/metadataprofile", headers=headers, timeout=10) - resp.raise_for_status() - profiles = resp.json() - if profiles: - return profiles[0]["id"] - except Exception as e: - print(f"[ERROR] Could not fetch metadata profiles: {e}") - return 1 # fallback to 1 if not found +def get_artist_from_metadata(meta: dict) -> str: + for key in ("artist", "creator", "uploader", "channel"): + if meta.get(key): + return meta[key] + if "title" in meta and " - " in meta["title"]: + return meta["title"].split(" - ", 1)[0].strip() + return "Unknown Artist" -def add_artist(metadata_artist, search_for_missing_albums=False): - foreign_id = metadata_artist.get("foreignArtistId") or metadata_artist.get("id") - artist_name = metadata_artist.get("artistName") or metadata_artist.get("title") - if not foreign_id or not artist_name: - if DEBUG: - print("[DEBUG] Metadata received:", metadata_artist) - raise ValueError("Could not find foreignArtistId or artistName in metadata.") +def handle_url(url: str, root: str, quality: str, dry_run: bool): + meta = run_yt_dlp_get_metadata(url) + artist = get_artist_from_metadata(meta) if meta else "Unknown Artist" + target = os.path.join(root, artist, "youtube") + yt_download(url, target, quality, dry_run) - metadata_profile_id = get_default_metadata_profile_id() - data = { - "foreignArtistId": foreign_id, - "artistName": artist_name, - "qualityProfileId": 1, - "metadataProfileId": metadata_profile_id, - "rootFolderPath": ROOT_FOLDER, - "monitored": True, - "addOptions": { - "searchForMissingAlbums": search_for_missing_albums, - "monitor": "all" - } - } +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- +def build_combined_hits(query, limit, yt_first, lidarr_only, yt_only) -> list[Hit]: + lidarr_hits: list[Hit] = [] + yt_hits: list[Hit] = [] + with ThreadPoolExecutor(max_workers=2) as ex: + f_lid = None if yt_only else ex.submit(lidarr_search, query, limit) + f_yt = None if lidarr_only else ex.submit(youtube_search, query, limit) + if f_lid: + lidarr_hits = f_lid.result() + if f_yt: + yt_hits = f_yt.result() + return (yt_hits + lidarr_hits) if yt_first else (lidarr_hits + yt_hits) - if DEBUG: - print("[DEBUG] Sending payload to Lidarr /api/v1/artist:", json.dumps(data, ensure_ascii=False)) - try: - resp = requests.post( - f"{LIDARR_URL}/api/v1/artist", - headers=headers, - json=data, - timeout=15 - ) - resp.raise_for_status() - return resp.json() - except requests.RequestException as e: - print("Lidarr add_artist failed. Falling back to yt-dlp.") - if DEBUG: - print(f"[DEBUG] Lidarr add_artist exception: {e}") - return None +def parse_args(): + p = argparse.ArgumentParser( + prog="musicfetch", + description="Fetch music via Lidarr (preferred) or YouTube Music.") + p.add_argument("query", nargs="+", help="Free-form query or a URL.") + p.add_argument("-n", "--noninteractive", action="store_true", + help="Auto-pick the top hit, no prompt.") + p.add_argument("-s", "--ytsearch", action="store_true", + help="YouTube first instead of Lidarr first.") + p.add_argument("-d", "--dry-run", action="store_true", + help="Show actions without executing them.") + p.add_argument("-q", "--quality", choices=QUALITY_CHOICES, default="best", + help="Audio quality/format (default: best).") + p.add_argument("--limit", type=int, default=10, help="Hits per source (default 10).") + p.add_argument("--lidarr-only", action="store_true", help="Skip YouTube.") + p.add_argument("--yt-only", action="store_true", help="Skip Lidarr.") + p.add_argument("-o", "--root", default=DEFAULT_ROOT, help=f"Output root (default {DEFAULT_ROOT}).") + p.add_argument("--search-all", action="store_true", + help="Search all albums when adding an artist to Lidarr.") + p.add_argument("--debug", action="store_true", help="Verbose output.") + return p.parse_args() -def search_local_album(track_name, artist_id=None, artist_name=None): - try: - print(f"--> Searching local albums for track: '{track_name}'") - resp = requests.get( - f"{LIDARR_URL}/api/v1/album", - headers=headers, - timeout=10 - ) - resp.raise_for_status() - albums = resp.json() - print(f"...Loaded {len(albums)} local albums") - - # Search by artist_id and track_name - for album in albums: - # Check artist ID match if provided - if artist_id and album.get("artistId") != artist_id: - continue - # Check artist name match if provided - if artist_name and album.get("artist", {}).get("artistName", "").lower() != artist_name.lower(): - continue - # Check if track_name matches album title (case insensitive substring) - if track_name.lower() in album.get("title", "").lower(): - if DEBUG: - print(f"[DEBUG] Found local album: {album.get('title')} by {album.get('artist', {}).get('artistName')}") - return album - if DEBUG: - print("[DEBUG] No matching local album found.") - except requests.exceptions.Timeout: - if DEBUG: - print("[DEBUG] Timeout during local album search.") - except requests.exceptions.RequestException as e: - if DEBUG: - print(f"[DEBUG] Exception during local album search: {e}") - return None - -def search_album(track_name, artist_id): - # Check local albums first - album = search_local_album(track_name, artist_id=artist_id) - if album: - return album - - # Fallback to external lookup - try: - if DEBUG: - print(f"[DEBUG] Searching album externally with track: '{track_name}' and artist ID: {artist_id}") - resp = requests.get( - f"{LIDARR_URL}/api/v1/album/lookup", - headers=headers, - params={"term": track_name, "artistId": artist_id}, - timeout=10 - ) - resp.raise_for_status() - results = resp.json() - if DEBUG: - print(f"[DEBUG] Found {len(results)} results from album lookup with artist ID.") - return results[0] if results else None - except requests.exceptions.Timeout: - if DEBUG: - print("[DEBUG] Timeout during album search (with artist ID).") - except requests.exceptions.RequestException as e: - if DEBUG: - print(f"[DEBUG] Exception during album search (with artist ID): {e}") - return None - -def search_album_by_artist(track_name, artist_name): - # Check local albums first - album = search_local_album(track_name, artist_name=artist_name) - if album: - return album - - # Fallback to external lookup - try: - if DEBUG: - print(f"[DEBUG] Fallback external search: '{artist_name} {track_name}'") - resp = requests.get( - f"{LIDARR_URL}/api/v1/album/lookup", - headers=headers, - params={"term": f"{artist_name} {track_name}"}, - timeout=10 - ) - resp.raise_for_status() - results = resp.json() - if DEBUG: - print(f"[DEBUG] Found {len(results)} results from fallback search.") - return results[0] if results else None - except requests.exceptions.Timeout: - if DEBUG: - print("[DEBUG] Timeout during fallback album search.") - except requests.exceptions.RequestException as e: - if DEBUG: - print(f"[DEBUG] Exception during fallback album search: {e}") - return None - -def trigger_album_search(album_id): - data = { - "name": "AlbumSearch", - "albumIds": [album_id] - } - resp = requests.post( - f"{LIDARR_URL}/api/v1/command", - headers=headers, - json=data - ) - resp.raise_for_status() - if DEBUG: - print(f"[DEBUG] Triggered Lidarr search for album ID {album_id}") def main(): global DEBUG - DEBUG = False - search_for_missing_albums = False + args = parse_args() + DEBUG = args.debug + query = " ".join(args.query).strip() - # Parse extra arguments - args = sys.argv[1:] - input_str = None - for arg in args: - if arg == '--debug': - DEBUG = True - elif arg == '--search-all': - search_for_missing_albums = True - elif not input_str: - input_str = arg - - if not input_str: - print("Usage: musicfetch.py [--debug] [--search-all] 'Artist - Track' OR 'https://youtube.com/watch?...'") + if args.lidarr_only and args.yt_only: + err("--lidarr-only and --yt-only are mutually exclusive.") sys.exit(1) - if is_url(input_str): - if DEBUG: - print("Input is a URL. Extracting metadata to find artist...") - metadata = run_yt_dlp_get_metadata(input_str) - if metadata: - artist_name = get_artist_from_metadata(metadata) - if DEBUG: - print(f"Extracted artist name: {artist_name}") - else: - if DEBUG: - print("Failed to get metadata from URL.") - artist_name = "Unknown Artist" - - target_folder = os.path.join(ROOT_FOLDER, artist_name, "youtube") - yt_dlp_download(input_str, target_folder) + if is_url(query): + handle_url(query, args.root, args.quality, args.dry_run) return - if " - " not in input_str: - print("Input must be in 'Artist - Track' format.") + hits = build_combined_hits(query, args.limit, args.ytsearch, + args.lidarr_only, args.yt_only) + if not hits: + print("No hits found from any source.") sys.exit(1) - artist_name, track_name = input_str.split(" - ", 1) - artist_name = artist_name.strip() - track_name = track_name.strip() + chosen = pick(hits, query, args.noninteractive, args.ytsearch) + if not chosen: + print("Nothing selected.") + return - if DEBUG: - print(f"Looking up artist: {artist_name}") - artist = get_existing_artist(artist_name) - - if not artist: - print("Artist not found. Searching Lidarr metadata (with timeout)...") - metadata = search_artist(artist_name) - if not metadata: - if DEBUG: - print("No match found or search timed out. Falling back to yt-dlp.") - yt_dlp_download(f"ytsearch:{input_str}", os.path.join(ROOT_FOLDER, artist_name, "youtube")) - return - - print(f"Adding artist: {metadata.get('artistName') or metadata.get('title')}") - artist = add_artist(metadata, search_for_missing_albums=search_for_missing_albums) - if not artist: - yt_dlp_download(f"ytsearch:{input_str}", os.path.join(ROOT_FOLDER, artist_name, "youtube")) - return - - album = search_album(track_name, artist["id"]) or search_album_by_artist(track_name, artist_name) - - if album: - if DEBUG: - print(f"Found album '{album['title']}' for track '{track_name}', triggering search...") - trigger_album_search(album["id"]) + if chosen.source == "lidarr": + if chosen.kind == "album": + handled = act_lidarr_album(chosen, args.root, args.search_all, args.dry_run) + if not handled and not args.lidarr_only: + # Fall through to the top YouTube hit for the same query. + yt_fallback = next((h for h in hits if h.source == "youtube"), None) + if yt_fallback: + print("Using top YouTube hit as fallback.") + act_youtube(yt_fallback, args.root, args.quality, args.dry_run) + else: + print("No YouTube fallback available.") + else: + act_lidarr_artist(chosen, args.root, args.search_all, args.dry_run) else: - if DEBUG: - print("No album found in Lidarr. Falling back to yt-dlp.") - yt_dlp_download(input_str, os.path.join(ROOT_FOLDER, artist_name, "youtube")) + act_youtube(chosen, args.root, args.quality, args.dry_run) + if __name__ == "__main__": main()