48 Commits

Author SHA1 Message Date
b26e321926 feat: throttle yt-dlp requests to dodge YouTube rate-limiting
Bulk metadata fetches trip YouTube's per-session rate limit ("This
content isn't available, try again later"), failing even single-worker
runs after a burst. Add --sleep-requests between extraction calls (and a
randomized --sleep-interval before downloads), default 1s, tunable via
--sleep / $YTDLP_SLEEP (0 disables). Applied to metadata, search, probe,
and download yt-dlp invocations.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 23:39:08 -07:00
47f3482192 fix: Odesli links 'not found' when Odesli omits YouTube link
Root cause: Odesli's linksByPlatform frequently lacks youtube/youtubeMusic
(confirmed live for many Spotify tracks). resolve_link_hits only added a
YouTube hit when Odesli supplied one, so with no Lidarr match the hit list
was empty -> server 404 'No results found'.

Fix: always run a normal youtube_search (via build_combined_hits) for the
fallback; when Odesli DID return a link, insert its exact track as the first
YouTube hit. Lidarr-first ordering preserved.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 22:14:57 -07:00
a4b5039e7f Merge odesli-link-resolution: any link -> Lidarr-first, YouTube fallback
Resolve any streaming link (Spotify/Apple/Tidal/...) via the Odesli/song.link
API, then run the existing Lidarr-first, exact-YouTube-fallback flow. YouTube
and SoundCloud links keep direct yt-dlp download; text queries unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 22:07:21 -07:00
b899d75930 fix: _is_direct_url label-boundary host match (no notyoutube.com false positive)
Review finding: bare endswith routed look-alike hosts to the direct yt-dlp
path. Match on a domain-label boundary and drop the redundant _DIRECT_HOSTS.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 22:02:57 -07:00
32acd038c8 docs: document any-link Odesli resolution in README
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:59:59 -07:00
eb45a3680f feat: server /fetch resolves non-direct links via Odesli (Lidarr-first)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:59:33 -07:00
44aaa1f93e feat: re-export odesli symbols through server/mf.py
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:56:56 -07:00
8daf780023 feat: route non-direct CLI links through Odesli (handle_link)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:56:02 -07:00
9c308fefc7 feat: resolve_link_hits + handle_link — Odesli link -> Lidarr-first flow
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:55:04 -07:00
a4e1dc1643 refactor: extract _dispatch_chosen from main() for reuse
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:53:30 -07:00
9fccf9015a feat: _is_direct_url — route YouTube/SoundCloud links to direct download
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:52:14 -07:00
a88f4c594a feat: odesli_resolve — resolve any song link to metadata via song.link
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 21:47:22 -07:00
95a448ef58 docs: implementation plan for Odesli link resolution
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 20:57:18 -07:00
dcb3014fb0 docs: spec for Odesli link resolution (any link -> Lidarr-first, YouTube fallback)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 20:33:26 -07:00
140bfef7c9 feat: yt-dlp cookie support + surface real failure reason; default workers 4
Bulk --repair on unauthenticated YouTube trips the bot-check (HTTP 429 "Sign
in to confirm you're not a bot"), after which every call fails until the IP
flag clears. Add cookie support so authenticated requests bypass it:

- --cookies FILE / --cookies-from-browser BROWSER (and $YTDLP_COOKIES /
  $YTDLP_COOKIES_FROM_BROWSER for the API container), threaded into every
  yt-dlp invocation (search, probe, download, repair metadata fetch).
- run_yt_dlp_get_metadata now logs yt-dlp's last stderr line (the actual 429 /
  bot-check / network reason) instead of a bare exit code.
- Default --repair workers lowered 8 -> 4 (safe without cookies; raise with).
- compose: optional YTDLP_COOKIES env + commented cookies mount.
- README: how to obtain cookies (Chrome/Firefox, browser-read vs cookies.txt
  export); gitignore cookies.txt.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 11:25:39 -07:00
92742b9ad6 perf: parallelize --repair with a thread pool (--workers, default 8)
Each repaired file is an independent yt-dlp metadata round-trip, so repair is
network-bound; run them concurrently via ThreadPoolExecutor. Adds --workers
(default 8) to cap concurrency and a progress line every 100 files. At ~50k
tracks this turns a ~day-long sequential run into hours. Lower --workers if
YouTube rate-limits (429/403).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 11:13:52 -07:00
0347a638cf fix: reliable YouTube tagging, loud Lidarr failures, deno runtime, repair recovery
Root cause of bad album/title tags: yt-dlp's --parse-metadata reads a
single-word FROM (matching field_to_template's ^[a-zA-Z_]+$) as a *field
name*, so literal one-word titles/albums like "Cochise" became "NA". Inject
literals via seed-then-replace into meta_<tag> instead (--parse-metadata to
create the field, --replace-in-metadata with literal args to set it), which
is immune to template parsing and also creates tags the source lacks.

- yt_download: literal-safe meta_artist/title/album; hit album no longer
  clobbered by the Unknown-Album default; artist tag now created when missing.
- lidarr_search: connection/timeout errors surface via err() ("Lidarr
  unreachable … falling back to YouTube") instead of silent dbg(), so the
  YouTube fallback isn't mistaken for "no Lidarr match".
- Dockerfile: install deno (arch-aware) — the JS runtime yt-dlp needs for
  YouTube; without it: "No supported JavaScript runtime" / HTTP 403.
- repair: treat NA/Unknown placeholders as bogus and overwrite title/artist
  from source (was fill-missing-only); normalise literal "NA" album to
  "Unknown Album"; rename bogus "NA [<id>]" filenames to the recovered title.
- README updated; .gitignore excludes server/log.txt.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 11:09:27 -07:00
33ca743a34 removed useless portion of compose 2026-06-11 22:02:51 -07:00
7951c436dd Merge feat/unknown-album: default blank album to 'Unknown Album' 2026-06-11 21:53:11 -07:00
8b881c14bf feat: always embed an album tag (default 'Unknown Album')
Downloads with no album (regular YouTube videos, fan edits, etc) left a blank
album, which trips up players like Plexamp. yt_download now appends
--parse-metadata "%(album|Unknown Album)s:%(meta_album)s" so the native/resolved
album is kept when present, else 'Unknown Album' is embedded. Applies to all
download paths (search, playlist, single URL).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-11 21:53:11 -07:00
530b5b0406 typo in mutagen. fixed 2026-06-11 21:51:16 -07:00
0a4e6d474a added mutagen to requirements.txt 2026-06-11 21:48:33 -07:00
c0503187c5 Merge feat/repair-fast-meta: faster --repair via player_skip=js 2026-06-10 22:52:39 -07:00
a6aa469084 perf(repair): skip YouTube JS signature step when fetching tags
--repair only reads metadata (never downloads), so pass
--extractor-args youtube:player_skip=js to yt-dlp. Keeps album/artist/year/title
but avoids the slow, throttle-prone nsig JS step (which crawls without a JS
runtime and trips YouTube rate-limiting during bulk runs). run_yt_dlp_get_metadata
gains an optional extra_args param; the download path is unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 22:52:39 -07:00
f071158c10 Merge feat/repair-exclude: -x/--exclude for repair/retag 2026-06-10 22:41:16 -07:00
c6bde6958a feat: -x/--exclude to skip folders during --repair/--retag-from-path
Repeatable -x/--exclude NAME skips any artist- or source-level folder whose name
matches (case-insensitive) when walking the library, so hand-curated folders like
/media/music/Unsorted or .../playlists are left untouched. Threaded through
_iter_source_files -> repair_library / retag_library_from_path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 22:41:16 -07:00
7ea3ad2538 Merge feat/retag-from-path: offline tag recovery 2026-06-10 22:30:25 -07:00
9af7f91a25 feat: --retag-from-path to recover tags damaged by a prior --repair
Offline re-tag of artist/title from the artist folder + filename: strips
(Official Video)/(Lyrics)-style decorations and trailing [id], and treats an
'Artist - Title' filename as authoritative (recovering the real artist for
music videos filed under a channel name). Overwrites artist/title only; leaves
album/year. Honors --dry-run.

Refactors the source-folder walk into _iter_source_files, shared by --repair
and --retag-from-path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 22:30:25 -07:00
567d7578ad Merge fix/repair-music-videos: conservative --repair + first-artist single folder 2026-06-10 18:53:27 -07:00
c6e28a4f75 fix: harden --repair against music videos; first-artist folder for single URLs
--repair was clobbering good tags and erroring on real libraries:
- Validate the parsed id per source (YouTube 11-char, SoundCloud numeric) so
  junk ids from bracketed descriptors ([Official Video]) are skipped, not queried.
- Skip files whose source returns no real music metadata (no album/year, e.g.
  music videos) instead of overwriting clean tags with channel/decorated titles.
- Year from release info only (sane 1000-2100), never upload_date (which gave
  wrong years for old songs and bogus values like 6577).
- album/year are authoritative; artist/title are fill-missing-only (no clobber).

Also: download_single now uses the first artist for the folder (matching the
search/playlist paths) so single-URL downloads stop creating multi-artist dirs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 18:53:27 -07:00
1a81f64cc3 Merge feat/repair: re-tag existing downloads via --repair 2026-06-09 19:11:43 -07:00
fdc3cc84a5 feat: --repair flag to re-tag existing downloads from source metadata
Walks <root>/<artist>/<source>/ (known yt-dlp source folders only; skips Lidarr
album dirs), re-queries each file's source by the [id] in its filename, and fixes
tags (album/year/artist/title) via mutagen. Honors --dry-run for preview. CLI-only
(not the REST API). Fixes downloads that landed with missing album / wrong year.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-09 19:11:43 -07:00
74eb63b243 Merge feat/multi-platform: any yt-dlp site, single tracks + playlists
- probe_url() classifies any URL via yt-dlp (SoundCloud sets, Bandcamp albums,
  etc); YouTube playlists still use ytmusicapi for richer metadata.
- Per-source folders <root>/<artist>/<extractor>/; non-YouTube tracks download
  by their native URL (YouTube keeps the music.youtube album-art URL).
- Sparse-metadata playlist tracks route via yt-dlp output template so they land
  under the real artist.
Live-verified: SoundCloud track + set, YouTube playlist regression.
2026-06-09 06:56:18 -07:00
6730f1f141 fix: route sparse-metadata playlist tracks by yt-dlp's own metadata
SoundCloud sets (and similar) return flat-playlist entries without per-track
artist/title. When a track Hit has no artist, download via an output template
(-o <root>/%(artist,uploader,channel)s/<source>/...) so yt-dlp places the file
under the real artist instead of "Unknown Artist". yt_download gains an optional
outtmpl mode.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-09 06:55:56 -07:00
f103b6c253 feat: multi-platform URL & playlist support via yt-dlp probe
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>
2026-06-09 00:25:58 -07:00
7309ad3a29 Merge feat/playlists-profiles: YouTube playlists + Lidarr profile hardening
- Playlist URLs download each track to per-artist folders (CLI + REST API).
  One playlist = one job; done if >=1 track succeeds ("Downloaded N/M tracks").
- REST API /fetch now routes URL/playlist queries to download jobs.
- Lidarr metadata/quality profiles selected by name with env overrides
  (LIDARR_METADATA_PROFILE/LIDARR_QUALITY_PROFILE), no more position-luck.
2026-06-09 00:13:54 -07:00
90b9a01872 fix(server): use .get() for title/artist in perform_url_fetch result
Defensive access guards against download_single returning ok=True
without title/artist keys, avoiding a KeyError in the job worker.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-09 00:00:45 -07:00
0f7ddd7697 feat(server): route URL/playlist /fetch to download jobs
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-08 23:58:37 -07:00
ca36d2bb27 feat(server): re-export URL helpers; callable job done_message
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-08 23:54:49 -07:00
aa9d177ed1 feat(youtube): playlist expansion + per-track download, success bools
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-08 23:49:26 -07:00
3ee49b17bd fix(lidarr): select metadata/quality profiles by name with env overrides
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-08 23:44:34 -07:00
6e6bec7a0d Plan playlists + profile hardening (5 TDD tasks)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-08 23:42:13 -07:00
a24c894c61 Add design spec: YouTube playlists + Lidarr profile hardening
Playlists download each track to per-track artist folders (CLI + REST API,
one job per playlist, done if >=1 track succeeds). Profile selection by name
with env overrides (LIDARR_METADATA_PROFILE/LIDARR_QUALITY_PROFILE).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-08 23:38:06 -07:00
a424fbfd2f Merge feat/rest-api: REST API + smarter Lidarr matching
- FastAPI async job-based REST API wrapping musicfetch (X-API-Key auth,
  Siri-friendly messages, dockerized for the Lidarr stack).
- Smarter Lidarr search: MusicBrainz track->album resolution + exact
  mbid: lookup (prefers own-artist studio album), no fuzzy ranking.
- Bug fixes from live testing: single first-artist tag (no doubling).
2026-06-08 23:31:41 -07:00
b99e5eb9cb fix(lidarr): prefer own-artist studio album over various-artists comps
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-08 23:30:15 -07:00
1661cb1742 refactor(lidarr): drop now-unused Timeout import 2026-06-08 23:24:59 -07:00
18f72a5626 feat(lidarr): exact MBID album lookup via MusicBrainz resolution
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-08 23:22:11 -07:00
babbd84fda feat(lidarr): MusicBrainz track-to-album resolver
Add musicbrainz_best_album() that resolves an artist+track pair to its
best studio album via the MusicBrainz search API, with a 1 req/sec
courtesy rate-limiter. Prefers plain studio albums over compilations,
singles, and live releases; falls back to any release group when no
studio album is found. Never raises — returns None on any failure.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-08 23:17:43 -07:00
24 changed files with 3963 additions and 97 deletions

3
.gitignore vendored
View File

@@ -1,2 +1,5 @@
__pycache__/
*.pyc
server/log.txt
cookies.txt

122
README.md
View File

@@ -8,6 +8,10 @@ whatever you choose. It accepts:
- 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).
- **Any streaming link** (Spotify, Apple Music, Tidal, Deezer, …): resolved to
metadata via [Odesli/song.link](https://odesli.co), then searched on Lidarr
first and downloaded from the matching YouTube track if no Lidarr release is
available. YouTube and SoundCloud links download directly.
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
@@ -94,6 +98,12 @@ export LIDARR_API_KEY="your-lidarr-api-key"
| `--yt-only` | Skip Lidarr. |
| `-o`, `--root PATH` | Output root folder (default `/media/music`). |
| `--search-all` | Search all albums when adding an artist to Lidarr. |
| `--repair` | Re-tag existing downloads under `--root` from source metadata (see below). |
| `--workers N` | Parallel metadata fetches during `--repair` (default 4). |
| `--cookies FILE` | yt-dlp `cookies.txt` for authenticated YouTube (avoids bot-check / rate limits). |
| `--cookies-from-browser BROWSER` | Load YouTube cookies from a local browser (e.g. `firefox`). |
| `--retag-from-path` | Offline: re-tag artist/title from folder + filename (see below). |
| `-x`, `--exclude NAME` | Folder under `--root` to skip during `--repair`/`--retag-from-path` (repeatable). |
| `--debug` | Verbose output. |
### Examples
@@ -115,8 +125,112 @@ export LIDARR_API_KEY="your-lidarr-api-key"
# YouTube only, lossless preferred
./musicfetch --yt-only -q flac "Bonobo - Kerala"
# Download by URL (YouTube Music URL preferred for correct art)
# Download by URL (single track or playlist/set/album, any yt-dlp site)
./musicfetch "https://music.youtube.com/watch?v=xxxxxxxxxxx"
./musicfetch "https://soundcloud.com/artist/sets/my-mix"
```
### 🔧 Repair existing tags
`--repair` walks `<root>/<artist>/<source>/` (the `youtube`/`soundcloud`/… download
folders — Lidarr album folders are skipped), re-fetches authoritative metadata for each
file using the `[id]` in its filename, and fixes tags. Useful when downloads landed with
missing album or wrong year.
It is deliberately **conservative**: it overwrites **album** and **year** (the usual
breakage), and fills in **artist**/**title** when they are missing *or* a known-bogus
placeholder (`NA`, `Unknown Album`, `Unknown Artist` — left behind by older buggy tagging) —
but it never overwrites a genuine existing artist/title with a channel name or decorated video
title. A bogus `NA [<id>].<ext>` filename is renamed to the recovered title, and a literal
`NA` album with no source album is normalised to `Unknown Album`.
Each file is its own yt-dlp network round-trip, so repair runs them in a thread pool;
`--workers N` (default 4) caps concurrency. Progress prints every 100 files. Requires
`mutagen` (a yt-dlp dependency, usually already present). CLI-only — not exposed via the REST API.
**Cookies (important for bulk repair).** Unauthenticated YouTube requests get throttled fast —
a large `--repair` (or even a `--dry-run`, which still fetches) will trip *"Sign in to confirm
you're not a bot"* (HTTP 429) and every subsequent call fails until the IP-level flag clears.
Pass authenticated cookies to avoid it:
```bash
./musicfetch --repair --cookies /path/cookies.txt -o /media/music # exported cookies.txt
./musicfetch --repair --cookies-from-browser firefox -o /media/music # or read from a browser
```
With cookies you can raise `--workers`; without them keep it low (≤4) and expect occasional
throttling. Cookies also apply to normal fetches/downloads. The same can be set for the API
container via `$YTDLP_COOKIES` / `$YTDLP_COOKIES_FROM_BROWSER`. If you do get flagged, **stop**
retrying extends it; wait ~30-60 min (429) or longer for a bot-check.
#### Getting YouTube cookies
> ⚠️ Use a **throwaway / secondary Google account**, not your main one — bulk automated
> requests can get the account flagged. You must be **logged in to YouTube** in the browser
> first.
**Option A — read straight from the browser (simplest, host CLI only).**
`--cookies-from-browser` reads the browser's own cookie store, so there's nothing to export:
```bash
./musicfetch --repair --cookies-from-browser firefox -o /media/music
./musicfetch --repair --cookies-from-browser chrome -o /media/music
```
- **Firefox:** works while open; just be logged in to YouTube.
- **Chrome / Chromium / Brave / Edge:** must be **fully quit** when you run this (Chrome locks
its cookie DB, and newer versions encrypt it — close the browser entirely first). On Linux a
running Chrome will usually fail with a "could not copy cookie database / locked" error.
- Specify a profile if not the default, e.g. `--cookies-from-browser "chrome:Profile 1"`.
This only works where the browser lives (your host), **not** inside the Docker container.
**Option B — export a `cookies.txt` (works anywhere, incl. the container/server).**
Use a Netscape-format cookie exporter, then point `--cookies` / `$YTDLP_COOKIES` at the file:
1. Install a cookies exporter extension:
- Firefox: *"cookies.txt"* (a.k.a. *Export Cookies*).
- Chrome: *"Get cookies.txt LOCALLY"* (pick a **LOCALLY**-running one — avoid extensions that
upload your cookies anywhere).
2. Log in to <https://www.youtube.com>, click the extension, **Export** → save `cookies.txt`.
3. Use it:
```bash
./musicfetch --repair --cookies ~/cookies.txt -o /media/music
```
For the API container, mount it and set the env var (see `server/docker-compose.yml`):
```yaml
environment:
YTDLP_COOKIES: "/cookies.txt"
volumes:
- /host/path/cookies.txt:/cookies.txt:ro
```
Cookies expire — if YouTube starts rejecting them, re-export. Treat `cookies.txt` like a
password (it *is* your logged-in session); keep it out of git (`.gitignore` it).
```bash
# Preview what would change (writes nothing)
./musicfetch --repair -d
# Apply fixes under a specific root
./musicfetch --repair -o /media/music
```
**`--retag-from-path`** is an offline companion: it derives **artist** and **title** purely
from the folder name + filename (stripping `(Official Video)` / `(Lyrics)`-style decorations,
and treating an `Artist - Title` filename correctly), with no network. Use it to undo bad
tags — e.g. titles/artists clobbered by an earlier `--repair` on music videos. It overwrites
artist/title and leaves album/year alone.
```bash
./musicfetch --retag-from-path -d # preview
./musicfetch --retag-from-path -o /media/music
# Skip folders (e.g. hand-curated playlists you don't want re-tagged)
./musicfetch --repair -x Unsorted -x playlists
```
### 📁 Output Structure
@@ -124,8 +238,10 @@ export LIDARR_API_KEY="your-lidarr-api-key"
```text
<root>/
├── Artist Name/
│ ├── Album Name/ (managed by Lidarr)
── youtube/ (yt-dlp downloads / fallbacks)
│ ├── Album Name/ (managed by Lidarr)
── youtube/ (YouTube / YouTube Music downloads)
│ ├── soundcloud/ (SoundCloud downloads)
│ └── <source>/ (one folder per yt-dlp source)
```
---

View File

@@ -0,0 +1,645 @@
# Playlists + Profile Hardening Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** (1) Harden Lidarr profile selection (pick metadata + quality profile by name with env overrides, not by array position). (2) Add YouTube playlist support that downloads each track to its own per-artist folder, via both the CLI and the REST API (one playlist = one job, `done` if ≥1 track succeeds).
**Architecture:** New helpers in the single-file `musicfetch` binary (profile lookup by name; `is_playlist_url`/`expand_playlist`/`download_playlist`/`download_single`; `yt_download`/`act_youtube` return success bools). `server/mf.py` re-exports the new URL helpers; `server/jobs.py` gains callable `done_message` (so a batch can report `N/M`); `server/actions.py` + `server/app.py` route URL/playlist `q` to a download job. Tests import the binary via the existing `server.mf` loader (`musicfetch_core`).
**Tech Stack:** Python 3.10+, stdlib `urllib.parse`, `requests`/`ytmusicapi`/`yt-dlp` (already deps), FastAPI, pytest+monkeypatch. No new deps.
---
## Context for the implementer
Work from `/home/zhering/Documents/musicfetch` on branch `feat/playlists-profiles` (already checked out). The `musicfetch` binary (no `.py` ext) already has, verified at these locations:
- `get_default_metadata_profile_id()` (line ~447): returns `profiles[0]["id"]` — to be replaced.
- `add_artist()` (line ~457): payload hardcodes `"qualityProfileId": 1` (line ~466) and calls `get_default_metadata_profile_id()` (line ~467).
- `yt_download(url_or_query, target_folder, quality, dry_run, hit=None)` (line ~579): builds the yt-dlp cmd, `subprocess.run(cmd)` at the end, returns None. `--no-playlist` is in the cmd.
- `act_youtube(hit, root, quality, dry_run)` (line ~611): builds `music.youtube` URL + per-first-artist folder, calls `yt_download`, returns None.
- `run_yt_dlp_get_metadata(url)` (line ~623), `get_artist_from_metadata(meta)` (line ~635), `handle_url(url, root, quality, dry_run)` (line ~644).
- `is_url(s)` (early), `Hit` dataclass, `_ytm_artists(item)` (in YouTube-search section), module-level `YTMusic` (None if not installed), `subprocess`, `json`, `os`, `requests`, `RequestException`, `dbg`, `err`, `lidarr_get`, `lidarr_post`.
`server/mf.py` re-exports a fixed symbol list + `__all__`; `server/jobs.py` has `run_job(job_id, fn, done_message, fail_message=...)` where `done_message` is currently a str; `server/app.py` `fetch()` treats `q` only as a search term; `server/actions.py` has `perform_fetch`, `started_message`, `done_message`, `failed_message`.
Tests: `import server.mf # noqa: F401` then `import musicfetch_core as mf`; monkeypatch `mf.lidarr_get`, `mf.act_youtube`, `mf.subprocess`, `mf.YTMusic`, and `monkeypatch.setenv`.
Add to the top imports block of `musicfetch` (Task 2): `from urllib.parse import urlparse, parse_qs`.
---
### Task 1: Lidarr profile hardening
**Files:**
- Modify: `musicfetch` (replace `get_default_metadata_profile_id`; add `_profile_id_by_name` and `get_quality_profile_id`; change `add_artist` payload)
- Test: `tests/test_profiles.py`
- [ ] **Step 1: Write the failing test**
Create `tests/test_profiles.py`:
```python
import server.mf # noqa: F401
import musicfetch_core as mf
META = [{"id": 1, "name": "Standard"}, {"id": 2, "name": "None"}, {"id": 3, "name": "OST"}]
QUAL = [{"id": 1, "name": "Any"}, {"id": 2, "name": "Lossless"}]
def test_metadata_profile_default_standard_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_METADATA_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1 # "Standard", not position-luck
def test_metadata_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "OST")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 3
def test_metadata_profile_unknown_name_falls_back_to_first(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "Nonexistent")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1
def test_quality_profile_default_any_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_QUALITY_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 1
def test_quality_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_QUALITY_PROFILE", "Lossless")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 2
def test_profile_fetch_error_returns_one(monkeypatch):
def boom(path, timeout=10):
raise mf.RequestException("down")
monkeypatch.setattr(mf, "lidarr_get", boom)
assert mf.get_default_metadata_profile_id() == 1
assert mf.get_quality_profile_id() == 1
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/test_profiles.py -v`
Expected: FAIL — `AttributeError: ... 'get_quality_profile_id'` and metadata env tests fail.
- [ ] **Step 3: Implement**
In `musicfetch`, replace `get_default_metadata_profile_id` with:
```python
def _profile_id_by_name(path: str, env_var: str, default_name: str) -> int:
"""Return the id of the profile whose name matches env_var (default
default_name, case-insensitive). Fall back to the first profile, then 1."""
name = os.environ.get(env_var, default_name)
try:
profiles = lidarr_get(path, timeout=10)
except RequestException as e:
dbg(f"{path} fetch failed: {e}")
return 1
if not profiles:
return 1
for p in profiles:
if p.get("name", "").casefold() == name.casefold():
return p["id"]
dbg(f"profile '{name}' not found at {path}; using first ('{profiles[0].get('name')}')")
return profiles[0]["id"]
def get_default_metadata_profile_id() -> int:
return _profile_id_by_name("/api/v1/metadataprofile", "LIDARR_METADATA_PROFILE", "Standard")
def get_quality_profile_id() -> int:
return _profile_id_by_name("/api/v1/qualityprofile", "LIDARR_QUALITY_PROFILE", "Any")
```
In `add_artist`, change the payload line `"qualityProfileId": 1,` to:
```python
"qualityProfileId": get_quality_profile_id(),
```
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/test_profiles.py -v`
Expected: PASS (6 passed)
- [ ] **Step 5: Commit**
```bash
git add musicfetch tests/test_profiles.py
git commit -m "fix(lidarr): select metadata/quality profiles by name with env overrides"
```
---
### Task 2: Playlist core in `musicfetch`
**Files:**
- Modify: `musicfetch` (add `from urllib.parse import urlparse, parse_qs`; add `is_playlist_url`, `_playlist_id`, `expand_playlist`, `download_playlist`, `download_single`; make `yt_download` + `act_youtube` return a success bool; rewrite `handle_url`)
- Test: `tests/test_playlist.py`
- [ ] **Step 1: Write the failing test**
Create `tests/test_playlist.py`:
```python
import server.mf # noqa: F401
import musicfetch_core as mf
# ---- is_playlist_url ----
def test_pure_playlist_url_is_playlist():
assert mf.is_playlist_url("https://music.youtube.com/playlist?list=PLabc") is True
assert mf.is_playlist_url("https://www.youtube.com/playlist?list=PLabc") is True
def test_watch_with_list_is_not_playlist():
assert mf.is_playlist_url("https://www.youtube.com/watch?v=abc&list=PLx") is False
def test_plain_watch_is_not_playlist():
assert mf.is_playlist_url("https://www.youtube.com/watch?v=abc") is False
def test_non_url_is_not_playlist():
assert mf.is_playlist_url("Daft Punk - Discovery") is False
# ---- expand_playlist (yt-dlp fallback path) ----
class _CP:
def __init__(self, stdout):
self.stdout = stdout
self.returncode = 0
def test_expand_playlist_ytdlp_fallback(monkeypatch):
import json as _json
monkeypatch.setattr(mf, "YTMusic", None) # force yt-dlp path
payload = {"title": "My Mix", "entries": [
{"id": "v1", "title": "Song One", "uploader": "Artist A"},
{"id": "v2", "title": "Song Two", "channel": "Artist B"},
{"id": None, "title": "skip"},
]}
monkeypatch.setattr(mf.subprocess, "run",
lambda *a, **k: _CP(_json.dumps(payload)))
title, hits = mf.expand_playlist("https://www.youtube.com/playlist?list=PLx")
assert title == "My Mix"
assert [h.payload["videoId"] for h in hits] == ["v1", "v2"]
assert hits[0].artist == "Artist A"
# ---- download_playlist ----
def test_download_playlist_counts_ok_and_total(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
h3 = mf.Hit(source="youtube", kind="track", title="C", artist="Z", payload={"videoId": "3"})
monkeypatch.setattr(mf, "expand_playlist", lambda url: ("PL Title", [h1, h2, h3]))
def fake_act(hit, root, quality, dry_run):
return hit.title != "B" # B "fails"
monkeypatch.setattr(mf, "act_youtube", fake_act)
ok, total, title = mf.download_playlist("u", "/tmp", "best", False)
assert (ok, total, title) == (2, 3, "PL Title")
def test_download_playlist_track_exception_counts_as_failure(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
monkeypatch.setattr(mf, "expand_playlist", lambda url: ("T", [h1, h2]))
def fake_act(hit, root, quality, dry_run):
if hit.title == "B":
raise RuntimeError("boom")
return True
monkeypatch.setattr(mf, "act_youtube", fake_act)
ok, total, _ = mf.download_playlist("u", "/tmp", "best", False)
assert (ok, total) == (1, 2)
# ---- yt_download returns success bool ----
def test_yt_download_returns_true_on_zero_exit(monkeypatch):
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
monkeypatch.setattr(mf.subprocess, "run", lambda *a, **k: _CP("")) # returncode 0
assert mf.yt_download("u", "/tmp/x", "best", False) is True
def test_yt_download_dry_run_returns_true(monkeypatch):
assert mf.yt_download("u", "/tmp/x", "best", True) is True
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/test_playlist.py -v`
Expected: FAIL — `AttributeError: ... 'is_playlist_url'`.
- [ ] **Step 3: Implement**
Add to the top imports block of `musicfetch`:
```python
from urllib.parse import urlparse, parse_qs
```
Make `yt_download` return a success bool. Change its tail (the `if dry_run:` block and the final `subprocess.run`) to:
```python
if dry_run:
print(f"[dry-run] mkdir -p {target_folder}")
print(f"[dry-run] {' '.join(cmd)}")
return True
os.makedirs(target_folder, exist_ok=True)
print(f"Downloading via yt-dlp -> {target_folder}")
return subprocess.run(cmd).returncode == 0
```
Make `act_youtube` return the bool — change its last line `yt_download(url, target, quality, dry_run, hit=hit)` to:
```python
return yt_download(url, target, quality, dry_run, hit=hit)
```
Add the playlist functions (place them in the URL-path section, after `handle_url`'s helpers / near `handle_url`):
```python
def _playlist_id(url: str) -> str:
return parse_qs(urlparse(url).query).get("list", [""])[0]
def is_playlist_url(url: str) -> bool:
"""True for a pure playlist URL (/playlist?list=… or list= without v=).
A watch?v=…&list=… URL is treated as a single track, not a batch."""
if not is_url(url):
return False
parsed = urlparse(url)
qs = parse_qs(parsed.query)
if "/playlist" in parsed.path:
return True
return "list" in qs and "v" not in qs
def expand_playlist(url: str) -> tuple[str, list[Hit]]:
"""Return (playlist_title, [track Hits]). Prefer ytmusicapi; fall back to
yt-dlp --flat-playlist. Returns ("", []) on failure."""
pid = _playlist_id(url)
if YTMusic is not None and pid:
try:
pl = YTMusic().get_playlist(pid, limit=None)
hits = []
for t in pl.get("tracks", []):
vid = t.get("videoId")
if not vid:
continue
alb = t.get("album")
album = alb.get("name", "") if isinstance(alb, dict) else (alb or "")
hits.append(Hit(source="youtube", kind="track", title=t.get("title", ""),
artist=_ytm_artists(t), album=album,
year=str(t.get("year") or ""), payload={"videoId": vid}))
if hits:
return pl.get("title", ""), hits
except Exception as e: # noqa: BLE001
dbg(f"ytmusicapi playlist expand failed: {e}")
try:
result = subprocess.run(["yt-dlp", "--flat-playlist", "-J", url],
capture_output=True, text=True, check=True)
data = json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
err(f"yt-dlp playlist expand failed: {e}")
return "", []
hits = []
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 "",
payload={"videoId": vid}))
return data.get("title", ""), hits
def download_playlist(url: str, root: str, quality: str, dry_run: bool) -> tuple[int, int, str]:
"""Download each playlist track via act_youtube. Returns (ok, total, title)."""
title, hits = expand_playlist(url)
ok = 0
for h in hits:
try:
if act_youtube(h, root, quality, dry_run):
ok += 1
except Exception as e: # noqa: BLE001 — one bad track shouldn't abort the batch
err(f"track failed ({h.title}): {e}")
return ok, len(hits), title
def download_single(url: str, root: str, quality: str, dry_run: bool) -> dict:
"""Download a single URL. Returns {title, artist, ok}."""
meta = run_yt_dlp_get_metadata(url)
artist = get_artist_from_metadata(meta) if meta else "Unknown Artist"
title = (meta or {}).get("title", "")
target = os.path.join(root, artist, "youtube")
ok = yt_download(url, target, quality, dry_run)
return {"title": title, "artist": artist, "ok": ok}
```
Rewrite `handle_url` to route playlists:
```python
def handle_url(url: str, root: str, quality: str, dry_run: bool):
if is_playlist_url(url):
ok, total, title = download_playlist(url, root, quality, dry_run)
label = f" from '{title}'" if title else ""
print(f"Downloaded {ok}/{total} tracks{label}")
return
download_single(url, root, quality, dry_run)
```
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/test_playlist.py -v`
Expected: PASS (9 passed)
- [ ] **Step 5: Full suite + compile**
Run: `pytest -q` (prior 43 + 6 profiles + 9 playlist = 58) and `python3 -m py_compile musicfetch`.
Expected: all green, clean compile.
- [ ] **Step 6: Commit**
```bash
git add musicfetch tests/test_playlist.py
git commit -m "feat(youtube): playlist expansion + per-track download, success bools"
```
---
### Task 3: Re-exports + callable job message
**Files:**
- Modify: `server/mf.py` (re-export new URL helpers)
- Modify: `server/jobs.py` (`run_job` accepts a callable `done_message`)
- Test: `tests/test_jobs.py` (add a callable-message test)
- [ ] **Step 1: Write the failing test**
Append to `tests/test_jobs.py`:
```python
def test_run_job_callable_done_message():
job = jobs.create_job(hit={}, message="m")
jobs.run_job(job.id, lambda: {"ok": 2, "total": 3},
done_message=lambda res: f"{res['ok']}/{res['total']} done")
j = _wait(job.id, "done")
assert j.message == "2/3 done"
```
Also add a re-export check — create `tests/test_mf_url_exports.py`:
```python
import server.mf as smf
def test_url_helpers_reexported():
assert callable(smf.is_url)
assert callable(smf.is_playlist_url)
assert callable(smf.download_playlist)
assert callable(smf.download_single)
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_jobs.py::test_run_job_callable_done_message tests/test_mf_url_exports.py -v`
Expected: FAIL (callable message not supported; `smf.is_playlist_url` missing).
- [ ] **Step 3: Implement**
In `server/jobs.py`, inside `run_job`'s `_task`, change the success branch to support a callable:
```python
result = fn()
msg = done_message(result) if callable(done_message) else done_message
_touch(job, status="done", result=result, message=msg)
```
(Update the `run_job` signature/type hint to `done_message` being `str | Callable[[dict], str]`; import `Callable` is already present.)
In `server/mf.py`, add to the re-export assignments and `__all__`:
```python
is_url = _mod.is_url
is_playlist_url = _mod.is_playlist_url
download_playlist = _mod.download_playlist
download_single = _mod.download_single
```
Add those four names to the `__all__` list.
- [ ] **Step 4: Run tests to verify they pass**
Run: `pytest tests/test_jobs.py tests/test_mf_url_exports.py -v`
Expected: PASS.
- [ ] **Step 5: Commit**
```bash
git add server/jobs.py server/mf.py tests/test_jobs.py tests/test_mf_url_exports.py
git commit -m "feat(server): re-export URL helpers; callable job done_message"
```
---
### Task 4: REST API URL/playlist routing
**Files:**
- Modify: `server/actions.py` (add `url_started_message`, `url_done_message`, `playlist_done_message`, `perform_url_fetch`)
- Modify: `server/app.py` (route URL `q` to a download job)
- Test: `tests/test_api_url.py`
- [ ] **Step 1: Write the failing test**
Create `tests/test_api_url.py`:
```python
import time
import pytest
from server import jobs as jobs_mod
@pytest.fixture(autouse=True)
def _clear_jobs():
jobs_mod.JOBS.clear()
yield
jobs_mod.JOBS.clear()
def _wait_done(client, auth, job_id, timeout=2.0):
end = time.time() + timeout
while time.time() < end:
b = client.get(f"/jobs/{job_id}", headers=auth).json()
if b["status"] in ("done", "failed"):
return b
time.sleep(0.01)
raise AssertionError("job never finished")
def test_playlist_url_batch_job(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_playlist",
lambda url, root, quality, dry_run: (2, 3, "My Mix"))
r = client.post("/fetch", params={"q": "https://music.youtube.com/playlist?list=PLx"}, headers=auth)
assert r.status_code == 200
body = r.json()
assert body["status"] == "queued"
assert body["hit"]["kind"] == "playlist"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "2/3" in done["message"]
assert done["result"]["ok"] == 2
def test_playlist_zero_success_fails(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_playlist",
lambda url, root, quality, dry_run: (0, 3, "Dead Mix"))
body = client.post("/fetch", params={"q": "https://www.youtube.com/playlist?list=PLy"}, headers=auth).json()
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "failed"
def test_single_video_url_download(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.download_single",
lambda url, root, quality, dry_run: {"title": "Song", "artist": "A", "ok": True})
body = client.post("/fetch", params={"q": "https://music.youtube.com/watch?v=abc"}, headers=auth).json()
assert body["hit"]["kind"] == "track"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "Song" in done["message"]
def test_search_query_still_works(client, auth, monkeypatch):
from server import mf
hit = mf.Hit(source="youtube", kind="track", title="T", artist="A", payload={"videoId": "x"})
monkeypatch.setattr("server.app.mf.build_combined_hits",
lambda q, limit, yt_first, lidarr_only, yt_only: [hit])
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr("server.app.actions.perform_fetch",
lambda chosen, hits, quality, root: {"path": "/x", "lidarr_album_id": None})
r = client.post("/fetch", params={"q": "Daft Punk - Discovery"}, headers=auth)
assert r.status_code == 200
assert r.json()["status"] == "queued"
```
- [ ] **Step 2: Run test to verify it fails**
Run: `pytest tests/test_api_url.py -v`
Expected: FAIL (URL `q` currently goes to search → no `download_playlist`/`download_single` calls; kind not "playlist").
- [ ] **Step 3: Implement**
In `server/actions.py`, add:
```python
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, quality: str, root: str) -> dict:
"""Download a URL (playlist → batch, else single). Raises if nothing
downloaded so the job is marked failed."""
if mf.is_playlist_url(url):
ok, total, title = mf.download_playlist(url, 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["title"], "artist": info["artist"],
"ok": 1, "total": 1, "path": None, "lidarr_album_id": None}
```
In `server/app.py` `fetch()`, add a URL branch BEFORE the search logic (after the `quality` validation; keep the existing `quality not in mf.QUALITY_CHOICES` 422 check above it). Insert:
```python
if mf.is_url(q):
kind = "playlist" if mf.is_playlist_url(q) else "track"
syn = mf.Hit(source="youtube", kind=kind, title="", artist="")
job = jobs.create_job(hit=syn, message=actions.url_started_message(kind))
response = _job_public(job)
done_msg = actions.playlist_done_message if kind == "playlist" else actions.url_done_message
jobs.run_job(
job.id,
lambda: actions.perform_url_fetch(q, quality, ROOT),
done_message=done_msg,
fail_message="Download failed.",
)
return response
```
(The existing `source` validation can stay; it's ignored for URLs. Leave the search path untouched below this branch.)
- [ ] **Step 4: Run test to verify it passes**
Run: `pytest tests/test_api_url.py -v`
Expected: PASS (4 passed)
- [ ] **Step 5: Full suite**
Run: `pytest -q`
Expected: all green (58 + callable/export + 4 api-url ≈ 64).
- [ ] **Step 6: Commit**
```bash
git add server/actions.py server/app.py tests/test_api_url.py
git commit -m "feat(server): route URL/playlist /fetch to download jobs"
```
---
### Task 5: Live verification
**Files:** none (controller-run).
- [ ] **Step 1: Profiles** — read-only confirm name selection against the real Lidarr:
```bash
cd /home/zhering/Documents/musicfetch
env LIDARR_URL=http://10.2.1.16:8686 LIDARR_API_KEY=49cf02acb4c7436b842df2150056d468 \
python3 -c "import server.mf, musicfetch_core as mf; print('meta', mf.get_default_metadata_profile_id(), 'qual', mf.get_quality_profile_id())"
```
Expected: `meta 1 qual 1` (Standard / Any). Then with `LIDARR_METADATA_PROFILE=OST``meta 3`.
- [ ] **Step 2: Playlist (CLI dry-run)** — confirm expansion + per-track routing without downloading. Pick a small real YT Music playlist URL:
```bash
./musicfetch -d "<small-playlist-url>"
```
Expected: prints a `[dry-run] yt-dlp …` line per track, each targeting `/media/music/<artist>/youtube`.
- [ ] **Step 3: Playlist (real, small)** — with user approval, run the API against a 3-5 track playlist:
```bash
fuser -k 6769/tcp 2>/dev/null; sleep 1
env MUSICFETCH_API_KEY=testkey MUSICFETCH_ROOT=/tmp \
python3 -m uvicorn server.app:app --host 127.0.0.1 --port 6769 --log-level warning &
sleep 4
curl -s -X POST 'http://127.0.0.1:6769/fetch?q=<small-playlist-url>' -H 'X-API-Key: testkey'
# poll /jobs/{id} → expect "Downloaded N/M tracks", files under /tmp/<artist>/youtube/
fuser -k 6769/tcp 2>/dev/null
```
---
## Self-Review
**Spec coverage:**
- Profile-by-name + env overrides + add_artist uses quality profile → Task 1. ✅
- `is_playlist_url` (watch?v&list → single) → Task 2. ✅
- `expand_playlist` (ytmusicapi → yt-dlp fallback) → Task 2. ✅
- `download_playlist` per-track via `act_youtube`, ok/total counting, per-track failures caught → Task 2. ✅
- `yt_download`/`act_youtube` success bools → Task 2. ✅
- CLI `handle_url` playlist routing → Task 2. ✅
- Re-exports + callable batch message → Task 3. ✅
- API URL routing, playlist batch job, `done` if ok≥1 else `failed`, single-URL job, Siri messages, search path unchanged → Task 4. ✅
- Live checks (profiles + playlist) → Task 5. ✅
- Out-of-scope (per-track fan-out, resume/dedup) excluded. ✅
**Placeholder scan:** none — all code/tests complete (the only `<…>` are real user-supplied URLs in the manual Task 5 steps).
**Type consistency:** `download_playlist -> (int,int,str)` consumed as `(ok,total,title)` in CLI + `perform_url_fetch`. `download_single -> {title,artist,ok}` consumed in `perform_url_fetch`. `yt_download`/`act_youtube` now return bool; `act_youtube`'s only other caller (`actions.perform_fetch._download_youtube` in the existing search path) ignores the return value — unaffected. `run_job(done_message)` accepts str or `Callable[[dict],str]`; existing search-path callers pass str (unchanged). `_profile_id_by_name(path, env_var, default_name)` used by both profile getters. New `mf.py` exports (`is_url`, `is_playlist_url`, `download_playlist`, `download_single`) match the names used in `server/app.py` and `server/actions.py`.

View File

@@ -0,0 +1,743 @@
# Odesli Link Resolution Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Let users hand MusicFetch any song link (Spotify/Apple/Tidal/…); resolve it to metadata via Odesli, then run the existing Lidarr-first, exact-YouTube-fallback flow. YouTube/SoundCloud links and text queries are unchanged.
**Architecture:** Add three core helpers to the `musicfetch` script — `odesli_resolve` (HTTP → metadata), `_is_direct_url` (router), and `resolve_link_hits`/`handle_link` (turn a non-direct link into a `"Artist - Title"` query + Lidarr hits + the exact YouTube track, then reuse the existing pick/dispatch). Wire CLI `main()` and server `/fetch` to route non-direct URLs through it. The server stays a thin reuse layer via `server/mf.py`.
**Tech Stack:** Python 3.10+, `requests`, `pytest`, FastAPI (server). Tests import the extension-less `musicfetch` script as `musicfetch_core` via `import server.mf`.
---
## File Structure
- `musicfetch` — add `Resolved` dataclass, `OdesliError`, `odesli_resolve`, `_is_direct_url`, `resolve_link_hits`, `handle_link`, and extract `_dispatch_chosen` from `main()`. Wire `main()`.
- `server/mf.py` — export the new symbols.
- `server/app.py` — route non-direct URLs in `/fetch` through resolve + existing `perform_fetch` job path.
- `tests/test_odesli.py` — new: `odesli_resolve`, `_is_direct_url`, `resolve_link_hits`, `handle_link`.
- `tests/test_api_url.py` — add: server non-direct-URL routing.
- `tests/test_mf_url_exports.py` — add: new symbols re-exported.
---
## Task 1: `Resolved` dataclass, `OdesliError`, and `odesli_resolve`
**Files:**
- Modify: `musicfetch` (add dataclass + exception near the `Hit` dataclass ~line 82; add `quote` to the `urllib.parse` import line 23; add function in the URL section)
- Test: `tests/test_odesli.py`
- [ ] **Step 1: Write the failing tests**
Create `tests/test_odesli.py`:
```python
import server.mf # noqa: F401 — loads musicfetch, registers musicfetch_core
import musicfetch_core as mf
class _FakeResp:
def __init__(self, status, payload):
self.status_code = status
self._payload = payload
def json(self):
return self._payload
FULL = {
"entityUniqueId": "SPOTIFY_SONG::abc",
"entitiesByUniqueId": {
"SPOTIFY_SONG::abc": {
"title": "Bloom",
"artistName": "ODESZA",
"thumbnailUrl": "https://img/cover.jpg",
}
},
"linksByPlatform": {
"youtubeMusic": {"url": "https://music.youtube.com/watch?v=YYY"},
"youtube": {"url": "https://youtube.com/watch?v=YYY"},
},
}
def test_odesli_resolve_full(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, FULL))
r = mf.odesli_resolve("https://open.spotify.com/track/abc")
assert r.title == "Bloom"
assert r.artist == "ODESZA"
assert r.thumb == "https://img/cover.jpg"
assert r.youtube_url == "https://music.youtube.com/watch?v=YYY"
def test_odesli_resolve_prefers_ytmusic_then_youtube(monkeypatch):
payload = {**FULL, "linksByPlatform": {"youtube": {"url": "https://youtube.com/watch?v=ZZZ"}}}
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, payload))
r = mf.odesli_resolve("https://open.spotify.com/track/abc")
assert r.youtube_url == "https://youtube.com/watch?v=ZZZ"
def test_odesli_resolve_no_youtube_link(monkeypatch):
payload = {**FULL, "linksByPlatform": {}}
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, payload))
r = mf.odesli_resolve("https://open.spotify.com/track/abc")
assert r.youtube_url == ""
assert r.title == "Bloom"
def test_odesli_resolve_non_200_returns_none(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(429, {}))
assert mf.odesli_resolve("https://open.spotify.com/track/abc") is None
def test_odesli_resolve_malformed_returns_none(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, {"nope": 1}))
assert mf.odesli_resolve("https://open.spotify.com/track/abc") is None
def test_odesli_resolve_network_error_returns_none(monkeypatch):
def boom(*a, **k):
raise mf.RequestException("down")
monkeypatch.setattr(mf.requests, "get", boom)
assert mf.odesli_resolve("https://open.spotify.com/track/abc") is None
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_odesli.py -v`
Expected: FAIL — `AttributeError: module 'musicfetch_core' has no attribute 'odesli_resolve'`.
- [ ] **Step 3: Implement**
In `musicfetch`, change the import on line 23 to add `quote`:
```python
from urllib.parse import urlparse, parse_qs, quote
```
Add after the `Hit` dataclass (after its `display_title` property, ~line 96):
```python
@dataclass
class Resolved:
title: str = ""
artist: str = ""
thumb: str = ""
youtube_url: str = ""
class OdesliError(Exception):
"""Raised when an Odesli link can't be resolved to usable metadata."""
ODESLI_URL = "https://api.song.link/v1-alpha.1/links"
```
Add in the URL section (e.g. just above `def probe_url`, ~line 765):
```python
def odesli_resolve(url: str) -> Optional[Resolved]:
"""Resolve any streaming link to {title, artist, thumb, youtube_url} via the
Odesli (song.link) public API. Returns None on any failure (network, non-200,
malformed body, missing title+artist) so callers can fall back loudly."""
try:
resp = requests.get(ODESLI_URL,
params={"url": url, "userCountry": "US"},
timeout=8)
if resp.status_code != 200:
dbg(f"odesli {resp.status_code} for {url}")
return None
data = resp.json()
entity = data["entitiesByUniqueId"][data["entityUniqueId"]]
title = entity.get("title", "")
artist = entity.get("artistName", "")
if not title and not artist:
return None
platforms = data.get("linksByPlatform", {})
yt = (platforms.get("youtubeMusic") or platforms.get("youtube") or {}).get("url", "")
return Resolved(title=title, artist=artist,
thumb=entity.get("thumbnailUrl", ""), youtube_url=yt)
except (RequestException, ValueError, KeyError, TypeError) as e:
dbg(f"odesli resolve failed for {url}: {e}")
return None
```
(`quote` is imported for completeness/consistency; `requests` handles encoding via `params`. Leave `quote` unused-import-free by relying on `params` — if a linter complains, drop the `quote` addition.)
Correction: do **not** add `quote` to the import — `params=` already encodes. Keep line 23 unchanged.
- [ ] **Step 4: Run tests to verify they pass**
Run: `pytest tests/test_odesli.py -v`
Expected: 6 passed.
- [ ] **Step 5: Commit**
```bash
git add musicfetch tests/test_odesli.py
git commit -m "feat: odesli_resolve — resolve any song link to metadata via song.link
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 2: `_is_direct_url` router
**Files:**
- Modify: `musicfetch` (add near `_is_youtube_playlist_url`, ~line 720)
- Test: `tests/test_odesli.py`
- [ ] **Step 1: Write the failing tests**
Append to `tests/test_odesli.py`:
```python
def test_is_direct_url_youtube():
assert mf._is_direct_url("https://music.youtube.com/watch?v=abc")
assert mf._is_direct_url("https://www.youtube.com/watch?v=abc")
assert mf._is_direct_url("https://youtu.be/abc")
def test_is_direct_url_soundcloud():
assert mf._is_direct_url("https://soundcloud.com/dj/track")
def test_is_direct_url_other_platforms_false():
assert not mf._is_direct_url("https://open.spotify.com/track/abc")
assert not mf._is_direct_url("https://music.apple.com/us/album/x/1?i=2")
assert not mf._is_direct_url("https://tidal.com/browse/track/123")
def test_is_direct_url_youtube_playlist_true():
assert mf._is_direct_url("https://www.youtube.com/playlist?list=PLabc")
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_odesli.py -k is_direct_url -v`
Expected: FAIL — `AttributeError: ... has no attribute '_is_direct_url'`.
- [ ] **Step 3: Implement**
Add after `_is_youtube_playlist_url` (~line 732):
```python
_DIRECT_HOSTS = ("youtube.com", "youtu.be", "music.youtube.com",
"soundcloud.com", "api.soundcloud.com")
def _is_direct_url(url: str) -> bool:
"""True for links yt-dlp downloads well directly (YouTube, SoundCloud).
These skip Odesli resolution and use the existing handle_url path."""
if not is_url(url):
return False
host = (urlparse(url).hostname or "").lower()
if host.startswith("www."):
host = host[4:]
if host.endswith(("youtube.com", "youtu.be", "soundcloud.com")):
return True
return host in _DIRECT_HOSTS
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `pytest tests/test_odesli.py -k is_direct_url -v`
Expected: 4 passed.
- [ ] **Step 5: Commit**
```bash
git add musicfetch tests/test_odesli.py
git commit -m "feat: _is_direct_url — route YouTube/SoundCloud links to direct download
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 3: Extract `_dispatch_chosen` from `main()`
Refactor only — pulls the chosen-Hit dispatch out of `main()` so `handle_link` can reuse it. Behavior unchanged; existing tests must stay green.
**Files:**
- Modify: `musicfetch` `main()` (lines ~1268-1284) and add `_dispatch_chosen` above `main()`.
- [ ] **Step 1: Add `_dispatch_chosen` above `main()`**
```python
def _dispatch_chosen(chosen: Hit, hits: list[Hit], root: str, quality: str,
dry_run: bool, lidarr_only: bool) -> None:
"""Act on a picked Hit: Lidarr album (add+search, fall to top YouTube hit on
no release), Lidarr artist, or a YouTube track. Shared by main() and handle_link."""
if chosen.source == "lidarr":
if chosen.kind == "album":
handled = act_lidarr_album(chosen, root, False, dry_run)
if not handled and not lidarr_only:
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, root, quality, dry_run)
else:
print("No YouTube fallback available.")
else:
act_lidarr_artist(chosen, root, False, dry_run)
else:
act_youtube(chosen, root, quality, dry_run)
```
Note: the original `main()` passed `args.search_all` to `act_lidarr_album`/`act_lidarr_artist`. Preserve that — add a `search_all` parameter:
```python
def _dispatch_chosen(chosen, hits, root, quality, dry_run, lidarr_only, search_all):
if chosen.source == "lidarr":
if chosen.kind == "album":
handled = act_lidarr_album(chosen, root, search_all, dry_run)
if not handled and not lidarr_only:
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, root, quality, dry_run)
else:
print("No YouTube fallback available.")
else:
act_lidarr_artist(chosen, root, search_all, dry_run)
else:
act_youtube(chosen, root, quality, dry_run)
```
- [ ] **Step 2: Replace the dispatch block in `main()`**
Replace lines ~1271-1284 (the `if chosen.source == "lidarr": ... else: act_youtube(...)` block) with:
```python
_dispatch_chosen(chosen, hits, args.root, args.quality, args.dry_run,
args.lidarr_only, args.search_all)
```
- [ ] **Step 3: Run the full suite to confirm no regressions**
Run: `pytest -q`
Expected: all existing tests pass (no behavior change).
- [ ] **Step 4: Commit**
```bash
git add musicfetch
git commit -m "refactor: extract _dispatch_chosen from main() for reuse
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 4: `resolve_link_hits` and `handle_link`
**Files:**
- Modify: `musicfetch` (add near `handle_url`, ~line 845)
- Test: `tests/test_odesli.py`
- [ ] **Step 1: Write the failing tests**
Append to `tests/test_odesli.py`:
```python
def _resolved(yt="https://music.youtube.com/watch?v=YYY"):
return mf.Resolved(title="Bloom", artist="ODESZA",
thumb="https://img/cover.jpg", youtube_url=yt)
def test_resolve_link_hits_builds_query_and_exact_yt(monkeypatch):
monkeypatch.setattr(mf, "odesli_resolve", lambda url: _resolved())
lid = mf.Hit(source="lidarr", kind="album", title="A Moment Apart",
artist="ODESZA", album="A Moment Apart", year="2017",
payload={"album": {"id": 9}})
monkeypatch.setattr(mf, "lidarr_search", lambda q, limit: [lid])
query, hits = mf.resolve_link_hits("https://open.spotify.com/track/abc", 10)
assert query == "ODESZA - Bloom"
assert hits[0].source == "lidarr"
yt = hits[-1]
assert yt.source == "youtube" and yt.kind == "track"
assert yt.title == "Bloom" and yt.artist == "ODESZA"
assert yt.payload["url"] == "https://music.youtube.com/watch?v=YYY"
def test_resolve_link_hits_no_yt_link_lidarr_only(monkeypatch):
monkeypatch.setattr(mf, "odesli_resolve", lambda url: _resolved(yt=""))
lid = mf.Hit(source="lidarr", kind="album", title="X", artist="ODESZA",
payload={"album": {"id": 9}})
monkeypatch.setattr(mf, "lidarr_search", lambda q, limit: [lid])
query, hits = mf.resolve_link_hits("https://open.spotify.com/track/abc", 10)
assert all(h.source == "lidarr" for h in hits)
def test_resolve_link_hits_odesli_miss_raises(monkeypatch):
monkeypatch.setattr(mf, "odesli_resolve", lambda url: None)
import pytest
with pytest.raises(mf.OdesliError):
mf.resolve_link_hits("https://open.spotify.com/track/abc", 10)
def test_handle_link_miss_prints_and_returns(monkeypatch, capsys):
monkeypatch.setattr(mf, "odesli_resolve", lambda url: None)
mf.handle_link("https://open.spotify.com/track/abc", "/m", "best",
False, True, False, 10)
out = capsys.readouterr()
assert "Couldn't resolve" in (out.err + out.out)
def test_handle_link_dispatches_chosen(monkeypatch):
monkeypatch.setattr(mf, "odesli_resolve", lambda url: _resolved())
monkeypatch.setattr(mf, "lidarr_search", lambda q, limit: [])
chosen = {}
monkeypatch.setattr(mf, "pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr(mf, "_dispatch_chosen",
lambda c, hits, root, quality, dry, lo, sa: chosen.update(c=c, root=root))
mf.handle_link("https://open.spotify.com/track/abc", "/m", "best",
False, True, False, 10)
assert chosen["c"].source == "youtube"
assert chosen["root"] == "/m"
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `pytest tests/test_odesli.py -k "resolve_link_hits or handle_link" -v`
Expected: FAIL — `AttributeError: ... has no attribute 'resolve_link_hits'`.
- [ ] **Step 3: Implement**
Add near `handle_url` (~line 845) in `musicfetch`:
```python
def resolve_link_hits(url: str, limit: int) -> tuple[str, list[Hit]]:
"""Resolve a non-YouTube/SoundCloud link via Odesli into a search query plus
hits: Lidarr album candidates for "Artist - Title", followed by the EXACT
YouTube track from the shared link (not a fuzzy re-search). Raises OdesliError
if the link can't be resolved."""
r = odesli_resolve(url)
if r is None:
raise OdesliError(url)
query = f"{r.artist} - {r.title}".strip(" -")
hits = lidarr_search(query, limit)
if r.youtube_url:
hits = hits + [Hit(source="youtube", kind="track", title=r.title,
artist=r.artist, thumbnail=r.thumb,
payload={"url": r.youtube_url})]
return query, hits
def handle_link(url: str, root: str, quality: str, dry_run: bool,
noninteractive: bool, yt_first: bool, limit: int) -> None:
"""CLI path for a non-direct link: resolve via Odesli, then run the normal
Lidarr-first pick/dispatch with the exact YouTube track as fallback."""
try:
query, hits = resolve_link_hits(url, limit)
except OdesliError:
err(f"Couldn't resolve {url}. Try the direct YouTube/SoundCloud link.")
return
if not hits:
err(f"No Lidarr or YouTube source found for '{query}'.")
return
chosen = pick(hits, query, noninteractive, yt_first)
if not chosen:
print("Nothing selected.")
return
_dispatch_chosen(chosen, hits, root, quality, dry_run,
lidarr_only=False, search_all=False)
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `pytest tests/test_odesli.py -v`
Expected: all pass.
- [ ] **Step 5: Commit**
```bash
git add musicfetch tests/test_odesli.py
git commit -m "feat: resolve_link_hits + handle_link — Odesli link -> Lidarr-first flow
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 5: Wire CLI `main()`
**Files:**
- Modify: `musicfetch` `main()` (the `if is_url(query):` block, ~line 1255)
- [ ] **Step 1: Replace the URL branch**
Replace:
```python
if is_url(query):
handle_url(query, args.root, args.quality, args.dry_run)
return
```
with:
```python
if is_url(query):
if _is_direct_url(query):
handle_url(query, args.root, args.quality, args.dry_run)
else:
handle_link(query, args.root, args.quality, args.dry_run,
args.noninteractive, args.ytsearch, args.limit)
return
```
- [ ] **Step 2: Manual smoke (dry-run, no network side effects expected on resolve failure)**
Run: `./musicfetch -d "https://open.spotify.com/track/0000000000" 2>&1 | head -5`
Expected: prints `Couldn't resolve https://open.spotify.com/track/0000000000. ...` (invalid id → Odesli miss) OR resolves and shows a picker/dry-run line. Either is acceptable — confirms routing, no traceback.
- [ ] **Step 3: Run the full suite**
Run: `pytest -q`
Expected: all pass.
- [ ] **Step 4: Commit**
```bash
git add musicfetch
git commit -m "feat: route non-direct CLI links through Odesli (handle_link)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 6: Export new symbols via `server/mf.py`
**Files:**
- Modify: `server/mf.py`
- Test: `tests/test_mf_url_exports.py`
- [ ] **Step 1: Write the failing test**
Append to `tests/test_mf_url_exports.py`:
```python
def test_odesli_symbols_reexported():
import server.mf as smf
assert callable(smf._is_direct_url)
assert callable(smf.odesli_resolve)
assert callable(smf.resolve_link_hits)
assert callable(smf.handle_link)
```
- [ ] **Step 2: Run to verify it fails**
Run: `pytest tests/test_mf_url_exports.py -v`
Expected: FAIL — `AttributeError: module 'server.mf' has no attribute '_is_direct_url'`.
- [ ] **Step 3: Implement**
In `server/mf.py`, after the existing `download_single = _mod.download_single` line, add:
```python
_is_direct_url = _mod._is_direct_url
odesli_resolve = _mod.odesli_resolve
resolve_link_hits = _mod.resolve_link_hits
handle_link = _mod.handle_link
OdesliError = _mod.OdesliError
```
And extend `__all__` to include `"_is_direct_url"`, `"odesli_resolve"`, `"resolve_link_hits"`, `"handle_link"`, `"OdesliError"`.
- [ ] **Step 4: Run to verify it passes**
Run: `pytest tests/test_mf_url_exports.py -v`
Expected: pass.
- [ ] **Step 5: Commit**
```bash
git add server/mf.py tests/test_mf_url_exports.py
git commit -m "feat: re-export odesli symbols through server/mf.py
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 7: Wire server `/fetch` non-direct URL branch
**Files:**
- Modify: `server/app.py` `fetch()` (the `if mf.is_url(q):` block)
- Test: `tests/test_api_url.py`
- [ ] **Step 1: Write the failing tests**
Append to `tests/test_api_url.py`:
```python
def test_non_direct_link_resolves_and_fetches(client, auth, monkeypatch):
from server import mf
lid = mf.Hit(source="lidarr", kind="album", title="A Moment Apart",
artist="ODESZA", album="A Moment Apart", year="2017",
payload={"album": {"id": 9}})
yt = mf.Hit(source="youtube", kind="track", title="Bloom", artist="ODESZA",
payload={"url": "https://music.youtube.com/watch?v=YYY"})
monkeypatch.setattr("server.app.mf.resolve_link_hits",
lambda url, limit: ("ODESZA - Bloom", [lid, yt]))
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr("server.app.actions.perform_fetch",
lambda chosen, hits, quality, root: {"path": None, "lidarr_album_id": 9})
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
assert r.status_code == 200
body = r.json()
assert body["status"] == "queued"
assert body["hit"]["album"] == "A Moment Apart"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
def test_non_direct_link_resolve_failure_422(client, auth, monkeypatch):
def boom(url, limit):
raise mf_mod.OdesliError(url)
from server import mf as mf_mod
monkeypatch.setattr("server.app.mf.resolve_link_hits", boom)
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/bad"}, headers=auth)
assert r.status_code == 422
assert "resolve" in r.json()["message"].lower()
def test_non_direct_link_no_hits_404(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.resolve_link_hits",
lambda url, limit: ("ODESZA - Bloom", []))
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
assert r.status_code == 404
```
Note: in `test_non_direct_link_resolve_failure_422`, move the `from server import mf as mf_mod` line above the `def boom` so the closure resolves it. Final form:
```python
def test_non_direct_link_resolve_failure_422(client, auth, monkeypatch):
from server import mf as mf_mod
def boom(url, limit):
raise mf_mod.OdesliError(url)
monkeypatch.setattr("server.app.mf.resolve_link_hits", boom)
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/bad"}, headers=auth)
assert r.status_code == 422
assert "resolve" in r.json()["message"].lower()
```
- [ ] **Step 2: Run to verify they fail**
Run: `pytest tests/test_api_url.py -k non_direct -v`
Expected: FAIL — current code routes every URL through `probe_url`; the spotify URL hits `download_single`, so assertions on `resolve_link_hits`/status differ.
- [ ] **Step 3: Implement**
In `server/app.py`, replace the URL branch:
```python
if mf.is_url(q):
kind, title, hits = mf.probe_url(q)
...
return response
```
with a direct-vs-Odesli split:
```python
if mf.is_url(q):
if mf._is_direct_url(q):
kind, title, hits = mf.probe_url(q)
syn = mf.Hit(source="youtube", kind=kind, title=title, artist="")
job = jobs.create_job(hit=syn, message=actions.url_started_message(kind, title))
response = _job_public(job)
done_msg = actions.playlist_done_message if kind == "playlist" else actions.url_done_message
jobs.run_job(
job.id,
lambda: actions.perform_url_fetch(q, kind, title, hits, quality, ROOT),
done_message=done_msg,
fail_message="Download failed.",
)
return response
# Non-direct link (Spotify/Apple/…): resolve via Odesli, then run the
# normal Lidarr-first pick/dispatch with the exact YouTube track fallback.
try:
_query, hits = mf.resolve_link_hits(q, 10)
except mf.OdesliError:
raise HTTPException(status_code=422,
detail=f"Couldn't resolve {q}. Try the direct YouTube or SoundCloud link.")
if not hits:
raise HTTPException(status_code=404, detail=f"No results found for '{q}'.")
chosen = mf.pick(hits, _query, True, False)
if chosen is None:
raise HTTPException(status_code=404, detail=f"No results found for '{q}'.")
job = jobs.create_job(hit=chosen, message=actions.started_message(chosen))
response = _job_public(job)
jobs.run_job(
job.id,
lambda: actions.perform_fetch(chosen, hits, quality, ROOT),
done_message=actions.done_message(chosen),
fail_message=actions.failed_message(chosen),
)
return response
```
The existing `source`/`build_combined_hits` text-query path below stays unchanged.
- [ ] **Step 4: Run to verify they pass**
Run: `pytest tests/test_api_url.py -v`
Expected: all pass (existing direct-URL tests still green).
- [ ] **Step 5: Commit**
```bash
git add server/app.py tests/test_api_url.py
git commit -m "feat: server /fetch resolves non-direct links via Odesli (Lidarr-first)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Task 8: Full suite + README note
**Files:**
- Modify: `README.md` (input description)
- [ ] **Step 1: Run the whole suite**
Run: `pytest -q`
Expected: all pass.
- [ ] **Step 2: Update README**
In the "It accepts" list near the top, change the URL bullet to mention any-platform links, e.g. add:
```markdown
- **Any streaming link** (Spotify, Apple Music, Tidal, Deezer, …): resolved to
metadata via [Odesli/song.link](https://odesli.co), then searched on Lidarr
first and downloaded from the matching YouTube track if no Lidarr release is
available. YouTube and SoundCloud links download directly.
```
- [ ] **Step 3: Commit**
```bash
git add README.md
git commit -m "docs: document any-link Odesli resolution in README
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>"
```
---
## Self-Review Notes
- **Spec coverage:** `odesli_resolve` (Task 1), `_is_direct_url` (Task 2), `resolve_link_hits`/`handle_link` (Task 4), CLI wiring (Task 5), `server/mf.py` exports (Task 6), server `/fetch` branch (Task 7), tests throughout. `_dispatch_chosen` extraction (Task 3) implements the "reuse existing dispatch" requirement.
- **Decision applied:** Lidarr-first default — `_dispatch_chosen` auto-picks/acts Lidarr album, falls to the exact YouTube hit only on no release.
- **Out-of-scope honored:** no album/year enrichment, no non-yt/sc playlist handling, no caching.
- **Type consistency:** `Resolved(title, artist, thumb, youtube_url)`, `resolve_link_hits -> (query, hits)`, `OdesliError`, `_dispatch_chosen(chosen, hits, root, quality, dry_run, lidarr_only, search_all)` used identically across tasks.

View File

@@ -0,0 +1,113 @@
# YouTube Playlists + Lidarr Profile Hardening — Design
**Date:** 2026-06-08
**Status:** Approved
## Context & Goal
Two additions to `musicfetch` (and its REST API):
1. **Lidarr profile hardening.** `add_artist` currently hardcodes
`qualityProfileId=1` and `get_default_metadata_profile_id()` returns
`profiles[0]["id"]` — the *first* profile, arbitrarily. If a user's profile
order differs, every added artist could silently get the wrong metadata
profile (e.g. **OST** or **None**) or quality profile. Select by name with an
env override instead.
2. **YouTube playlist support.** A playlist URL should download **each track as
its own file**, routed into per-track artist folders, via both the CLI and the
REST API (one playlist = one job).
## Decisions (confirmed with user)
- Playlists work in **CLI + REST API**; one playlist = **one job** (option B).
- Tracks land in **per-track artist folders** `<root>/<artist>/youtube/` (option A),
reusing the existing single-track path.
- Partial failures: job is **`done` if ≥1 track succeeded** (message
`"Downloaded 12/14 tracks (2 failed)"`); `failed` only if zero succeed.
- Playlist expansion prefers **ytmusicapi** for YT Music playlists, falls back to
**`yt-dlp --flat-playlist`**; each track then downloads through the existing
`act_youtube` (first-artist folders, tag overrides, `music.youtube` URLs).
- A `watch?v=X&list=Y` URL stays a **single track** (no surprise batch); only a
pure playlist URL (`/playlist?list=` or `list=` without `v=`) triggers a batch.
## Architecture
### Feature 1 — Profile hardening (`musicfetch`)
- `get_default_metadata_profile_id()` → look up `/api/v1/metadataprofile`, pick the
profile whose name matches env `LIDARR_METADATA_PROFILE` (default `"Standard"`,
case-insensitive); fall back to the first profile's id, then `1`.
- New `get_quality_profile_id()``/api/v1/qualityprofile`, match env
`LIDARR_QUALITY_PROFILE` (default `"Any"`); fall back to first id, then `1`.
- `add_artist` uses `get_quality_profile_id()` instead of literal `1`.
### Feature 2 — Playlists
New units in `musicfetch`:
```
is_playlist_url(url) -> bool
expand_playlist(url) -> tuple[str, list[Hit]] # (playlist_title, track Hits)
download_playlist(url, root, quality, dry_run) -> tuple[int, int] # (ok, total)
```
- **`is_playlist_url`**: true when the URL has `list=` AND no `v=` param, or path
contains `/playlist`. (So `watch?v=...&list=...` → False.)
- **`expand_playlist`**: if `ytmusicapi` is available and the URL is a YT Music
playlist, use `YTMusic().get_playlist(playlist_id)` → tracks with
artist/album/year/videoId. Otherwise `yt-dlp --flat-playlist -J <url>` → entries
(title, uploader→artist, id). Map each to `Hit(source="youtube", kind="track",
…, payload={"videoId": id})`. Returns the playlist title + Hits. Empty/none → `("", [])`.
- **`download_playlist`**: for each Hit call `act_youtube(hit, root, quality,
dry_run)`, catching per-track exceptions (count ok/total); returns `(ok, total)`.
- **`handle_url`** (CLI): if `is_playlist_url` → `download_playlist` and print
`"Downloaded N/M tracks"`; else the existing single-URL download.
### Feature 2 — REST API (`server/`)
`POST /fetch` currently treats `q` only as a search term. Add URL routing:
- In `server/app.py` `fetch()`: if `mf.is_url(q)` → create a **download job**
(not a search). The job runs in `server/actions.py`:
- `is_playlist_url(q)` → `download_playlist(q, ROOT, quality, False)` → result
`{"ok": n, "total": m, "path": None, "lidarr_album_id": None}`; message
`"Downloaded {n}/{m} tracks"` (+ ` ({m-n} failed)` when failures).
- else single URL → reuse `handle_url`-equivalent single download; message
`"Downloaded '<title>'"`.
- Response `hit` for a playlist: `{"source":"youtube","kind":"playlist",
"title": <playlist title>, "artist":"", "album":"", "year":""}`.
- Status: `done` if `ok ≥ 1` (or single URL succeeded), else `failed`.
- `actions.perform_fetch` (search path) is unchanged; a new
`actions.perform_url_fetch(q, quality, root) -> dict` handles the URL branch, and
`started_message`/`done_message` get URL/playlist-aware variants.
## Error Handling
- Per-track download failures in `download_playlist` are caught and counted; the
batch continues. A batch with zero successes → job `failed`.
- `expand_playlist` degrades ytmusicapi → yt-dlp → `("", [])`; an empty expansion
yields a `failed` job with message `"No tracks found in playlist."`.
- Profile lookups already degrade to a sane fallback id on any HTTP error.
## Testing
Unit (mock network / `act_youtube`, no real downloads):
- `is_playlist_url`: `/playlist?list=…` → True; `watch?v=X&list=Y` → False;
`watch?v=X` → False; non-URL → False.
- `get_default_metadata_profile_id` / `get_quality_profile_id`: pick by env name;
fall back to first when name absent; fall back to `1` on error.
- `expand_playlist`: maps ytmusicapi playlist JSON → Hits (title/artist/videoId);
yt-dlp fallback path; empty → `("", [])`.
- `download_playlist`: counts ok/total with one track's `act_youtube` raising.
- API: `POST /fetch` with a playlist URL → job, batch message `"Downloaded n/m…"`,
`done` when ok≥1; single URL → single-download job; zero-success → `failed`.
Live check: a small real YT Music playlist (3-5 tracks) → each track lands in
`<root>/<artist>/youtube/` with correct single-artist tags; job message reports
`N/M`.
## Out of Scope (YAGNI)
Per-track job fan-out (option C), resume/skip-already-downloaded, playlist→Lidarr
album matching, dedup across runs, progress streaming during a batch.

View File

@@ -0,0 +1,149 @@
# Odesli Link Resolution Design
**Date:** 2026-06-13
**Status:** Approved, pending implementation
## Goal
Let users hand MusicFetch *any* song link (Spotify, Apple Music, Tidal, Deezer,
etc.) and have it resolve to real metadata, then run the normal **Lidarr-first,
YouTube-fallback** flow — instead of failing because `yt-dlp` can't download
those platforms.
YouTube and SoundCloud links keep their current direct-`yt-dlp` behavior (they
already download with good metadata). Free-text queries are unchanged.
## Scope
| Input | Behavior |
|-------|----------|
| youtube / youtube-music / soundcloud URL (incl. playlists) | Current `handle_url` direct download — **unchanged** |
| any other URL (spotify/apple/tidal/deezer/…) | **New:** Odesli resolve → Lidarr-first → exact-YouTube fallback |
| free-text query | Current combined search — **unchanged** |
## Approach
Approach A — *resolve-then-reuse*. Turn a non-direct link into a canonical
`"Artist - Title"` query and feed the existing, battle-tested combined flow.
The only metadata source added is Odesli; album/year still come from the Lidarr
path (MusicBrainz) and the YouTube exact-track tags from `yt-dlp`/`ytmusic`.
Rejected:
- **B (full enrichment):** Odesli + MusicBrainz to also resolve album/year and
force-tag precisely. Marginal gain over yt-dlp's own tagging; more calls +
complexity. Noted as an easy later add.
- **C (Odesli for every URL):** uniform but pointless for youtube/soundcloud,
which already work. Out of scope by decision.
## Components
### `odesli_resolve(url) -> Optional[Resolved]` (musicfetch core)
- Request: `GET https://api.song.link/v1-alpha.1/links?url=<urlencoded>&userCountry=US`,
~5s timeout.
- Parse: entity = `data["entitiesByUniqueId"][data["entityUniqueId"]]`.
- `title``entity["title"]`
- `artist``entity["artistName"]`
- `thumb``entity.get("thumbnailUrl", "")`
- `youtube_url``data["linksByPlatform"]["youtubeMusic"]["url"]`,
else `["youtube"]["url"]`, else `""`.
- Returns a small `Resolved` struct (or dict) with the above.
- Returns `None` on **any** failure: network error, non-200, JSON error,
missing `entityUniqueId`/entity, or missing both title and artist.
- No API key required (Odesli public tier, ~10 req/min — fine for interactive use).
### `_is_direct_url(url) -> bool` (musicfetch core)
True when the host is YouTube, YouTube Music, or SoundCloud, OR the URL is a
playlist per the existing `_is_youtube_playlist_url`. These route to the
existing `handle_url`. Everything else (that `is_url`) routes to `handle_link`.
### `handle_link(url, root, quality, dry_run, noninteractive, yt_first, limit)` (musicfetch core)
1. `r = odesli_resolve(url)`.
- If `None`: print error
`"Couldn't resolve <url>. Try the direct YouTube/SoundCloud link."`
and stop. No silent yt-dlp fallback — Spotify et al. aren't downloadable
by yt-dlp, so a fallback would just fail confusingly.
2. `query = f"{r.artist} - {r.title}".strip(" -")`.
3. Lidarr hits ← existing Lidarr search for `query` (`lidarr_search`, which
resolves Artist-Track → MusicBrainz album → Lidarr, filling year).
4. YouTube hit ← **exact** track from Odesli, only if `r.youtube_url`:
`Hit(source="youtube", kind="track", title=r.title, artist=r.artist,
payload={"url": r.youtube_url})`. (Not a fuzzy YouTube re-search — it's the
same track the user shared.) `act_youtube``_track_url` already prefers
`payload["url"]`.
5. `hits = lidarr_hits + ([yt_hit] if yt_hit else [])`.
- If empty → error `"No Lidarr or YouTube source found for <query>."`.
6. `chosen = pick(hits, query, noninteractive, yt_first)`; then the **existing**
dispatch (mirrors `main()`): Lidarr album → `act_lidarr_album`; if no indexer
release, fall through to the YouTube hit; Lidarr artist → `act_lidarr_artist`;
YouTube → `act_youtube`.
Default action when a Lidarr album is found: **Lidarr first** (auto-pick in
noninteractive/server, add + indexer search, fall to exact YouTube only on no
release) — consistent with the current text-query/server behavior.
## Wiring
### CLI `main()`
Replace:
```python
if is_url(query):
handle_url(query, args.root, args.quality, args.dry_run)
return
```
with:
```python
if is_url(query):
if _is_direct_url(query):
handle_url(query, args.root, args.quality, args.dry_run)
else:
handle_link(query, args.root, args.quality, args.dry_run,
args.noninteractive, args.ytsearch, args.limit)
return
```
### Server `/fetch`
In the `mf.is_url(q)` branch, split on `mf._is_direct_url(q)`:
- direct → existing `probe_url` + `perform_url_fetch` job (unchanged).
- non-direct → resolve + build hits via a new `mf.resolve_link_hits(q, limit)`
helper returning `(query, hits)`; then reuse the **existing** combined-search
job path: `pick``actions.perform_fetch(chosen, hits, quality, ROOT)` with
the normal started/done/failed messages. If `resolve_link_hits` yields no
hits → 404 `"No results found for '<url>'."`; if Odesli itself fails → 422/404
with the speakable resolve-failure message.
Keep core download/search logic in `musicfetch`; the server stays a thin reuse
layer. `resolve_link_hits` lives in core so both CLI and server share it (the
CLI `handle_link` is `resolve_link_hits` + the existing dispatch).
### `server/mf.py`
Export new symbols: `_is_direct_url`, `odesli_resolve`, `handle_link`,
`resolve_link_hits`.
## Error handling
- Odesli miss/failure → single speakable error, no crash, no silent bad fallback.
- Lidarr unreachable → existing loud-failure logging + YouTube fallback already
applies (the exact YouTube hit is in `hits`).
- Missing youtube link from Odesli but Lidarr has the album → still works
(Lidarr path), just no YouTube fallback for that one.
## Out of scope (YAGNI)
- Album/year enrichment from Odesli (Approach B) — later if tags prove weak.
- Non-yt/sc playlist links — single-track resolution only for now.
- Caching Odesli responses.
## Tests
- `odesli_resolve`: mocked HTTP — full response, missing-youtube-link, missing
entity / malformed JSON (→ `None`), non-200 (→ `None`).
- `_is_direct_url`: youtube / music.youtube / soundcloud / playlist → True;
spotify / apple / tidal → False.
- `handle_link` / `resolve_link_hits`: Odesli miss → error/empty; Odesli hit →
`hits` contains Lidarr album(s) + exact YouTube track; Lidarr-first dispatch.

File diff suppressed because it is too large Load Diff

View File

@@ -1,7 +1,20 @@
FROM python:3.12-slim
# ffmpeg for audio extraction/embedding; deno is the JS runtime yt-dlp needs
# for YouTube (without it: "No supported JavaScript runtime" -> missing formats
# / HTTP 403). yt-dlp auto-detects deno on PATH.
RUN apt-get update \
&& apt-get install -y --no-install-recommends ffmpeg \
&& apt-get install -y --no-install-recommends ffmpeg ca-certificates curl unzip \
&& arch="$(uname -m)" \
&& case "$arch" in \
x86_64) deno_arch=x86_64-unknown-linux-gnu ;; \
aarch64) deno_arch=aarch64-unknown-linux-gnu ;; \
*) echo "unsupported arch: $arch" >&2; exit 1 ;; \
esac \
&& curl -fsSL "https://github.com/denoland/deno/releases/latest/download/deno-${deno_arch}.zip" -o /tmp/deno.zip \
&& unzip /tmp/deno.zip -d /usr/local/bin \
&& rm /tmp/deno.zip \
&& apt-get purge -y --auto-remove curl unzip \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app

View File

@@ -61,3 +61,38 @@ def perform_fetch(chosen, hits: list, quality: str, root: str) -> dict:
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}

View File

@@ -51,6 +51,44 @@ def fetch(q: str = Query(..., min_length=1),
source: str = Query("auto")):
if quality not in mf.QUALITY_CHOICES:
raise HTTPException(status_code=422, detail=f"Invalid quality '{quality}'.")
if mf.is_url(q):
if mf._is_direct_url(q):
kind, title, hits = mf.probe_url(q)
syn = mf.Hit(source="youtube", kind=kind, title=title, artist="")
job = jobs.create_job(hit=syn, message=actions.url_started_message(kind, title))
response = _job_public(job)
done_msg = actions.playlist_done_message if kind == "playlist" else actions.url_done_message
jobs.run_job(
job.id,
lambda: actions.perform_url_fetch(q, kind, title, hits, quality, ROOT),
done_message=done_msg,
fail_message="Download failed.",
)
return response
# Non-direct link (Spotify/Apple/...): resolve via Odesli, then run the
# normal Lidarr-first pick/dispatch with the exact YouTube track fallback.
try:
query, hits = mf.resolve_link_hits(q, 10)
except mf.OdesliError:
raise HTTPException(status_code=422,
detail=f"Couldn't resolve {q}. Try the direct YouTube or SoundCloud link.")
if not hits:
raise HTTPException(status_code=404, detail=f"No results found for '{q}'.")
chosen = mf.pick(hits, query, True, False)
if chosen is None:
raise HTTPException(status_code=404, detail=f"No results found for '{q}'.")
job = jobs.create_job(hit=chosen, message=actions.started_message(chosen))
response = _job_public(job)
jobs.run_job(
job.id,
lambda: actions.perform_fetch(chosen, hits, quality, ROOT),
done_message=actions.done_message(chosen),
fail_message=actions.failed_message(chosen),
)
return response
if source not in ("auto", "lidarr", "youtube"):
raise HTTPException(status_code=422, detail=f"Invalid source '{source}'.")

View File

@@ -13,13 +13,10 @@ services:
MUSICFETCH_API_KEY: "${MUSICFETCH_API_KEY}"
MUSICFETCH_ROOT: "/media/music"
MUSICFETCH_PORT: "6769"
# Optional: authenticated YouTube cookies to avoid bot-check / rate limits.
# Mount a cookies.txt below and point this at it (in-container path).
YTDLP_COOKIES: "${YTDLP_COOKIES:-}"
volumes:
- /media/music:/media/music
networks:
- lidarr_net
networks:
lidarr_net:
external: true
# Set to the actual network name of your existing Lidarr stack, e.g.:
# name: media_default
# Uncomment and set host path to supply cookies (see YTDLP_COOKIES above):
# - /path/to/cookies.txt:/cookies.txt:ro

View File

@@ -48,7 +48,8 @@ def get_job(job_id: str) -> Optional["Job"]:
return JOBS.get(job_id)
def run_job(job_id: str, fn: Callable[[], dict], done_message: str,
def run_job(job_id: str, fn: Callable[[], dict],
done_message: "str | Callable[[dict], str]",
fail_message: str = "Something went wrong while fetching.") -> None:
def _task():
job = JOBS.get(job_id)
@@ -57,7 +58,8 @@ def run_job(job_id: str, fn: Callable[[], dict], done_message: str,
_touch(job, status="running")
try:
result = fn()
_touch(job, status="done", result=result, message=done_message)
msg = done_message(result) if callable(done_message) else done_message
_touch(job, status="done", result=result, message=msg)
except Exception as e: # noqa: BLE001 — record any failure on the job
_touch(job, status="failed", error=f"{type(e).__name__}: {e}",
message=fail_message)

View File

@@ -24,6 +24,18 @@ act_youtube = _mod.act_youtube
act_lidarr_album = _mod.act_lidarr_album
act_lidarr_artist = _mod.act_lidarr_artist
QUALITY_CHOICES = _mod.QUALITY_CHOICES
is_url = _mod.is_url
probe_url = _mod.probe_url
download_hits = _mod.download_hits
download_single = _mod.download_single
_is_direct_url = _mod._is_direct_url
odesli_resolve = _mod.odesli_resolve
resolve_link_hits = _mod.resolve_link_hits
handle_link = _mod.handle_link
OdesliError = _mod.OdesliError
__all__ = ["Hit", "build_combined_hits", "pick", "act_youtube",
"act_lidarr_album", "act_lidarr_artist", "QUALITY_CHOICES"]
"act_lidarr_album", "act_lidarr_artist", "QUALITY_CHOICES",
"is_url", "probe_url", "download_hits", "download_single",
"_is_direct_url", "odesli_resolve", "resolve_link_hits",
"handle_link", "OdesliError"]

View File

@@ -4,3 +4,4 @@ requests
ytmusicapi
rich
yt-dlp
mutagen

115
tests/test_api_url.py Normal file
View File

@@ -0,0 +1,115 @@
import time
import pytest
from server import jobs as jobs_mod
@pytest.fixture(autouse=True)
def _clear_jobs():
jobs_mod.JOBS.clear()
yield
jobs_mod.JOBS.clear()
def _wait_done(client, auth, job_id, timeout=2.0):
end = time.time() + timeout
while time.time() < end:
b = client.get(f"/jobs/{job_id}", headers=auth).json()
if b["status"] in ("done", "failed"):
return b
time.sleep(0.01)
raise AssertionError("job never finished")
def _mk_hit():
from server import mf
return mf.Hit(source="youtube", kind="track", title="t", artist="a", payload={"videoId": "1"})
def test_playlist_url_batch_job(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.probe_url",
lambda url: ("playlist", "My Mix", [_mk_hit(), _mk_hit(), _mk_hit()]))
monkeypatch.setattr("server.app.mf.download_hits",
lambda hits, root, quality, dry_run: (2, 3))
r = client.post("/fetch", params={"q": "https://soundcloud.com/dj/sets/mix"}, headers=auth)
assert r.status_code == 200
body = r.json()
assert body["status"] == "queued"
assert body["hit"]["kind"] == "playlist"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "2/3" in done["message"]
assert done["result"]["ok"] == 2
def test_playlist_zero_success_fails(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.probe_url",
lambda url: ("playlist", "Dead Mix", [_mk_hit()]))
monkeypatch.setattr("server.app.mf.download_hits",
lambda hits, root, quality, dry_run: (0, 3))
body = client.post("/fetch", params={"q": "https://www.youtube.com/playlist?list=PLy"}, headers=auth).json()
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "failed"
def test_single_video_url_download(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.probe_url", lambda url: ("track", "Song", []))
monkeypatch.setattr("server.app.mf.download_single",
lambda url, root, quality, dry_run: {"title": "Song", "artist": "A", "ok": True})
body = client.post("/fetch", params={"q": "https://soundcloud.com/a/song"}, headers=auth).json()
assert body["hit"]["kind"] == "track"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
assert "Song" in done["message"]
def test_search_query_still_works(client, auth, monkeypatch):
from server import mf
hit = mf.Hit(source="youtube", kind="track", title="T", artist="A", payload={"videoId": "x"})
monkeypatch.setattr("server.app.mf.build_combined_hits",
lambda q, limit, yt_first, lidarr_only, yt_only: [hit])
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr("server.app.actions.perform_fetch",
lambda chosen, hits, quality, root: {"path": "/x", "lidarr_album_id": None})
r = client.post("/fetch", params={"q": "Daft Punk - Discovery"}, headers=auth)
assert r.status_code == 200
assert r.json()["status"] == "queued"
def test_non_direct_link_resolves_and_fetches(client, auth, monkeypatch):
from server import mf
lid = mf.Hit(source="lidarr", kind="album", title="A Moment Apart",
artist="ODESZA", album="A Moment Apart", year="2017",
payload={"album": {"id": 9}})
yt = mf.Hit(source="youtube", kind="track", title="Bloom", artist="ODESZA",
payload={"url": "https://music.youtube.com/watch?v=YYY"})
monkeypatch.setattr("server.app.mf.resolve_link_hits",
lambda url, limit: ("ODESZA - Bloom", [lid, yt]))
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr("server.app.actions.perform_fetch",
lambda chosen, hits, quality, root: {"path": None, "lidarr_album_id": 9})
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
assert r.status_code == 200
body = r.json()
assert body["status"] == "queued"
assert body["hit"]["album"] == "A Moment Apart"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
def test_non_direct_link_resolve_failure_422(client, auth, monkeypatch):
from server import mf as mf_mod
def boom(url, limit):
raise mf_mod.OdesliError(url)
monkeypatch.setattr("server.app.mf.resolve_link_hits", boom)
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/bad"}, headers=auth)
assert r.status_code == 422
assert "resolve" in r.json()["message"].lower()
def test_non_direct_link_no_hits_404(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.resolve_link_hits",
lambda url, limit: ("ODESZA - Bloom", []))
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
assert r.status_code == 404

View File

@@ -51,5 +51,13 @@ def test_eviction_keeps_within_cap():
jobs.JOBS.clear()
def test_run_job_callable_done_message():
job = jobs.create_job(hit={}, message="m")
jobs.run_job(job.id, lambda: {"ok": 2, "total": 3},
done_message=lambda res: f"{res['ok']}/{res['total']} done")
j = _wait(job.id, "done")
assert j.message == "2/3 done"
def teardown_module():
jobs.JOBS.clear()

110
tests/test_lidarr_search.py Normal file
View File

@@ -0,0 +1,110 @@
import server.mf # noqa: F401
import musicfetch_core as mf
DISCOVERY_MBID = "48117b90-a16e-34ca-a514-19c702df1158"
DISCOVERY_ALBUM = {"title": "Discovery", "artist": {"artistName": "Daft Punk"},
"releaseDate": "2001-01-01", "foreignAlbumId": DISCOVERY_MBID}
def test_artist_track_uses_mbid_exact_lookup(monkeypatch):
monkeypatch.setattr(mf, "API_KEY", "testkey")
monkeypatch.setattr(mf, "musicbrainz_best_album",
lambda artist, track: {"album_title": "Discovery", "artist": "Daft Punk",
"year": "2001", "rg_mbid": DISCOVERY_MBID})
seen = {}
def fake_get(path, params=None, timeout=15):
seen["term"] = (params or {}).get("term")
if path == "/api/v1/album/lookup" and seen["term"] == f"mbid:{DISCOVERY_MBID}":
return [DISCOVERY_ALBUM]
return []
monkeypatch.setattr(mf, "lidarr_get", fake_get)
hits = mf.lidarr_search("Daft Punk - Harder Better Faster Stronger", 10)
assert seen["term"] == f"mbid:{DISCOVERY_MBID}"
assert hits[0].album == "Discovery"
assert hits[0].artist == "Daft Punk"
assert hits[0].payload["album"]["foreignAlbumId"] == DISCOVERY_MBID
def test_year_enriched_from_musicbrainz(monkeypatch):
monkeypatch.setattr(mf, "API_KEY", "testkey")
monkeypatch.setattr(mf, "musicbrainz_best_album",
lambda artist, track: {"album_title": "Discovery", "artist": "Daft Punk",
"year": "2001", "rg_mbid": DISCOVERY_MBID})
no_year = [{"title": "Discovery", "artist": {"artistName": "Daft Punk"},
"releaseDate": "", "foreignAlbumId": DISCOVERY_MBID}]
monkeypatch.setattr(mf, "lidarr_get",
lambda path, params=None, timeout=15: no_year if path == "/api/v1/album/lookup" else [])
hits = mf.lidarr_search("Daft Punk - Discovery", 10)
assert hits[0].year == "2001"
def test_no_api_key_returns_empty(monkeypatch):
monkeypatch.setattr(mf, "API_KEY", "")
assert mf.lidarr_search("Daft Punk - Discovery", 10) == []
def test_mb_miss_falls_back_to_lookup(monkeypatch):
monkeypatch.setattr(mf, "API_KEY", "testkey")
monkeypatch.setattr(mf, "musicbrainz_best_album", lambda artist, track: None)
monkeypatch.setattr(mf, "lidarr_get",
lambda path, params=None, timeout=15: [DISCOVERY_ALBUM] if path == "/api/v1/album/lookup" else [])
hits = mf.lidarr_search("Daft Punk - Discovery", 10)
assert hits[0].album == "Discovery"
def test_single_term_is_artist_first(monkeypatch):
monkeypatch.setattr(mf, "API_KEY", "testkey")
def fake_get(path, params=None, timeout=15):
if path == "/api/v1/artist/lookup":
return [{"artistName": "Daft Punk"}]
if path == "/api/v1/album/lookup":
return [DISCOVERY_ALBUM]
return []
monkeypatch.setattr(mf, "lidarr_get", fake_get)
hits = mf.lidarr_search("Daft Punk", 10)
assert hits[0].kind == "artist"
assert hits[0].artist == "Daft Punk"
def test_last_resort_universal_search(monkeypatch):
monkeypatch.setattr(mf, "API_KEY", "testkey")
monkeypatch.setattr(mf, "musicbrainz_best_album", lambda artist, track: None)
def fake_get(path, params=None, timeout=15):
if path == "/api/v1/search":
return [{"album": DISCOVERY_ALBUM}]
return []
monkeypatch.setattr(mf, "lidarr_get", fake_get)
hits = mf.lidarr_search("Daft Punk - Discovery", 10)
assert hits and hits[0].album == "Discovery"
def test_unreachable_lidarr_warns_loudly(monkeypatch, capsys):
# A connection error must surface on stderr (not silent dbg) so the
# YouTube fallback isn't mistaken for "Lidarr had no match".
monkeypatch.setattr(mf, "API_KEY", "testkey")
monkeypatch.setattr(mf, "DEBUG", False)
def boom(path, params=None, timeout=15):
raise mf.ReqConnectionError("Name or service not known")
monkeypatch.setattr(mf, "lidarr_get", boom)
hits = mf._lidarr_album_candidates("anything")
assert hits == []
assert "Lidarr unreachable" in capsys.readouterr().err
def test_http_error_stays_quiet(monkeypatch, capsys):
monkeypatch.setattr(mf, "API_KEY", "testkey")
monkeypatch.setattr(mf, "DEBUG", False)
def boom(path, params=None, timeout=15):
raise mf.RequestException("500 Server Error")
monkeypatch.setattr(mf, "lidarr_get", boom)
assert mf._lidarr_album_candidates("anything") == []
assert "Lidarr unreachable" not in capsys.readouterr().err

View File

@@ -0,0 +1,16 @@
import server.mf as smf
def test_url_helpers_reexported():
assert callable(smf.is_url)
assert callable(smf.probe_url)
assert callable(smf.download_hits)
assert callable(smf.download_single)
def test_odesli_symbols_reexported():
import server.mf as smf
assert callable(smf._is_direct_url)
assert callable(smf.odesli_resolve)
assert callable(smf.resolve_link_hits)
assert callable(smf.handle_link)

View File

@@ -0,0 +1,93 @@
import server.mf # noqa: F401 — loads musicfetch, registers musicfetch_core
import musicfetch_core as mf
# ---- _sanitize_source ----
def test_sanitize_source():
assert mf._sanitize_source("Youtube") == "youtube"
assert mf._sanitize_source("Soundcloud") == "soundcloud"
assert mf._sanitize_source("") == "downloads"
# ---- _entry_to_hit ----
def test_entry_to_hit_soundcloud_keeps_url_no_videoid():
h = mf._entry_to_hit({"id": "t1", "title": "Track", "uploader": "DJ",
"ie_key": "Soundcloud", "url": "https://soundcloud.com/dj/track"})
assert h.payload["extractor"] == "soundcloud"
assert h.payload["url"] == "https://soundcloud.com/dj/track"
assert h.payload["videoId"] is None
assert h.artist == "DJ"
def test_entry_to_hit_youtube_keeps_videoid():
h = mf._entry_to_hit({"id": "vid123", "title": "Song", "channel": "Chan",
"ie_key": "Youtube", "url": "https://youtube.com/watch?v=vid123"})
assert h.payload["extractor"] == "youtube"
assert h.payload["videoId"] == "vid123"
# ---- _track_url ----
def test_track_url_youtube_prefers_music_youtube():
h = mf.Hit(source="youtube", kind="track", title="T", artist="A",
payload={"videoId": "vid", "extractor": "youtube", "url": "https://youtube.com/watch?v=vid"})
assert mf._track_url(h) == "https://music.youtube.com/watch?v=vid"
def test_track_url_soundcloud_uses_native_url():
h = mf.Hit(source="youtube", kind="track", title="T", artist="A",
payload={"videoId": None, "extractor": "soundcloud", "url": "https://soundcloud.com/a/t"})
assert mf._track_url(h) == "https://soundcloud.com/a/t"
def test_track_url_ytmusic_search_hit_default_youtube():
# ytmusicapi search hits carry only videoId (no extractor) -> music.youtube.
h = mf.Hit(source="youtube", kind="track", title="T", artist="A", payload={"videoId": "vid"})
assert mf._track_url(h) == "https://music.youtube.com/watch?v=vid"
# ---- act_youtube routes to per-source folder ----
def test_act_youtube_soundcloud_folder(monkeypatch):
captured = {}
monkeypatch.setattr(mf, "yt_download",
lambda url, target, quality, dry_run, hit=None: captured.update(url=url, target=target) or True)
h = mf.Hit(source="youtube", kind="track", title="T", artist="DJ, Other",
payload={"videoId": None, "extractor": "soundcloud", "url": "https://soundcloud.com/dj/t"})
mf.act_youtube(h, "/media/music", "best", False)
assert captured["target"] == "/media/music/DJ/soundcloud" # first artist only
assert captured["url"] == "https://soundcloud.com/dj/t"
def test_act_youtube_youtube_folder(monkeypatch):
captured = {}
monkeypatch.setattr(mf, "yt_download",
lambda url, target, quality, dry_run, hit=None, outtmpl=None:
captured.update(target=target) or True)
h = mf.Hit(source="youtube", kind="track", title="T", artist="A",
payload={"videoId": "vid", "extractor": "youtube"})
mf.act_youtube(h, "/media/music", "best", False)
assert captured["target"] == "/media/music/A/youtube"
def test_act_youtube_unknown_artist_uses_metadata_template(monkeypatch):
captured = {}
monkeypatch.setattr(mf, "yt_download",
lambda url, target, quality, dry_run, hit=None, outtmpl=None:
captured.update(target=target, outtmpl=outtmpl) or True)
h = mf.Hit(source="youtube", kind="track", title="", artist="",
payload={"videoId": None, "extractor": "soundcloud", "url": "https://soundcloud.com/a/t"})
mf.act_youtube(h, "/media/music", "best", False)
assert captured["target"] is None
assert "%(artist,uploader,channel)s" in captured["outtmpl"]
assert captured["outtmpl"].endswith("/soundcloud/%(title)s [%(id)s].%(ext)s")
# ---- download_single per-source folder ----
def test_download_single_bandcamp_folder(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url: {"title": "Song", "artist": "Band", "extractor": "Bandcamp"})
captured = {}
monkeypatch.setattr(mf, "yt_download",
lambda url, target, quality, dry_run, hit=None: captured.update(target=target) or True)
info = mf.download_single("https://band.bandcamp.com/track/song", "/media/music", "best", False)
assert captured["target"] == "/media/music/Band/bandcamp"
assert info == {"title": "Song", "artist": "Band", "ok": True}

96
tests/test_musicbrainz.py Normal file
View File

@@ -0,0 +1,96 @@
import server.mf # noqa: F401
import musicfetch_core as mf
class _FakeResp:
def __init__(self, payload):
self._payload = payload
def raise_for_status(self):
pass
def json(self):
return self._payload
MB_PAYLOAD = {
"recordings": [
{
"artist-credit": [{"name": "Daft Punk"}],
"releases": [
{"date": "2001",
"release-group": {"id": "single-mbid", "title": "Harder, Better, Faster, Stronger",
"primary-type": "Single", "secondary-types": []}},
{"date": "2002",
"release-group": {"id": "comp-mbid", "title": "Musique, Vol. 1",
"primary-type": "Album", "secondary-types": ["Compilation"]}},
{"date": "2001",
"release-group": {"id": "48117b90-a16e-34ca-a514-19c702df1158",
"title": "Discovery", "primary-type": "Album",
"secondary-types": []}},
],
}
]
}
def test_picks_studio_album_over_single_and_comp(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(MB_PAYLOAD))
monkeypatch.setattr(mf.time, "sleep", lambda *_: None)
out = mf.musicbrainz_best_album("Daft Punk", "Harder Better Faster Stronger")
assert out["album_title"] == "Discovery"
assert out["artist"] == "Daft Punk"
assert out["year"] == "2001"
assert out["rg_mbid"] == "48117b90-a16e-34ca-a514-19c702df1158"
def test_returns_none_on_empty(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp({"recordings": []}))
monkeypatch.setattr(mf.time, "sleep", lambda *_: None)
assert mf.musicbrainz_best_album("Nobody", "Nothing") is None
def test_returns_none_on_exception(monkeypatch):
def boom(*a, **k):
raise mf.requests.exceptions.RequestException("network down")
monkeypatch.setattr(mf.requests, "get", boom)
monkeypatch.setattr(mf.time, "sleep", lambda *_: None)
assert mf.musicbrainz_best_album("Daft Punk", "Discovery") is None
def test_falls_back_to_any_releasegroup_when_no_studio(monkeypatch):
payload = {"recordings": [{"artist-credit": [{"name": "X"}], "releases": [
{"date": "2010", "release-group": {"id": "live1", "title": "Live Thing",
"primary-type": "Album", "secondary-types": ["Live"]}},
]}]}
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(payload))
monkeypatch.setattr(mf.time, "sleep", lambda *_: None)
out = mf.musicbrainz_best_album("X", "Y")
assert out["album_title"] == "Live Thing"
def test_first_artist_credit_only(monkeypatch):
payload = {"recordings": [{"artist-credit": [{"name": "SLVMLORD"}, {"name": "Travis Bradley"}],
"releases": [{"date": "2025",
"release-group": {"id": "x", "title": "Album X",
"primary-type": "Album",
"secondary-types": []}}]}]}
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(payload))
monkeypatch.setattr(mf.time, "sleep", lambda *_: None)
out = mf.musicbrainz_best_album("SLVMLORD", "Under My Skin")
assert out["artist"] == "SLVMLORD"
def test_prefers_own_artist_studio_over_various_artists(monkeypatch):
# A studio-looking VA compilation dated earlier must NOT beat the artist's own album.
payload = {"recordings": [{"artist-credit": [{"name": "Daft Punk"}], "releases": [
{"date": "2001-10-26", "artist-credit": [{"name": "Various Artists"}],
"release-group": {"id": "va-mbid", "title": "All The Hits Now",
"primary-type": "Album", "secondary-types": []}},
{"date": "2002", "artist-credit": [{"name": "Daft Punk"}],
"release-group": {"id": "48117b90-a16e-34ca-a514-19c702df1158", "title": "Discovery",
"primary-type": "Album", "secondary-types": []}},
]}]}
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(payload))
monkeypatch.setattr(mf.time, "sleep", lambda *_: None)
out = mf.musicbrainz_best_album("Daft Punk", "Harder Better Faster Stronger")
assert out["album_title"] == "Discovery"
assert out["rg_mbid"] == "48117b90-a16e-34ca-a514-19c702df1158"

169
tests/test_odesli.py Normal file
View File

@@ -0,0 +1,169 @@
import server.mf # noqa: F401 — loads musicfetch, registers musicfetch_core
import musicfetch_core as mf
class _FakeResp:
def __init__(self, status, payload):
self.status_code = status
self._payload = payload
def json(self):
return self._payload
FULL = {
"entityUniqueId": "SPOTIFY_SONG::abc",
"entitiesByUniqueId": {
"SPOTIFY_SONG::abc": {
"title": "Bloom",
"artistName": "ODESZA",
"thumbnailUrl": "https://img/cover.jpg",
}
},
"linksByPlatform": {
"youtubeMusic": {"url": "https://music.youtube.com/watch?v=YYY"},
"youtube": {"url": "https://youtube.com/watch?v=YYY"},
},
}
def test_odesli_resolve_full(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, FULL))
r = mf.odesli_resolve("https://open.spotify.com/track/abc")
assert r.title == "Bloom"
assert r.artist == "ODESZA"
assert r.thumb == "https://img/cover.jpg"
assert r.youtube_url == "https://music.youtube.com/watch?v=YYY"
def test_odesli_resolve_prefers_ytmusic_then_youtube(monkeypatch):
payload = {**FULL, "linksByPlatform": {"youtube": {"url": "https://youtube.com/watch?v=ZZZ"}}}
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, payload))
r = mf.odesli_resolve("https://open.spotify.com/track/abc")
assert r.youtube_url == "https://youtube.com/watch?v=ZZZ"
def test_odesli_resolve_no_youtube_link(monkeypatch):
payload = {**FULL, "linksByPlatform": {}}
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, payload))
r = mf.odesli_resolve("https://open.spotify.com/track/abc")
assert r.youtube_url == ""
assert r.title == "Bloom"
def test_odesli_resolve_non_200_returns_none(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(429, {}))
assert mf.odesli_resolve("https://open.spotify.com/track/abc") is None
def test_odesli_resolve_malformed_returns_none(monkeypatch):
monkeypatch.setattr(mf.requests, "get", lambda *a, **k: _FakeResp(200, {"nope": 1}))
assert mf.odesli_resolve("https://open.spotify.com/track/abc") is None
def test_odesli_resolve_network_error_returns_none(monkeypatch):
def boom(*a, **k):
raise mf.RequestException("down")
monkeypatch.setattr(mf.requests, "get", boom)
assert mf.odesli_resolve("https://open.spotify.com/track/abc") is None
def test_is_direct_url_youtube():
assert mf._is_direct_url("https://music.youtube.com/watch?v=abc")
assert mf._is_direct_url("https://www.youtube.com/watch?v=abc")
assert mf._is_direct_url("https://youtu.be/abc")
def test_is_direct_url_soundcloud():
assert mf._is_direct_url("https://soundcloud.com/dj/track")
def test_is_direct_url_other_platforms_false():
assert not mf._is_direct_url("https://open.spotify.com/track/abc")
assert not mf._is_direct_url("https://music.apple.com/us/album/x/1?i=2")
assert not mf._is_direct_url("https://tidal.com/browse/track/123")
def test_is_direct_url_youtube_playlist_true():
assert mf._is_direct_url("https://www.youtube.com/playlist?list=PLabc")
def test_is_direct_url_lookalike_hosts_false():
# Trailing-substring look-alikes must NOT be treated as direct (label boundary).
assert not mf._is_direct_url("https://notyoutube.com/watch?v=abc")
assert not mf._is_direct_url("https://myyoutube.com/x")
assert not mf._is_direct_url("https://evilyoutu.be/x")
assert not mf._is_direct_url("https://youtube.com.evil.com/x")
def test_is_direct_url_subdomains_true():
assert mf._is_direct_url("https://m.youtube.com/watch?v=abc")
assert mf._is_direct_url("https://on.soundcloud.com/x")
assert mf._is_direct_url("https://api.soundcloud.com/tracks/1")
def _resolved(yt="https://music.youtube.com/watch?v=YYY"):
return mf.Resolved(title="Bloom", artist="ODESZA",
thumb="https://img/cover.jpg", youtube_url=yt)
def _ytsearch_hit():
return mf.Hit(source="youtube", kind="track", title="Bloom (search)",
artist="ODESZA", payload={"videoId": "srch"})
def test_resolve_link_hits_lidarr_first_then_exact_then_search(monkeypatch):
# Odesli supplies a YouTube link -> exact track is the FIRST youtube hit,
# ahead of the fuzzy youtube_search results, and after the Lidarr hits.
monkeypatch.setattr(mf, "odesli_resolve", lambda url: _resolved())
lid = mf.Hit(source="lidarr", kind="album", title="A Moment Apart",
artist="ODESZA", album="A Moment Apart", year="2017",
payload={"album": {"id": 9}})
monkeypatch.setattr(mf, "lidarr_search", lambda q, limit: [lid])
monkeypatch.setattr(mf, "youtube_search", lambda q, limit: [_ytsearch_hit()])
query, hits = mf.resolve_link_hits("https://open.spotify.com/track/abc", 10)
assert query == "ODESZA - Bloom"
assert hits[0].source == "lidarr"
yt = [h for h in hits if h.source == "youtube"]
assert yt[0].payload.get("url") == "https://music.youtube.com/watch?v=YYY" # exact first
assert any(h.payload.get("videoId") == "srch" for h in yt) # search fallback present
def test_resolve_link_hits_no_odesli_yt_uses_search_fallback(monkeypatch):
# Regression: Odesli often omits YouTube links. With no Lidarr match and no
# Odesli YouTube link, a normal youtube_search must still yield hits (not empty).
monkeypatch.setattr(mf, "odesli_resolve", lambda url: _resolved(yt=""))
monkeypatch.setattr(mf, "lidarr_search", lambda q, limit: [])
monkeypatch.setattr(mf, "youtube_search", lambda q, limit: [_ytsearch_hit()])
query, hits = mf.resolve_link_hits("https://open.spotify.com/track/abc", 10)
assert hits, "must not be empty when YouTube search finds the track"
assert all(h.source == "youtube" for h in hits)
assert not any(h.payload.get("url") for h in hits) # no exact Odesli hit injected
def test_resolve_link_hits_odesli_miss_raises(monkeypatch):
import pytest
monkeypatch.setattr(mf, "odesli_resolve", lambda url: None)
with pytest.raises(mf.OdesliError):
mf.resolve_link_hits("https://open.spotify.com/track/abc", 10)
def test_handle_link_miss_prints_and_returns(monkeypatch, capsys):
monkeypatch.setattr(mf, "odesli_resolve", lambda url: None)
mf.handle_link("https://open.spotify.com/track/abc", "/m", "best",
False, True, False, 10)
out = capsys.readouterr()
assert "Couldn't resolve" in (out.err + out.out)
def test_handle_link_dispatches_chosen(monkeypatch):
monkeypatch.setattr(mf, "odesli_resolve", lambda url: _resolved())
monkeypatch.setattr(mf, "lidarr_search", lambda q, limit: [])
chosen = {}
monkeypatch.setattr(mf, "pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr(mf, "_dispatch_chosen",
lambda c, hits, root, quality, dry, lo, sa: chosen.update(c=c, root=root))
mf.handle_link("https://open.spotify.com/track/abc", "/m", "best",
False, True, False, 10)
assert chosen["c"].source == "youtube"
assert chosen["root"] == "/m"

144
tests/test_playlist.py Normal file
View File

@@ -0,0 +1,144 @@
import json as _json
import server.mf # noqa: F401 — loads musicfetch, registers musicfetch_core
import musicfetch_core as mf
class _CP:
def __init__(self, stdout):
self.stdout = stdout
self.returncode = 0
# ---- _is_youtube_playlist_url ----
def test_youtube_playlist_url_true():
assert mf._is_youtube_playlist_url("https://music.youtube.com/playlist?list=PLabc") is True
assert mf._is_youtube_playlist_url("https://www.youtube.com/playlist?list=PLabc") is True
def test_youtube_watch_with_list_is_not_playlist():
assert mf._is_youtube_playlist_url("https://www.youtube.com/watch?v=abc&list=PLx") is False
def test_non_youtube_url_not_youtube_playlist():
# SoundCloud sets are not matched here — probe_url handles them via yt-dlp.
assert mf._is_youtube_playlist_url("https://soundcloud.com/user/sets/mix") is False
# ---- probe_url ----
def test_probe_url_youtube_playlist_uses_ytmusic(monkeypatch):
h = mf.Hit(source="youtube", kind="track", title="A", artist="X",
payload={"videoId": "1", "extractor": "youtube"})
monkeypatch.setattr(mf, "_ytmusic_playlist", lambda pid: ("My YT Mix", [h]))
monkeypatch.setattr(mf, "YTMusic", object()) # non-None to enter ytmusic branch
kind, title, hits = mf.probe_url("https://music.youtube.com/playlist?list=PLx")
assert kind == "playlist"
assert title == "My YT Mix"
assert hits == [h]
def test_probe_url_generic_playlist_via_ytdlp(monkeypatch):
monkeypatch.setattr(mf, "YTMusic", None)
payload = {"title": "SC Set", "_type": "playlist", "entries": [
{"id": "t1", "title": "One", "uploader": "DJ", "ie_key": "Soundcloud",
"url": "https://soundcloud.com/dj/one"},
{"id": None, "url": None, "title": "skip"},
]}
monkeypatch.setattr(mf.subprocess, "run", lambda *a, **k: _CP(_json.dumps(payload)))
kind, title, hits = mf.probe_url("https://soundcloud.com/dj/sets/sc-set")
assert kind == "playlist"
assert title == "SC Set"
assert len(hits) == 1
assert hits[0].payload["extractor"] == "soundcloud"
assert hits[0].payload["url"] == "https://soundcloud.com/dj/one"
def test_probe_url_single_track(monkeypatch):
monkeypatch.setattr(mf, "YTMusic", None)
payload = {"title": "A Song", "extractor": "soundcloud"} # no entries -> single
monkeypatch.setattr(mf.subprocess, "run", lambda *a, **k: _CP(_json.dumps(payload)))
kind, title, hits = mf.probe_url("https://soundcloud.com/dj/one")
assert kind == "track"
assert title == "A Song"
assert hits == []
def test_probe_url_failure_returns_track(monkeypatch):
monkeypatch.setattr(mf, "YTMusic", None)
def boom(*a, **k):
raise mf.subprocess.CalledProcessError(1, "yt-dlp")
monkeypatch.setattr(mf.subprocess, "run", boom)
assert mf.probe_url("https://example.com/x") == ("track", "", [])
# ---- download_hits ----
def test_download_hits_counts(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
h3 = mf.Hit(source="youtube", kind="track", title="C", artist="Z", payload={"videoId": "3"})
monkeypatch.setattr(mf, "act_youtube", lambda hit, root, quality, dry_run: hit.title != "B")
assert mf.download_hits([h1, h2, h3], "/tmp", "best", False) == (2, 3)
def test_download_hits_track_exception_is_failure(monkeypatch):
h1 = mf.Hit(source="youtube", kind="track", title="A", artist="X", payload={"videoId": "1"})
h2 = mf.Hit(source="youtube", kind="track", title="B", artist="Y", payload={"videoId": "2"})
def fake_act(hit, root, quality, dry_run):
if hit.title == "B":
raise RuntimeError("boom")
return True
monkeypatch.setattr(mf, "act_youtube", fake_act)
assert mf.download_hits([h1, h2], "/tmp", "best", False) == (1, 2)
# ---- yt_download bool ----
def test_yt_download_returns_true_on_zero_exit(monkeypatch):
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
monkeypatch.setattr(mf.subprocess, "run", lambda *a, **k: _CP(""))
assert mf.yt_download("u", "/tmp/x", "best", False) is True
def test_yt_download_dry_run_returns_true():
assert mf.yt_download("u", "/tmp/x", "best", True) is True
def test_yt_download_always_sets_album_default(monkeypatch):
captured = {}
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _CP(""))
mf.yt_download("u", "/tmp/x", "best", False)
assert "%(album|Unknown Album)s:%(meta_album)s" in captured["cmd"]
def test_yt_download_single_word_tags_injected_literally(monkeypatch):
# Regression: `--parse-metadata "Cochise:%(title)s"` makes yt-dlp treat the
# bare word 'Cochise' as a FIELD name (field_to_template's r'[a-zA-Z_]+$'),
# producing 'NA'. Single-word album/title must reach yt-dlp as literals.
captured = {}
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _CP(""))
hit = mf.Hit(source="youtube", kind="track", title="Cochise",
artist="Audioslave", album="Solid", payload={"videoId": "x"})
mf.yt_download("u", "/tmp/x", "best", False, hit=hit)
cmd = captured["cmd"]
joined = " ".join(cmd)
# The buggy bare-word parse-metadata FROM must be gone.
assert "Solid:%(album)s" not in joined
assert "Cochise:%(title)s" not in joined
# Literal values must be passed as literal args (immune to template parsing).
assert "Solid" in cmd
assert "Cochise" in cmd
# A hit album must not be clobbered by the Unknown-Album default.
assert "%(album|Unknown Album)s:%(meta_album)s" not in cmd
def test_yt_download_passes_cookies(monkeypatch):
captured = {}
monkeypatch.setattr(mf, "COOKIES_FILE", "/cookies.txt")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _CP(""))
mf.yt_download("u", "/tmp/x", "best", False)
assert "--cookies" in captured["cmd"] and "/cookies.txt" in captured["cmd"]

43
tests/test_profiles.py Normal file
View File

@@ -0,0 +1,43 @@
import server.mf # noqa: F401
import musicfetch_core as mf
META = [{"id": 1, "name": "Standard"}, {"id": 2, "name": "None"}, {"id": 3, "name": "OST"}]
QUAL = [{"id": 1, "name": "Any"}, {"id": 2, "name": "Lossless"}]
def test_metadata_profile_default_standard_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_METADATA_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1
def test_metadata_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "OST")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 3
def test_metadata_profile_unknown_name_falls_back_to_first(monkeypatch):
monkeypatch.setenv("LIDARR_METADATA_PROFILE", "Nonexistent")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: META)
assert mf.get_default_metadata_profile_id() == 1
def test_quality_profile_default_any_by_name(monkeypatch):
monkeypatch.delenv("LIDARR_QUALITY_PROFILE", raising=False)
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 1
def test_quality_profile_env_override(monkeypatch):
monkeypatch.setenv("LIDARR_QUALITY_PROFILE", "Lossless")
monkeypatch.setattr(mf, "lidarr_get", lambda path, timeout=10: QUAL)
assert mf.get_quality_profile_id() == 2
def test_profile_fetch_error_returns_one(monkeypatch):
def boom(path, timeout=10):
raise mf.RequestException("down")
monkeypatch.setattr(mf, "lidarr_get", boom)
assert mf.get_default_metadata_profile_id() == 1
assert mf.get_quality_profile_id() == 1

411
tests/test_repair.py Normal file
View File

@@ -0,0 +1,411 @@
import server.mf # noqa: F401 — loads musicfetch, registers musicfetch_core
import musicfetch_core as mf
YT_ID = "dQw4w9WgXcQ" # valid 11-char YouTube id
# ---- _is_source_dir ----
def test_is_source_dir():
assert mf._is_source_dir("youtube") is True
assert mf._is_source_dir("soundcloud") is True
assert mf._is_source_dir("downloads") is True
assert mf._is_source_dir("Discovery") is False # Lidarr album folder
assert mf._is_source_dir("Random Access Memories") is False
assert mf._is_source_dir("") is False
# ---- _parse_track_file ----
def test_parse_track_file():
assert mf._parse_track_file("Under My Skin [nGSNF2l44Zc].opus") == ("Under My Skin", "nGSNF2l44Zc")
assert mf._parse_track_file("Ignomon [2202690443].m4a") == ("Ignomon", "2202690443")
# greedy title: real id is the LAST bracket
assert mf._parse_track_file("WHO GON' SLIDE [Official Music Video] [AxjP9s6J3uY].opus") \
== ("WHO GON' SLIDE [Official Music Video]", "AxjP9s6J3uY")
assert mf._parse_track_file("no-id-here.opus") is None
assert mf._parse_track_file("cover.jpg") is None
# ---- _repair_id_ok ----
def test_repair_id_ok():
assert mf._repair_id_ok("youtube", YT_ID) is True
assert mf._repair_id_ok("youtube", "Official Video") is False # space, wrong length
assert mf._repair_id_ok("youtube", "Cover") is False
assert mf._repair_id_ok("soundcloud", "2202690443") is True
assert mf._repair_id_ok("soundcloud", "abc") is False
assert mf._repair_id_ok("bandcamp", "x") is False
# ---- _valid_year ----
def test_valid_year():
assert mf._valid_year({"release_year": 2001}) == "2001"
assert mf._valid_year({"release_date": "1976-09-10"}) == "1976"
assert mf._valid_year({"upload_date": "20110101"}) == "" # upload date ignored
assert mf._valid_year({"release_year": 6577}) == "" # out of range
assert mf._valid_year({}) == ""
# ---- _repair_probe_url ----
def test_repair_probe_url():
assert mf._repair_probe_url("youtube", YT_ID) == f"https://music.youtube.com/watch?v={YT_ID}"
assert mf._repair_probe_url("soundcloud", "123") == "https://api.soundcloud.com/tracks/123"
assert mf._repair_probe_url("bandcamp", "x") is None
# ---- repair_file (fake audio + mocked metadata) ----
class _FakeAudio(dict):
def __init__(self, initial):
super().__init__(initial)
self.saved = False
def save(self):
self.saved = True
def test_repair_file_fixes_album_and_year(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Daft Punk", "title": "Aerodynamic",
"album": "Discovery", "release_year": 2001})
audio = _FakeAudio({"artist": ["Daft Punk"], "title": ["Aerodynamic"]}) # album/date missing
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Aerodynamic [{YT_ID}].opus", "youtube", dry_run=False)
assert set(changed) == {"album=Discovery", "date=2001"}
assert audio["album"] == ["Discovery"]
assert audio["date"] == ["2001"]
assert audio.saved is True
def test_repair_file_dry_run_writes_nothing(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "A", "title": "T", "album": "Alb", "release_year": 2020})
audio = _FakeAudio({})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/T [{YT_ID}].opus", "youtube", dry_run=True)
assert changed
assert audio == {}
assert audio.saved is False
def test_repair_file_skips_music_video(monkeypatch):
# No album AND no valid release year -> treat as a video, leave tags alone.
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"title": "Artist - Song (Official Music Video)",
"uploader": "SomeVEVO", "upload_date": "20110101"})
audio = _FakeAudio({"artist": ["Real Artist"], "title": ["Song"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Song [{YT_ID}].opus", "youtube", dry_run=False)
assert changed == []
assert audio == {"artist": ["Real Artist"], "title": ["Song"]} # untouched
def test_repair_file_fills_missing_but_never_clobbers(monkeypatch):
# Source artist is a channel name; existing artist must be kept.
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "SomeChannelVEVO", "title": "Channel Decorated Title",
"album": "Real Album", "release_year": 2019})
audio = _FakeAudio({"artist": ["Correct Artist"], "title": ["Clean Title"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/x [{YT_ID}].opus", "youtube", dry_run=False)
assert set(changed) == {"album=Real Album", "date=2019"}
assert audio["artist"] == ["Correct Artist"] # NOT overwritten with channel
assert audio["title"] == ["Clean Title"] # NOT overwritten with decorated title
def test_repair_file_fills_missing_artist_when_absent(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Real Artist", "title": "T",
"album": "Alb", "release_year": 2020})
audio = _FakeAudio({}) # nothing present -> fill artist + title too
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/T [{YT_ID}].opus", "youtube", dry_run=False)
assert set(changed) == {"album=Alb", "date=2020", "artist=Real Artist", "title=T"}
def test_repair_file_skips_bad_id(monkeypatch):
called = {"meta": False}
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: called.update(meta=True) or {})
# last bracket is a descriptor, not a real id
assert mf.repair_file("X/youtube/Song [Official Video].opus", "youtube", dry_run=False) == []
assert called["meta"] is False # never hit the network
def test_repair_file_skips_unparseable(monkeypatch):
called = {"meta": False}
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: called.update(meta=True) or {})
assert mf.repair_file("X/youtube/no-id.opus", "youtube", dry_run=False) == []
assert called["meta"] is False
def test_run_yt_dlp_get_metadata_passes_extra_args(monkeypatch):
captured = {}
class _R:
stdout = '{"title": "x"}'
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _R())
mf.run_yt_dlp_get_metadata("http://u", ["--extractor-args", "youtube:player_skip=js"])
assert "youtube:player_skip=js" in captured["cmd"]
def test_repair_uses_player_skip_fast_args(monkeypatch):
captured = {}
def fake_meta(url, extra_args=None):
captured["extra"] = extra_args
return {"album": "A", "release_year": 2020, "artist": "X", "title": "T"}
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", fake_meta)
monkeypatch.setattr(mf, "_open_audio", lambda p: (_FakeAudio({}), None))
mf.repair_file(f"X/youtube/T [{YT_ID}].opus", "youtube", dry_run=True)
assert captured["extra"] == mf._REPAIR_META_ARGS
# ---- repair_library (real temp tree, repair_file mocked) ----
def test_repair_library_scans_only_source_dirs(tmp_path, monkeypatch):
root = tmp_path
(root / "Daft Punk" / "youtube").mkdir(parents=True)
(root / "Daft Punk" / "youtube" / f"Aerodynamic [{YT_ID}].opus").write_text("x")
(root / "Daft Punk" / "Discovery").mkdir(parents=True) # Lidarr album -> skip
(root / "Daft Punk" / "Discovery" / "Aerodynamic.flac").write_text("x")
(root / "Ephixa" / "soundcloud").mkdir(parents=True)
(root / "Ephixa" / "soundcloud" / "Ignomon [123].m4a").write_text("x")
visited = []
monkeypatch.setattr(mf, "repair_file",
lambda path, source, dry_run: visited.append((source, path)) or ["album=X"])
scanned, changed = mf.repair_library(str(root), dry_run=False)
assert scanned == 2 and changed == 2
assert sorted(s for s, _ in visited) == ["soundcloud", "youtube"] # album folder skipped
def test_repair_library_missing_root():
assert mf.repair_library("/no/such/dir", dry_run=False) == (0, 0)
def test_repair_library_exclude_skips_folders(tmp_path, monkeypatch):
root = tmp_path
(root / "Daft Punk" / "youtube").mkdir(parents=True)
(root / "Daft Punk" / "youtube" / f"A [{YT_ID}].opus").write_text("x")
(root / "Unsorted" / "youtube").mkdir(parents=True) # excluded artist folder
(root / "Unsorted" / "youtube" / f"B [{YT_ID}].opus").write_text("x")
(root / "Ephixa" / "playlists").mkdir(parents=True) # excluded source folder
(root / "Ephixa" / "playlists" / f"C [{YT_ID}].opus").write_text("x")
visited = []
monkeypatch.setattr(mf, "repair_file",
lambda path, source, dry_run: visited.append(path) or ["x"])
scanned, _ = mf.repair_library(str(root), dry_run=False, exclude=["unsorted", "playlists"])
assert scanned == 1
assert visited and "Daft Punk" in visited[0]
# ---- offline retag-from-path ----
def test_title_from_filename():
assert mf._title_from_filename(f"Song [{YT_ID}].opus") == "Song"
assert mf._title_from_filename("STARDUST (Official Music Video) [3nsYNXtALhA].opus") \
== "STARDUST (Official Music Video)"
assert mf._title_from_filename("no brackets.mp3") == "no brackets"
def test_strip_decorations():
assert mf._strip_decorations("STARDUST (Official Music Video)") == "STARDUST"
assert mf._strip_decorations("Away From You (Lyrics)") == "Away From You"
assert mf._strip_decorations("More Than a Feeling (Official HD Video)") == "More Than a Feeling"
# real info like a feature credit is kept
assert mf._strip_decorations("WHO GON' SLIDE (Feat. Shakewell) [Official Music Video]") \
== "WHO GON' SLIDE (Feat. Shakewell)"
def test_derive_from_filename():
# plain title -> folder is the artist
assert mf._derive_from_filename(f"Aerodynamic [{YT_ID}].opus", "Daft Punk") == ("Daft Punk", "Aerodynamic")
# decorated music video filed under the artist
assert mf._derive_from_filename("STARDUST (Official Music Video) [3nsYNXtALhA].opus", "1nonly") \
== ("1nonly", "STARDUST")
# 'Artist - Title' name wins over a channel folder
assert mf._derive_from_filename("BLCKLGHT - Away From You (Lyrics) [QapF4b1jYw8].opus", "7clouds Techno") \
== ("BLCKLGHT", "Away From You")
def test_retag_file_from_path_fixes_clobbered_tags(monkeypatch):
audio = _FakeAudio({"artist": ["7clouds Techno"], "title": ["BLCKLGHT - Away From You (Lyrics)"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.retag_file_from_path(
"X/7clouds Techno/youtube/BLCKLGHT - Away From You (Lyrics) [QapF4b1jYw8].opus",
"7clouds Techno", dry_run=False)
assert set(changed) == {"artist=BLCKLGHT", "title=Away From You"}
assert audio["artist"] == ["BLCKLGHT"]
assert audio["title"] == ["Away From You"]
assert audio.saved is True
def test_retag_file_from_path_dry_run(monkeypatch):
audio = _FakeAudio({"artist": ["wrong"], "title": ["wrong"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.retag_file_from_path(f"X/Daft Punk/youtube/Aerodynamic [{YT_ID}].opus",
"Daft Punk", dry_run=True)
assert changed
assert audio == {"artist": ["wrong"], "title": ["wrong"]}
assert audio.saved is False
def test_retag_library_walks_source_files(tmp_path, monkeypatch):
root = tmp_path
(root / "Daft Punk" / "youtube").mkdir(parents=True)
(root / "Daft Punk" / "youtube" / f"Aerodynamic [{YT_ID}].opus").write_text("x")
(root / "Daft Punk" / "Discovery").mkdir(parents=True) # album folder -> skip
(root / "Daft Punk" / "Discovery" / "x.flac").write_text("x")
visited = []
monkeypatch.setattr(mf, "retag_file_from_path",
lambda path, artist, dry_run: visited.append(artist) or ["artist=x"])
scanned, changed = mf.retag_library_from_path(str(root), dry_run=False)
assert (scanned, changed) == (1, 1)
assert visited == ["Daft Punk"]
# ---- bogus-tag recovery (old-code NA / Unknown breakage) ----
def test_is_bogus():
for v in ("", "NA", "na", "N/A", "Unknown", "Unknown Album", "unknown artist", " NA "):
assert mf._is_bogus(v) is True, v
for v in ("Cochise", "Solid", "Brother Stoon", "Discovery"):
assert mf._is_bogus(v) is False, v
def test_repair_file_overwrites_bogus_title(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
"album": "Audioslave", "release_year": 2002})
audio = _FakeAudio({"artist": ["Audioslave"], "title": ["NA"]}) # bogus title
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Brother Stoon [{YT_ID}].opus", "youtube", dry_run=False)
assert "title=Cochise" in changed
assert audio["title"] == ["Cochise"]
def test_repair_file_overwrites_bogus_artist(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Real Artist", "title": "Real Title",
"album": "Alb", "release_year": 2020})
audio = _FakeAudio({"artist": ["NA"], "title": ["Good Title"]}) # bogus artist, good title
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Good Title [{YT_ID}].opus", "youtube", dry_run=False)
assert "artist=Real Artist" in changed
assert audio["artist"] == ["Real Artist"]
assert audio["title"] == ["Good Title"] # good title untouched
def test_repair_file_normalizes_na_album_when_source_has_none(monkeypatch):
# Music video: no source album/year, but album tag is the literal 'NA' -> Unknown Album.
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"title": "Some Live Thing", "uploader": "Chan"})
audio = _FakeAudio({"artist": ["X"], "title": ["Y"], "album": ["NA"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Y [{YT_ID}].opus", "youtube", dry_run=False)
assert "album=Unknown Album" in changed
assert audio["album"] == ["Unknown Album"]
def test_repair_file_renames_bogus_filename(tmp_path, monkeypatch):
d = tmp_path / "Audioslave" / "youtube"
d.mkdir(parents=True)
f = d / f"NA [{YT_ID}].opus"
f.write_text("x")
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
"album": "Audioslave", "release_year": 2002})
audio = _FakeAudio({"artist": ["Audioslave"], "title": ["NA"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(str(f), "youtube", dry_run=False)
assert (d / f"Cochise [{YT_ID}].opus").exists()
assert not f.exists()
assert any("rename" in c.lower() or c.startswith("title=") for c in changed)
def test_repair_file_dry_run_does_not_rename(tmp_path, monkeypatch):
d = tmp_path / "Audioslave" / "youtube"
d.mkdir(parents=True)
f = d / f"NA [{YT_ID}].opus"
f.write_text("x")
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
"album": "Audioslave", "release_year": 2002})
monkeypatch.setattr(mf, "_open_audio", lambda path: (_FakeAudio({"title": ["NA"]}), None))
mf.repair_file(str(f), "youtube", dry_run=True)
assert f.exists() # untouched in dry-run
assert not (d / f"Cochise [{YT_ID}].opus").exists()
def test_fs_safe_replaces_slash():
assert "/" not in mf._fs_safe("AC/DC Live")
# ---- parallel repair ----
def test_repair_library_parallel_visits_all(tmp_path, monkeypatch):
root = tmp_path
n = 50
for i in range(n):
d = root / f"Artist{i}" / "youtube"
d.mkdir(parents=True)
(d / f"T{i} [{YT_ID}].opus").write_text("x")
import threading
seen = set()
lock = threading.Lock()
def fake(path, source, dry_run):
with lock:
seen.add(path)
return ["album=X"]
monkeypatch.setattr(mf, "repair_file", fake)
scanned, changed = mf.repair_library(str(root), dry_run=False, workers=8)
assert scanned == n and changed == n
assert len(seen) == n
def test_repair_library_default_workers_still_works(tmp_path, monkeypatch):
root = tmp_path
(root / "A" / "youtube").mkdir(parents=True)
(root / "A" / "youtube" / f"T [{YT_ID}].opus").write_text("x")
monkeypatch.setattr(mf, "repair_file", lambda p, s, d: ["x"])
assert mf.repair_library(str(root), dry_run=False) == (1, 1)
# ---- cookies + error visibility ----
def test_cookie_args_file_takes_precedence(monkeypatch):
monkeypatch.setattr(mf, "COOKIES_FILE", "/c.txt")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "firefox")
assert mf._cookie_args() == ["--cookies", "/c.txt"]
def test_cookie_args_browser(monkeypatch):
monkeypatch.setattr(mf, "COOKIES_FILE", "")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "firefox")
assert mf._cookie_args() == ["--cookies-from-browser", "firefox"]
def test_cookie_args_none(monkeypatch):
monkeypatch.setattr(mf, "COOKIES_FILE", "")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
assert mf._cookie_args() == []
def test_metadata_fetch_passes_cookies(monkeypatch):
captured = {}
class _R:
stdout = '{"title": "x"}'
monkeypatch.setattr(mf, "COOKIES_FILE", "/cookies.txt")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _R())
mf.run_yt_dlp_get_metadata("http://u")
assert "--cookies" in captured["cmd"]
assert "/cookies.txt" in captured["cmd"]
def test_metadata_fetch_logs_stderr(monkeypatch, capsys):
def boom(cmd, **k):
raise mf.subprocess.CalledProcessError(
1, cmd, output="", stderr="WARNING: foo\nERROR: Sign in to confirm you're not a bot.")
monkeypatch.setattr(mf.subprocess, "run", boom)
assert mf.run_yt_dlp_get_metadata("http://u") is None
out = capsys.readouterr().err
assert "not a bot" in out # the actionable last stderr line surfaces