izebra_projects/musicfetch
1
0
Fork 0
Code Issues Pull Requests Activity

Rewrite musicfetch as v2: dual-source search with interactive picker

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Zeb Hering
2026-06-08 19:30:47 -07:00
parent 51871fd296
commit 9fd4c8585b
3 changed files with 661 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
__pycache__/
*.pyc

147
README.md
View File

@@ -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/
<root>/
├── 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.
---

888
musicfetch Normal file → Executable file
View File

@@ -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()