Compare commits
16 Commits
fix/yt-tag
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| b26e321926 | |||
| 47f3482192 | |||
| a4b5039e7f | |||
| b899d75930 | |||
| 32acd038c8 | |||
| eb45a3680f | |||
| 44aaa1f93e | |||
| 8daf780023 | |||
| 9c308fefc7 | |||
| a4e1dc1643 | |||
| 9fccf9015a | |||
| a88f4c594a | |||
| 95a448ef58 | |||
| dcb3014fb0 | |||
| 140bfef7c9 | |||
| 92742b9ad6 |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -2,3 +2,4 @@ __pycache__/
|
||||
*.pyc
|
||||
|
||||
server/log.txt
|
||||
cookies.txt
|
||||
|
||||
73
README.md
73
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
|
||||
@@ -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
|
||||
|
||||
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,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.
|
||||
242
musicfetch
242
musicfetch
@@ -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__":
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
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"
|
||||
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user