16 Commits

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

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

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

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

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

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

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

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

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 11:13:52 -07:00
13 changed files with 1521 additions and 38 deletions

1
.gitignore vendored
View File

@@ -2,3 +2,4 @@ __pycache__/
*.pyc
server/log.txt
cookies.txt

View File

@@ -8,6 +8,10 @@ whatever you choose. It accepts:
- A **free-form query**: an artist, an album, a track title, or combos like
`"Artist - Title"` or `"Artist - Album"` (e.g. `"ODESZA - Bloom"`, `"Daft Punk"`, `"Discovery"`).
- A **URL** (e.g. `"https://music.youtube.com/watch?v=..."` or a regular YouTube URL).
- **Any streaming link** (Spotify, Apple Music, Tidal, Deezer, …): resolved to
metadata via [Odesli/song.link](https://odesli.co), then searched on Lidarr
first and downloaded from the matching YouTube track if no Lidarr release is
available. YouTube and SoundCloud links download directly.
Lidarr is tried first by default. If you pick a Lidarr album but **no indexer
release is available**, MusicFetch automatically falls through to the top
@@ -95,6 +99,9 @@ export LIDARR_API_KEY="your-lidarr-api-key"
| `-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. |
@@ -137,9 +144,73 @@ but it never overwrites a genuine existing artist/title with a channel name or d
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`.
It re-queries the source over the network, so run it occasionally, not constantly. Requires
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

View File

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

View File

@@ -0,0 +1,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.

View File

@@ -47,6 +47,45 @@ HEADERS = {"X-Api-Key": API_KEY, "Content-Type": "application/json"}
# Runtime flags, populated in main().
DEBUG = False
# yt-dlp cookies — authenticated requests bypass YouTube's bot-check ("Sign in
# to confirm you're not a bot") and lift rate limits, which is essential for
# bulk --repair. Set via CLI (--cookies / --cookies-from-browser) or env so the
# REST API container can supply them too.
COOKIES_FILE = os.environ.get("YTDLP_COOKIES", "")
COOKIES_FROM_BROWSER = os.environ.get("YTDLP_COOKIES_FROM_BROWSER", "")
# Per-request throttle. YouTube rate-limits a session after a burst of metadata
# fetches ("This content isn't available, try again later"), tripping even
# single-worker runs after a while. Sleeping between requests (and randomly
# before each download) keeps the session under the limit. Seconds; 0 disables.
# Tunable via env or --sleep; default 1s mirrors yt-dlp's own `-t sleep` advice.
SLEEP_REQUESTS = os.environ.get("YTDLP_SLEEP", "1")
def _cookie_args() -> list:
"""yt-dlp cookie flags (file wins over browser); empty when neither is set."""
if COOKIES_FILE:
return ["--cookies", COOKIES_FILE]
if COOKIES_FROM_BROWSER:
return ["--cookies-from-browser", COOKIES_FROM_BROWSER]
return []
def _sleep_args(download: bool = False) -> list:
"""yt-dlp throttle flags; empty when SLEEP_REQUESTS<=0. Sleeps between
extraction HTTP requests; for downloads also adds a randomized pre-download
interval (secs..2*secs) to further space out hits to YouTube."""
try:
secs = float(SLEEP_REQUESTS)
except (TypeError, ValueError):
secs = 0.0
if secs <= 0:
return []
args = ["--sleep-requests", str(secs)]
if download:
args += ["--sleep-interval", str(secs), "--max-sleep-interval", str(secs * 2)]
return args
# Quality choices for --quality.
QUALITY_CHOICES = ["best", "320", "m4a", "opus", "flac"]
@@ -79,6 +118,21 @@ class Hit:
return self.title or self.album or self.artist
@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"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@@ -337,7 +391,7 @@ def _ytmusic_search(query: str, limit: int) -> list[Hit]:
def _ytdlp_search(query: str, limit: int) -> list[Hit]:
try:
result = subprocess.run(
["yt-dlp", "--flat-playlist", "-J", f"ytsearch{limit}:{query}"],
["yt-dlp", *_cookie_args(), *_sleep_args(), "--flat-playlist", "-J", f"ytsearch{limit}:{query}"],
capture_output=True, text=True, check=True,
)
data = json.loads(result.stdout)
@@ -608,6 +662,8 @@ def _quality_args(quality: str) -> list[str]:
def yt_download(url_or_query: str, target_folder: Optional[str], quality: str, dry_run: bool,
hit: Optional[Hit] = None, outtmpl: Optional[str] = None):
cmd = ["yt-dlp",
*_cookie_args(),
*_sleep_args(download=True),
*_quality_args(quality),
"--embed-metadata",
"--embed-thumbnail",
@@ -714,6 +770,19 @@ def _is_youtube_playlist_url(url: str) -> bool:
return "list" in qs and "v" not in qs
_DIRECT_DOMAINS = ("youtube.com", "youtu.be", "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. Matches on
a label boundary so look-alikes (notyoutube.com) don't slip through."""
if not is_url(url):
return False
host = (urlparse(url).hostname or "").lower()
return any(host == d or host.endswith("." + d) for d in _DIRECT_DOMAINS)
def _ytmusic_playlist(pid: str) -> tuple[str, list[Hit]]:
"""Expand a YouTube Music playlist via ytmusicapi. Returns ("", []) on failure."""
try:
@@ -745,6 +814,32 @@ def _entry_to_hit(entry: dict) -> Hit:
"extractor": source})
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
def probe_url(url: str) -> tuple[str, str, list[Hit]]:
"""Classify a URL via yt-dlp. Returns (kind, title, hits) where kind is
'playlist' (hits populated) or 'track' (hits empty; caller downloads the URL).
@@ -756,7 +851,7 @@ def probe_url(url: str) -> tuple[str, str, list[Hit]]:
if hits:
return "playlist", title, hits
try:
result = subprocess.run(["yt-dlp", "--flat-playlist", "-J", url],
result = subprocess.run(["yt-dlp", *_cookie_args(), *_sleep_args(), "--flat-playlist", "-J", url],
capture_output=True, text=True, check=True)
data = json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
@@ -795,12 +890,19 @@ def download_single(url: str, root: str, quality: str, dry_run: bool) -> dict:
def run_yt_dlp_get_metadata(url: str, extra_args=None) -> Optional[dict]:
cmd = ["yt-dlp", "-j", "--no-playlist", *(extra_args or []), url]
cmd = ["yt-dlp", *_cookie_args(), *_sleep_args(), "-j", "--no-playlist", *(extra_args or []), url]
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
return json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
err(f"yt-dlp metadata extraction failed: {e}")
# Surface yt-dlp's own last stderr line (e.g. 429 / "not a bot") instead
# of a bare exit code — the actual reason is what you need to act on.
detail = ""
stderr = getattr(e, "stderr", "") or ""
lines = [ln for ln in stderr.strip().splitlines() if ln.strip()]
if lines:
detail = f" — {lines[-1]}"
err(f"yt-dlp metadata extraction failed for {url}{detail}")
return None
@@ -818,6 +920,46 @@ def get_artist_from_metadata(meta: dict) -> str:
return "Unknown Artist"
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
Lidarr-first, YouTube-fallback hits for "Artist - Title". Odesli frequently
omits a YouTube link, so we always run a normal youtube_search for the
fallback; when Odesli DID supply a link, its exact track is inserted as the
preferred (first) YouTube hit. Raises OdesliError if the link can't resolve."""
r = odesli_resolve(url)
if r is None:
raise OdesliError(url)
query = f"{r.artist} - {r.title}".strip(" -")
hits = build_combined_hits(query, limit, yt_first=False,
lidarr_only=False, yt_only=False)
if r.youtube_url:
exact = Hit(source="youtube", kind="track", title=r.title,
artist=r.artist, thumbnail=r.thumb,
payload={"url": r.youtube_url})
idx = next((i for i, h in enumerate(hits) if h.source == "youtube"), len(hits))
hits = hits[:idx] + [exact] + hits[idx:]
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, False, False)
def handle_url(url: str, root: str, quality: str, dry_run: bool):
kind, title, hits = probe_url(url)
if kind == "playlist":
@@ -1020,19 +1162,31 @@ def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
return changed
def repair_library(root: str, dry_run: bool, exclude=()) -> tuple[int, int]:
"""Walk <root>/<artist>/<source>/ and re-tag audio files. Returns (scanned, changed)."""
def repair_library(root: str, dry_run: bool, exclude=(), workers: int = 8) -> tuple[int, int]:
"""Walk <root>/<artist>/<source>/ and re-tag audio files. Returns (scanned, changed).
Each file is an independent yt-dlp network round-trip, so they run in a
thread pool (network-bound); `workers` caps concurrency. Each thread owns
its own file + request, so no shared state needs locking beyond the counts.
Lower `workers` if YouTube starts rate-limiting (HTTP 429/403)."""
if not os.path.isdir(root):
err(f"Root folder not found: {root}")
return 0, 0
scanned = changed = 0
for path, source, _artist in _iter_source_files(root, exclude):
scanned += 1
def _one(path, source):
try:
if repair_file(path, source, dry_run):
changed += 1
return bool(repair_file(path, source, dry_run))
except Exception as e: # noqa: BLE001 — one bad file shouldn't abort
err(f"repair failed ({os.path.basename(path)}): {e}")
return False
scanned = changed = 0
files = ((p, s) for p, s, _a in _iter_source_files(root, exclude))
with ThreadPoolExecutor(max_workers=max(1, workers)) as ex:
for ok in ex.map(lambda ps: _one(*ps), files):
scanned += 1
changed += int(ok)
if scanned % 100 == 0:
print(f"… {scanned} scanned, {changed} changed", flush=True)
verb = "Would repair" if dry_run else "Repaired"
print(f"{verb} {changed}/{scanned} files")
return scanned, changed
@@ -1171,6 +1325,19 @@ def parse_args():
help="Search all albums when adding an artist to Lidarr.")
p.add_argument("--repair", action="store_true",
help="Re-tag existing downloads under --root from source metadata.")
p.add_argument("--workers", type=int, default=4,
help="Parallel yt-dlp metadata fetches during --repair (default 4; "
"raise with cookies, lower if YouTube rate-limits).")
p.add_argument("--sleep", type=float, metavar="SECS",
help="Seconds to sleep between yt-dlp requests (0 disables). "
"Avoids YouTube rate-limiting on bulk runs. Overrides "
"$YTDLP_SLEEP (default 1).")
p.add_argument("--cookies", metavar="FILE",
help="Path to a yt-dlp cookies.txt (authenticated requests avoid "
"YouTube's bot-check / rate limits). Overrides $YTDLP_COOKIES.")
p.add_argument("--cookies-from-browser", metavar="BROWSER",
help="Load YouTube cookies from a local browser, e.g. firefox or "
"chrome. Overrides $YTDLP_COOKIES_FROM_BROWSER.")
p.add_argument("--retag-from-path", action="store_true",
help="Offline: re-tag artist/title from folder + filename "
"(fixes tags damaged by a prior --repair).")
@@ -1181,10 +1348,36 @@ def parse_args():
return p.parse_args()
def _dispatch_chosen(chosen: Hit, hits: list[Hit], root: str, quality: str,
dry_run: bool, lidarr_only: bool, search_all: 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, 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)
def main():
global DEBUG
global DEBUG, COOKIES_FILE, COOKIES_FROM_BROWSER, SLEEP_REQUESTS
args = parse_args()
DEBUG = args.debug
if args.sleep is not None:
SLEEP_REQUESTS = args.sleep
if args.cookies:
COOKIES_FILE = args.cookies
if args.cookies_from_browser:
COOKIES_FROM_BROWSER = args.cookies_from_browser
query = " ".join(args.query).strip()
if args.retag_from_path:
@@ -1192,7 +1385,7 @@ def main():
return
if args.repair:
repair_library(args.root, args.dry_run, args.exclude)
repair_library(args.root, args.dry_run, args.exclude, args.workers)
return
if not query:
@@ -1204,7 +1397,11 @@ def main():
sys.exit(1)
if is_url(query):
handle_url(query, args.root, args.quality, args.dry_run)
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
hits = build_combined_hits(query, args.limit, args.ytsearch,
@@ -1218,21 +1415,8 @@ def main():
print("Nothing selected.")
return
if chosen.source == "lidarr":
if chosen.kind == "album":
handled = act_lidarr_album(chosen, args.root, args.search_all, args.dry_run)
if not handled and not args.lidarr_only:
# Fall through to the top YouTube hit for the same query.
yt_fallback = next((h for h in hits if h.source == "youtube"), None)
if yt_fallback:
print("Using top YouTube hit as fallback.")
act_youtube(yt_fallback, args.root, args.quality, args.dry_run)
else:
print("No YouTube fallback available.")
else:
act_lidarr_artist(chosen, args.root, args.search_all, args.dry_run)
else:
act_youtube(chosen, args.root, args.quality, args.dry_run)
_dispatch_chosen(chosen, hits, args.root, args.quality, args.dry_run,
args.lidarr_only, args.search_all)
if __name__ == "__main__":

View File

@@ -53,16 +53,39 @@ def fetch(q: str = Query(..., min_length=1),
raise HTTPException(status_code=422, detail=f"Invalid quality '{quality}'.")
if mf.is_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))
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)
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.",
lambda: actions.perform_fetch(chosen, hits, quality, ROOT),
done_message=actions.done_message(chosen),
fail_message=actions.failed_message(chosen),
)
return response

View File

@@ -13,5 +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
# Uncomment and set host path to supply cookies (see YTDLP_COOKIES above):
# - /path/to/cookies.txt:/cookies.txt:ro

View File

@@ -28,7 +28,14 @@ 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",
"is_url", "probe_url", "download_hits", "download_single"]
"is_url", "probe_url", "download_hits", "download_single",
"_is_direct_url", "odesli_resolve", "resolve_link_hits",
"handle_link", "OdesliError"]

View File

@@ -73,3 +73,43 @@ def test_search_query_still_works(client, auth, monkeypatch):
r = client.post("/fetch", params={"q": "Daft Punk - Discovery"}, headers=auth)
assert r.status_code == 200
assert r.json()["status"] == "queued"
def test_non_direct_link_resolves_and_fetches(client, auth, monkeypatch):
from server import mf
lid = mf.Hit(source="lidarr", kind="album", title="A Moment Apart",
artist="ODESZA", album="A Moment Apart", year="2017",
payload={"album": {"id": 9}})
yt = mf.Hit(source="youtube", kind="track", title="Bloom", artist="ODESZA",
payload={"url": "https://music.youtube.com/watch?v=YYY"})
monkeypatch.setattr("server.app.mf.resolve_link_hits",
lambda url, limit: ("ODESZA - Bloom", [lid, yt]))
monkeypatch.setattr("server.app.mf.pick", lambda hits, q, ni, yf: hits[0])
monkeypatch.setattr("server.app.actions.perform_fetch",
lambda chosen, hits, quality, root: {"path": None, "lidarr_album_id": 9})
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
assert r.status_code == 200
body = r.json()
assert body["status"] == "queued"
assert body["hit"]["album"] == "A Moment Apart"
done = _wait_done(client, auth, body["job_id"])
assert done["status"] == "done"
def test_non_direct_link_resolve_failure_422(client, auth, monkeypatch):
from server import mf as mf_mod
def boom(url, limit):
raise mf_mod.OdesliError(url)
monkeypatch.setattr("server.app.mf.resolve_link_hits", boom)
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/bad"}, headers=auth)
assert r.status_code == 422
assert "resolve" in r.json()["message"].lower()
def test_non_direct_link_no_hits_404(client, auth, monkeypatch):
monkeypatch.setattr("server.app.mf.resolve_link_hits",
lambda url, limit: ("ODESZA - Bloom", []))
r = client.post("/fetch", params={"q": "https://open.spotify.com/track/abc"}, headers=auth)
assert r.status_code == 404

View File

@@ -6,3 +6,11 @@ def test_url_helpers_reexported():
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)

169
tests/test_odesli.py Normal file
View File

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

View File

@@ -132,3 +132,13 @@ def test_yt_download_single_word_tags_injected_literally(monkeypatch):
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"]

View File

@@ -336,3 +336,76 @@ def test_repair_file_dry_run_does_not_rename(tmp_path, monkeypatch):
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