Compare commits
48 Commits
feat/rest-
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| b26e321926 | |||
| 47f3482192 | |||
| a4b5039e7f | |||
| b899d75930 | |||
| 32acd038c8 | |||
| eb45a3680f | |||
| 44aaa1f93e | |||
| 8daf780023 | |||
| 9c308fefc7 | |||
| a4e1dc1643 | |||
| 9fccf9015a | |||
| a88f4c594a | |||
| 95a448ef58 | |||
| dcb3014fb0 | |||
| 140bfef7c9 | |||
| 92742b9ad6 | |||
| 0347a638cf | |||
| 33ca743a34 | |||
| 7951c436dd | |||
| 8b881c14bf | |||
| 530b5b0406 | |||
| 0a4e6d474a | |||
| c0503187c5 | |||
| a6aa469084 | |||
| f071158c10 | |||
| c6bde6958a | |||
| 7ea3ad2538 | |||
| 9af7f91a25 | |||
| 567d7578ad | |||
| c6e28a4f75 | |||
| 1a81f64cc3 | |||
| fdc3cc84a5 | |||
| 74eb63b243 | |||
| 6730f1f141 | |||
| f103b6c253 | |||
| 7309ad3a29 | |||
| 90b9a01872 | |||
| 0f7ddd7697 | |||
| ca36d2bb27 | |||
| aa9d177ed1 | |||
| 3ee49b17bd | |||
| 6e6bec7a0d | |||
| a24c894c61 | |||
| a424fbfd2f | |||
| b99e5eb9cb | |||
| 1661cb1742 | |||
| 18f72a5626 | |||
| babbd84fda |
3
.gitignore
vendored
3
.gitignore
vendored
@@ -1,2 +1,5 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
server/log.txt
|
||||
cookies.txt
|
||||
|
||||
122
README.md
122
README.md
@@ -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)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
@@ -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`.
|
||||
743
docs/superpowers/plans/2026-06-13-odesli-link-resolution.md
Normal file
743
docs/superpowers/plans/2026-06-13-odesli-link-resolution.md
Normal 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.
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
958
musicfetch
958
musicfetch
File diff suppressed because it is too large
Load Diff
@@ -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
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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}'.")
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
14
server/mf.py
14
server/mf.py
@@ -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"]
|
||||
|
||||
@@ -4,3 +4,4 @@ requests
|
||||
ytmusicapi
|
||||
rich
|
||||
yt-dlp
|
||||
mutagen
|
||||
|
||||
115
tests/test_api_url.py
Normal file
115
tests/test_api_url.py
Normal 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
|
||||
@@ -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
110
tests/test_lidarr_search.py
Normal 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
|
||||
16
tests/test_mf_url_exports.py
Normal file
16
tests/test_mf_url_exports.py
Normal 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)
|
||||
93
tests/test_multiplatform.py
Normal file
93
tests/test_multiplatform.py
Normal 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
96
tests/test_musicbrainz.py
Normal 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
169
tests/test_odesli.py
Normal 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
144
tests/test_playlist.py
Normal 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
43
tests/test_profiles.py
Normal 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
411
tests/test_repair.py
Normal 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
|
||||
Reference in New Issue
Block a user