Compare commits
28 Commits
7ea3ad2538
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| fc2d082a25 | |||
| 82bc524155 | |||
| b26e321926 | |||
| 47f3482192 | |||
| a4b5039e7f | |||
| b899d75930 | |||
| 32acd038c8 | |||
| eb45a3680f | |||
| 44aaa1f93e | |||
| 8daf780023 | |||
| 9c308fefc7 | |||
| a4e1dc1643 | |||
| 9fccf9015a | |||
| a88f4c594a | |||
| 95a448ef58 | |||
| dcb3014fb0 | |||
| 140bfef7c9 | |||
| 92742b9ad6 | |||
| 0347a638cf | |||
| 33ca743a34 | |||
| 7951c436dd | |||
| 8b881c14bf | |||
| 530b5b0406 | |||
| 0a4e6d474a | |||
| c0503187c5 | |||
| a6aa469084 | |||
| f071158c10 | |||
| c6bde6958a |
4
.gitignore
vendored
4
.gitignore
vendored
@@ -1,2 +1,6 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
server/log.txt
|
||||
server/.env
|
||||
cookies.txt
|
||||
|
||||
95
README.md
95
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,7 +99,12 @@ 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`). |
|
||||
| `--remote-components SPEC` | yt-dlp EJS challenge-solver source (default `ejs:github`; needs `deno`). Fixes "signature/n challenge solving failed". `''` to disable. |
|
||||
| `--retag-from-path` | Offline: re-tag artist/title from folder + filename (see below). |
|
||||
| `-x`, `--exclude NAME` | Folder under `--root` to skip during `--repair`/`--retag-from-path` (repeatable). |
|
||||
| `--debug` | Verbose output. |
|
||||
|
||||
### Examples
|
||||
@@ -130,13 +139,88 @@ file using the `[id]` in its filename, and fixes tags. Useful when downloads lan
|
||||
missing album or wrong year.
|
||||
|
||||
It is deliberately **conservative**: it overwrites **album** and **year** (the usual
|
||||
breakage), but only *fills in* a missing **artist**/**title** — it never overwrites existing
|
||||
artist/title with channel names or decorated video titles. Files whose source returns no real
|
||||
music metadata (no album and no release year — e.g. plain music videos) are left untouched.
|
||||
breakage), and fills in **artist**/**title** when they are missing *or* a known-bogus
|
||||
placeholder (`NA`, `Unknown Album`, `Unknown Artist` — left behind by older buggy tagging) —
|
||||
but it never overwrites a genuine existing artist/title with a channel name or decorated video
|
||||
title. A bogus `NA [<id>].<ext>` filename is renamed to the recovered title, and a literal
|
||||
`NA` album with no source album is normalised to `Unknown Album`.
|
||||
|
||||
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.
|
||||
|
||||
**JS challenge solving (not a cookie issue).** Separately from auth, YouTube gates media formats
|
||||
behind signature / "n" JS challenges. If yt-dlp warns *"Signature solving failed"* / *"n challenge
|
||||
solving failed: Some formats may be missing"*, downloads break even with valid cookies. musicfetch
|
||||
passes `--remote-components ejs:github` by default so yt-dlp fetches the [EJS](https://github.com/yt-dlp/yt-dlp/wiki/EJS)
|
||||
solver and runs it via a local JS runtime — install **[`deno`](https://deno.land)** and use a recent
|
||||
yt-dlp. Override with `--remote-components` / `$YTDLP_REMOTE_COMPONENTS` (empty disables).
|
||||
|
||||
#### 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, just point it at the host file via `.env` — no compose
|
||||
edits needed (see `server/.env.example`):
|
||||
|
||||
```bash
|
||||
# server/.env
|
||||
YTDLP_COOKIES_HOST=/host/path/cookies.txt
|
||||
```
|
||||
|
||||
The compose mounts it read-only at `/cookies.txt` and sets `YTDLP_COOKIES`
|
||||
automatically. Leave `YTDLP_COOKIES_HOST` unset to run without cookies.
|
||||
|
||||
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
|
||||
@@ -154,6 +238,9 @@ artist/title and leaves album/year alone.
|
||||
```bash
|
||||
./musicfetch --retag-from-path -d # preview
|
||||
./musicfetch --retag-from-path -o /media/music
|
||||
|
||||
# Skip folders (e.g. hand-curated playlists you don't want re-tagged)
|
||||
./musicfetch --repair -x Unsorted -x playlists
|
||||
```
|
||||
|
||||
### 📁 Output Structure
|
||||
|
||||
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.
|
||||
434
musicfetch
434
musicfetch
@@ -18,7 +18,8 @@ from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
import requests
|
||||
from requests.exceptions import RequestException
|
||||
from requests.exceptions import ConnectionError as ReqConnectionError
|
||||
from requests.exceptions import RequestException, Timeout
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
|
||||
# Optional deps — degrade gracefully if missing.
|
||||
@@ -46,6 +47,60 @@ 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")
|
||||
|
||||
# yt-dlp JS-challenge solver. YouTube now gates formats behind signature / "n"
|
||||
# challenges that yt-dlp's bundled solver can't reliably crack; without the EJS
|
||||
# remote-component solver script it warns "Signature/n challenge solving failed:
|
||||
# Some formats may be missing" and downloads break (looks like a cookie problem,
|
||||
# isn't). Enabling ejs:github makes yt-dlp fetch the solver and run it via a local
|
||||
# JS runtime (deno). Needs a recent yt-dlp + deno present. Set empty to disable.
|
||||
REMOTE_COMPONENTS = os.environ.get("YTDLP_REMOTE_COMPONENTS", "ejs:github")
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
def _ejs_args() -> list:
|
||||
"""yt-dlp EJS challenge-solver flags; empty when REMOTE_COMPONENTS is unset."""
|
||||
if REMOTE_COMPONENTS:
|
||||
return ["--remote-components", REMOTE_COMPONENTS]
|
||||
return []
|
||||
|
||||
# Quality choices for --quality.
|
||||
QUALITY_CHOICES = ["best", "320", "m4a", "opus", "flac"]
|
||||
|
||||
@@ -78,6 +133,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
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -236,11 +306,22 @@ def lidarr_search(query: str, limit: int) -> list[Hit]:
|
||||
return _fallback_lookup(query, limit, artist_first=True)
|
||||
|
||||
|
||||
def _log_lidarr_failure(label: str, e: Exception) -> None:
|
||||
"""A connection/timeout error means Lidarr is unreachable — the silent
|
||||
YouTube fallback that follows is easy to mistake for "Lidarr had no match",
|
||||
so surface it loudly. Ordinary HTTP errors stay debug-only."""
|
||||
if isinstance(e, (ReqConnectionError, Timeout)):
|
||||
err(f"Lidarr unreachable ({label} at {LIDARR_URL}): {e}. "
|
||||
f"Falling back to YouTube.")
|
||||
else:
|
||||
dbg(f"{label} failed: {e}")
|
||||
|
||||
|
||||
def _lidarr_album_candidates(term: str) -> list[Hit]:
|
||||
try:
|
||||
return [_album_to_hit(a) for a in lidarr_get("/api/v1/album/lookup", params={"term": term})]
|
||||
except RequestException as e:
|
||||
dbg(f"album/lookup failed: {e}")
|
||||
_log_lidarr_failure("album/lookup", e)
|
||||
return []
|
||||
|
||||
|
||||
@@ -248,7 +329,7 @@ def _lidarr_artist_candidates(term: str) -> list[Hit]:
|
||||
try:
|
||||
return [_artist_to_hit(a) for a in lidarr_get("/api/v1/artist/lookup", params={"term": term})]
|
||||
except RequestException as e:
|
||||
dbg(f"artist/lookup failed: {e}")
|
||||
_log_lidarr_failure("artist/lookup", e)
|
||||
return []
|
||||
|
||||
|
||||
@@ -325,7 +406,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(), *_ejs_args(), *_sleep_args(), "--flat-playlist", "-J", f"ytsearch{limit}:{query}"],
|
||||
capture_output=True, text=True, check=True,
|
||||
)
|
||||
data = json.loads(result.stdout)
|
||||
@@ -596,6 +677,9 @@ 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(),
|
||||
*_ejs_args(),
|
||||
*_sleep_args(download=True),
|
||||
*_quality_args(quality),
|
||||
"--embed-metadata",
|
||||
"--embed-thumbnail",
|
||||
@@ -605,19 +689,34 @@ def yt_download(url_or_query: str, target_folder: Optional[str], quality: str, d
|
||||
cmd += ["-o", outtmpl]
|
||||
else:
|
||||
cmd += ["-P", target_folder]
|
||||
# Override tags from the chosen hit so they don't rely on scraped titles.
|
||||
# Override embedded tags from the chosen hit. Inject literals via a
|
||||
# seed-then-replace pair: --parse-metadata first copies an always-present
|
||||
# field into meta_<tag> (so the tag exists even when the source lacks it,
|
||||
# e.g. YouTube videos with no album), then --replace-in-metadata overwrites
|
||||
# it with the literal value. This dodges yt-dlp's output-template trap where
|
||||
# a bare-word FROM (e.g. "Cochise") matches field_to_template's r'[a-zA-Z_]+$'
|
||||
# and is read as a *field name* -> "NA". --replace-in-metadata args are
|
||||
# literal, so single-word values and parens survive intact.
|
||||
def _force_tag(field: str, value: str) -> list[str]:
|
||||
repl = value.replace("\\", r"\\") # backslash is special in re.sub repl
|
||||
return ["--parse-metadata", f"%(title,id)s:%(meta_{field})s",
|
||||
"--replace-in-metadata", f"meta_{field}", "^.*$", repl]
|
||||
|
||||
if hit:
|
||||
if hit.artist:
|
||||
# First artist only; anchored ^.*$ replaces the whole field exactly once
|
||||
# (a bare .* matches twice and doubles the value).
|
||||
primary_artist = hit.artist.split(",")[0].strip()
|
||||
cmd += ["--replace-in-metadata", "artist", "^.*$", primary_artist]
|
||||
if hit.album:
|
||||
cmd += ["--parse-metadata", f"{hit.album}:%(album)s"]
|
||||
primary_artist = hit.artist.split(",")[0].strip() if hit.artist else ""
|
||||
if primary_artist:
|
||||
cmd += _force_tag("artist", primary_artist)
|
||||
if hit.title:
|
||||
cmd += ["--parse-metadata", f"{hit.title}:%(title)s"]
|
||||
cmd += _force_tag("title", hit.title)
|
||||
if hit.album:
|
||||
cmd += _force_tag("album", hit.album)
|
||||
if hit.year:
|
||||
cmd += ["--parse-metadata", f"{hit.year}:%(release_year)s"]
|
||||
# When the hit carried no album, still embed one: the resolved/native album
|
||||
# if present, else a placeholder so players (e.g. Plexamp) don't choke on a
|
||||
# blank album. (A hit album is already forced above and must not be clobbered.)
|
||||
if not (hit and hit.album):
|
||||
cmd += ["--parse-metadata", "%(album|Unknown Album)s:%(meta_album)s"]
|
||||
cmd.append(url_or_query)
|
||||
|
||||
dest = outtmpl or target_folder
|
||||
@@ -687,6 +786,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:
|
||||
@@ -718,6 +830,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).
|
||||
@@ -729,7 +867,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(), *_ejs_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:
|
||||
@@ -767,18 +905,28 @@ def download_single(url: str, root: str, quality: str, dry_run: bool) -> dict:
|
||||
return {"title": title, "artist": artist, "ok": ok}
|
||||
|
||||
|
||||
def run_yt_dlp_get_metadata(url: str) -> Optional[dict]:
|
||||
def run_yt_dlp_get_metadata(url: str, extra_args=None) -> Optional[dict]:
|
||||
cmd = ["yt-dlp", *_cookie_args(), *_ejs_args(), *_sleep_args(), "-j", "--no-playlist", *(extra_args or []), url]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["yt-dlp", "-j", "--no-playlist", url],
|
||||
capture_output=True, text=True, check=True,
|
||||
)
|
||||
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
|
||||
|
||||
|
||||
# Repair only reads tags — skip YouTube's slow/throttled JS signature step
|
||||
# (we never download here), which keeps metadata but is far faster per file.
|
||||
_REPAIR_META_ARGS = ["--extractor-args", "youtube:player_skip=js"]
|
||||
|
||||
|
||||
def get_artist_from_metadata(meta: dict) -> str:
|
||||
for key in ("artist", "creator", "uploader", "channel"):
|
||||
if meta.get(key):
|
||||
@@ -788,6 +936,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":
|
||||
@@ -875,12 +1063,53 @@ def _read_tag(audio, key_map, field: str) -> str:
|
||||
return str(val[0]) if isinstance(val, list) else str(val)
|
||||
|
||||
|
||||
# Placeholder tag values the old tagging bug left behind (yt-dlp's "NA" missing
|
||||
# marker, and the "Unknown *" fallbacks). Treated as empty so repair overwrites
|
||||
# them rather than mistaking them for a real, present tag.
|
||||
_BOGUS_TAGS = {"", "na", "n/a", "unknown", "unknown album", "unknown artist"}
|
||||
|
||||
|
||||
def _is_bogus(value: str) -> bool:
|
||||
return (value or "").strip().casefold() in _BOGUS_TAGS
|
||||
|
||||
|
||||
def _fs_safe(name: str) -> str:
|
||||
"""Filesystem-safe filename stem: mirror yt-dlp's default '/'->'⧸' so the
|
||||
path stays a single segment, and drop NULs."""
|
||||
return name.replace("/", "⧸").replace("\0", "").strip()
|
||||
|
||||
|
||||
def _maybe_rename_bogus(path: str, title: str, dry_run: bool) -> tuple[str, Optional[str]]:
|
||||
"""When the filename stem is a placeholder (e.g. 'NA [<id>]'), rename to
|
||||
'<title> [<id>].<ext>'. Returns (current_path, change_note_or_None)."""
|
||||
fname = os.path.basename(path)
|
||||
parsed = _parse_track_file(fname)
|
||||
if not parsed:
|
||||
return path, None
|
||||
stem_title, vid = parsed
|
||||
if not _is_bogus(stem_title) or _is_bogus(title):
|
||||
return path, None
|
||||
ext = fname.rsplit(".", 1)[-1]
|
||||
new_name = f"{_fs_safe(title)} [{vid}].{ext}"
|
||||
new_path = os.path.join(os.path.dirname(path), new_name)
|
||||
if new_path == path or not new_name:
|
||||
return path, None
|
||||
note = f"renamed -> {new_name}"
|
||||
if dry_run:
|
||||
print(f"[dry-run] would rename {fname} -> {new_name}")
|
||||
return path, note
|
||||
os.rename(path, new_path)
|
||||
print(f"renamed {fname} -> {new_name}")
|
||||
return new_path, note
|
||||
|
||||
|
||||
def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
|
||||
"""Re-tag one file from source metadata. Conservative: fixes album/year
|
||||
(the common breakage) and only fills MISSING artist/title — never clobbers
|
||||
existing tags with channel names or decorated music-video titles. Files whose
|
||||
source has no real music metadata (no album/year, e.g. music videos) are left
|
||||
untouched. Returns the list of changed fields."""
|
||||
"""Re-tag one file from source metadata. album/year are authoritative
|
||||
(overwrite); artist/title are filled when MISSING *or* a known-bogus
|
||||
placeholder ('NA', 'Unknown …') — the old tagging bug wrote those — but a
|
||||
genuine existing tag is never clobbered with a channel name or decorated
|
||||
music-video title. A bogus 'NA [<id>]' filename is renamed to the recovered
|
||||
title. Returns the list of changed fields."""
|
||||
parsed = _parse_track_file(os.path.basename(path))
|
||||
if not parsed:
|
||||
dbg(f"skip (no id): {path}")
|
||||
@@ -893,17 +1122,11 @@ def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
|
||||
if not url:
|
||||
dbg(f"skip (source '{source}' not re-queryable): {path}")
|
||||
return []
|
||||
meta = run_yt_dlp_get_metadata(url)
|
||||
meta = run_yt_dlp_get_metadata(url, _REPAIR_META_ARGS)
|
||||
if not meta:
|
||||
dbg(f"skip (no metadata): {path}")
|
||||
return []
|
||||
|
||||
album = (meta.get("album") or "").strip()
|
||||
year = _valid_year(meta)
|
||||
if not album and not year:
|
||||
dbg(f"skip (no music metadata, likely a video): {path}")
|
||||
return []
|
||||
|
||||
try:
|
||||
audio, key_map = _open_audio(path)
|
||||
except Exception as e: # noqa: BLE001
|
||||
@@ -912,18 +1135,28 @@ def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
|
||||
if audio is None:
|
||||
return []
|
||||
|
||||
# album/year are authoritative (overwrite); artist/title fill-missing only.
|
||||
album = (meta.get("album") or "").strip()
|
||||
year = _valid_year(meta)
|
||||
cur_artist = _read_tag(audio, key_map, "artist")
|
||||
cur_title = _read_tag(audio, key_map, "title")
|
||||
cur_album = _read_tag(audio, key_map, "album")
|
||||
meta_artist = get_artist_from_metadata(meta)
|
||||
meta_title = (meta.get("title") or "").strip()
|
||||
|
||||
updates = {}
|
||||
if album:
|
||||
updates["album"] = album
|
||||
elif cur_album and _is_bogus(cur_album) and cur_album.strip().casefold() != "unknown album":
|
||||
# No source album, but the tag is a literal 'NA' — normalise it so no
|
||||
# file keeps the placeholder (a blank album is left blank, as before).
|
||||
updates["album"] = "Unknown Album"
|
||||
if year:
|
||||
updates["date"] = year
|
||||
artist = get_artist_from_metadata(meta)
|
||||
if artist and artist != "Unknown Artist" and not _read_tag(audio, key_map, "artist"):
|
||||
updates["artist"] = artist
|
||||
title = meta.get("title", "")
|
||||
if title and not _read_tag(audio, key_map, "title"):
|
||||
updates["title"] = title
|
||||
# Fill artist/title when missing OR bogus; never overwrite a genuine value.
|
||||
if meta_artist and not _is_bogus(meta_artist) and _is_bogus(cur_artist):
|
||||
updates["artist"] = meta_artist
|
||||
if meta_title and not _is_bogus(meta_title) and _is_bogus(cur_title):
|
||||
updates["title"] = meta_title
|
||||
|
||||
changed = []
|
||||
for field, value in updates.items():
|
||||
@@ -936,35 +1169,59 @@ def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
|
||||
if changed:
|
||||
prefix = "[dry-run] would set" if dry_run else "set"
|
||||
print(f"{prefix} [{', '.join(changed)}] on {path}")
|
||||
|
||||
# Repair a placeholder filename using the final (recovered) title.
|
||||
final_title = updates.get("title") or cur_title
|
||||
_, rename_note = _maybe_rename_bogus(path, final_title, dry_run)
|
||||
if rename_note:
|
||||
changed.append(rename_note)
|
||||
return changed
|
||||
|
||||
|
||||
def repair_library(root: str, dry_run: bool) -> 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):
|
||||
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
|
||||
|
||||
|
||||
def _iter_source_files(root: str):
|
||||
def _iter_source_files(root: str, exclude=()):
|
||||
"""Yield (path, source, artist) for audio files under <root>/<artist>/<source>/
|
||||
where source is a yt-dlp source folder (Lidarr album folders are skipped)."""
|
||||
where source is a yt-dlp source folder (Lidarr album folders are skipped).
|
||||
Skips any artist or source folder whose name is in `exclude` (case-insensitive)."""
|
||||
skip = {e.lower() for e in exclude}
|
||||
for artist in sorted(os.listdir(root)):
|
||||
if artist.lower() in skip:
|
||||
continue
|
||||
adir = os.path.join(root, artist)
|
||||
if not os.path.isdir(adir):
|
||||
continue
|
||||
for source in sorted(os.listdir(adir)):
|
||||
if source.lower() in skip:
|
||||
continue
|
||||
sdir = os.path.join(adir, source)
|
||||
if not os.path.isdir(sdir) or not _is_source_dir(source):
|
||||
continue
|
||||
@@ -1029,13 +1286,13 @@ def retag_file_from_path(path: str, folder_artist: str, dry_run: bool) -> list[s
|
||||
return changed
|
||||
|
||||
|
||||
def retag_library_from_path(root: str, dry_run: bool) -> tuple[int, int]:
|
||||
def retag_library_from_path(root: str, dry_run: bool, exclude=()) -> tuple[int, int]:
|
||||
"""Re-tag artist/title offline from folder+filename for every source file."""
|
||||
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):
|
||||
for path, _source, artist in _iter_source_files(root, exclude):
|
||||
scanned += 1
|
||||
try:
|
||||
if retag_file_from_path(path, artist, dry_run):
|
||||
@@ -1084,25 +1341,73 @@ 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("--remote-components", metavar="SPEC",
|
||||
help="yt-dlp EJS challenge-solver source (default ejs:github; "
|
||||
"needs deno). Fixes 'signature/n challenge solving failed'. "
|
||||
"Pass '' to disable. Overrides $YTDLP_REMOTE_COMPONENTS.")
|
||||
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).")
|
||||
p.add_argument("-x", "--exclude", action="append", default=[], metavar="NAME",
|
||||
help="Folder name under --root to skip during --repair/--retag-from-path "
|
||||
"(repeatable, e.g. -x Unsorted -x playlists).")
|
||||
p.add_argument("--debug", action="store_true", help="Verbose output.")
|
||||
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, REMOTE_COMPONENTS
|
||||
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
|
||||
if args.remote_components is not None: # '' disables; None keeps env default
|
||||
REMOTE_COMPONENTS = args.remote_components
|
||||
query = " ".join(args.query).strip()
|
||||
|
||||
if args.retag_from_path:
|
||||
retag_library_from_path(args.root, args.dry_run)
|
||||
retag_library_from_path(args.root, args.dry_run, args.exclude)
|
||||
return
|
||||
|
||||
if args.repair:
|
||||
repair_library(args.root, args.dry_run)
|
||||
repair_library(args.root, args.dry_run, args.exclude, args.workers)
|
||||
return
|
||||
|
||||
if not query:
|
||||
@@ -1114,7 +1419,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,
|
||||
@@ -1128,21 +1437,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__":
|
||||
|
||||
13
server/.env.example
Normal file
13
server/.env.example
Normal file
@@ -0,0 +1,13 @@
|
||||
# Copy to server/.env and fill in. docker-compose reads it automatically.
|
||||
|
||||
# Lidarr (preferred source)
|
||||
LIDARR_API_KEY=
|
||||
|
||||
# API key clients must send to musicfetch-api
|
||||
MUSICFETCH_API_KEY=
|
||||
|
||||
# Optional: authenticated YouTube cookies to dodge the bot-check
|
||||
# ("Sign in to confirm you're not a bot") and rate limits.
|
||||
# Set to the host path of a Netscape-format cookies.txt (see README
|
||||
# "Getting YouTube cookies"). Leave unset to run without cookies.
|
||||
# YTDLP_COOKIES_HOST=/host/path/cookies.txt
|
||||
@@ -1,7 +1,20 @@
|
||||
FROM python:3.12-slim
|
||||
|
||||
# ffmpeg for audio extraction/embedding; deno is the JS runtime yt-dlp needs
|
||||
# for YouTube (without it: "No supported JavaScript runtime" -> missing formats
|
||||
# / HTTP 403). yt-dlp auto-detects deno on PATH.
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y --no-install-recommends ffmpeg \
|
||||
&& apt-get install -y --no-install-recommends ffmpeg ca-certificates curl unzip \
|
||||
&& arch="$(uname -m)" \
|
||||
&& case "$arch" in \
|
||||
x86_64) deno_arch=x86_64-unknown-linux-gnu ;; \
|
||||
aarch64) deno_arch=aarch64-unknown-linux-gnu ;; \
|
||||
*) echo "unsupported arch: $arch" >&2; exit 1 ;; \
|
||||
esac \
|
||||
&& curl -fsSL "https://github.com/denoland/deno/releases/latest/download/deno-${deno_arch}.zip" -o /tmp/deno.zip \
|
||||
&& unzip /tmp/deno.zip -d /usr/local/bin \
|
||||
&& rm /tmp/deno.zip \
|
||||
&& apt-get purge -y --auto-remove curl unzip \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
@@ -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,13 +13,14 @@ services:
|
||||
MUSICFETCH_API_KEY: "${MUSICFETCH_API_KEY}"
|
||||
MUSICFETCH_ROOT: "/media/music"
|
||||
MUSICFETCH_PORT: "6769"
|
||||
# Optional: authenticated YouTube cookies to avoid bot-check ("Sign in to
|
||||
# confirm you're not a bot") / rate limits. To enable, set in a .env file
|
||||
# (see server/.env.example):
|
||||
# YTDLP_COOKIES_HOST=/host/path/cookies.txt
|
||||
# The host file is mounted read-only at /cookies.txt and used automatically.
|
||||
YTDLP_COOKIES: "${YTDLP_COOKIES_HOST:+/cookies.txt}"
|
||||
volumes:
|
||||
- /media/music:/media/music
|
||||
networks:
|
||||
- lidarr_net
|
||||
|
||||
networks:
|
||||
lidarr_net:
|
||||
external: true
|
||||
# Set to the actual network name of your existing Lidarr stack, e.g.:
|
||||
# name: media_default
|
||||
# Cookies bind: maps to the host cookies.txt when YTDLP_COOKIES_HOST is set,
|
||||
# otherwise /dev/null (harmless no-op so the container still starts).
|
||||
- "${YTDLP_COOKIES_HOST:-/dev/null}:/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"]
|
||||
|
||||
@@ -4,3 +4,4 @@ requests
|
||||
ytmusicapi
|
||||
rich
|
||||
yt-dlp
|
||||
mutagen
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -81,3 +81,30 @@ def test_last_resort_universal_search(monkeypatch):
|
||||
monkeypatch.setattr(mf, "lidarr_get", fake_get)
|
||||
hits = mf.lidarr_search("Daft Punk - Discovery", 10)
|
||||
assert hits and hits[0].album == "Discovery"
|
||||
|
||||
|
||||
def test_unreachable_lidarr_warns_loudly(monkeypatch, capsys):
|
||||
# A connection error must surface on stderr (not silent dbg) so the
|
||||
# YouTube fallback isn't mistaken for "Lidarr had no match".
|
||||
monkeypatch.setattr(mf, "API_KEY", "testkey")
|
||||
monkeypatch.setattr(mf, "DEBUG", False)
|
||||
|
||||
def boom(path, params=None, timeout=15):
|
||||
raise mf.ReqConnectionError("Name or service not known")
|
||||
monkeypatch.setattr(mf, "lidarr_get", boom)
|
||||
|
||||
hits = mf._lidarr_album_candidates("anything")
|
||||
assert hits == []
|
||||
assert "Lidarr unreachable" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_http_error_stays_quiet(monkeypatch, capsys):
|
||||
monkeypatch.setattr(mf, "API_KEY", "testkey")
|
||||
monkeypatch.setattr(mf, "DEBUG", False)
|
||||
|
||||
def boom(path, params=None, timeout=15):
|
||||
raise mf.RequestException("500 Server Error")
|
||||
monkeypatch.setattr(mf, "lidarr_get", boom)
|
||||
|
||||
assert mf._lidarr_album_candidates("anything") == []
|
||||
assert "Lidarr unreachable" not in capsys.readouterr().err
|
||||
|
||||
@@ -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"
|
||||
@@ -102,3 +102,43 @@ def test_yt_download_returns_true_on_zero_exit(monkeypatch):
|
||||
|
||||
def test_yt_download_dry_run_returns_true():
|
||||
assert mf.yt_download("u", "/tmp/x", "best", True) is True
|
||||
|
||||
|
||||
def test_yt_download_always_sets_album_default(monkeypatch):
|
||||
captured = {}
|
||||
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
|
||||
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _CP(""))
|
||||
mf.yt_download("u", "/tmp/x", "best", False)
|
||||
assert "%(album|Unknown Album)s:%(meta_album)s" in captured["cmd"]
|
||||
|
||||
|
||||
def test_yt_download_single_word_tags_injected_literally(monkeypatch):
|
||||
# Regression: `--parse-metadata "Cochise:%(title)s"` makes yt-dlp treat the
|
||||
# bare word 'Cochise' as a FIELD name (field_to_template's r'[a-zA-Z_]+$'),
|
||||
# producing 'NA'. Single-word album/title must reach yt-dlp as literals.
|
||||
captured = {}
|
||||
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
|
||||
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _CP(""))
|
||||
hit = mf.Hit(source="youtube", kind="track", title="Cochise",
|
||||
artist="Audioslave", album="Solid", payload={"videoId": "x"})
|
||||
mf.yt_download("u", "/tmp/x", "best", False, hit=hit)
|
||||
cmd = captured["cmd"]
|
||||
joined = " ".join(cmd)
|
||||
# The buggy bare-word parse-metadata FROM must be gone.
|
||||
assert "Solid:%(album)s" not in joined
|
||||
assert "Cochise:%(title)s" not in joined
|
||||
# Literal values must be passed as literal args (immune to template parsing).
|
||||
assert "Solid" in cmd
|
||||
assert "Cochise" in cmd
|
||||
# A hit album must not be clobbered by the Unknown-Album default.
|
||||
assert "%(album|Unknown Album)s:%(meta_album)s" not in cmd
|
||||
|
||||
|
||||
def test_yt_download_passes_cookies(monkeypatch):
|
||||
captured = {}
|
||||
monkeypatch.setattr(mf, "COOKIES_FILE", "/cookies.txt")
|
||||
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
|
||||
monkeypatch.setattr(mf.os, "makedirs", lambda *a, **k: None)
|
||||
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _CP(""))
|
||||
mf.yt_download("u", "/tmp/x", "best", False)
|
||||
assert "--cookies" in captured["cmd"] and "/cookies.txt" in captured["cmd"]
|
||||
|
||||
@@ -63,7 +63,7 @@ class _FakeAudio(dict):
|
||||
|
||||
def test_repair_file_fixes_album_and_year(monkeypatch):
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url: {"artist": "Daft Punk", "title": "Aerodynamic",
|
||||
lambda url, *a: {"artist": "Daft Punk", "title": "Aerodynamic",
|
||||
"album": "Discovery", "release_year": 2001})
|
||||
audio = _FakeAudio({"artist": ["Daft Punk"], "title": ["Aerodynamic"]}) # album/date missing
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
@@ -76,7 +76,7 @@ def test_repair_file_fixes_album_and_year(monkeypatch):
|
||||
|
||||
def test_repair_file_dry_run_writes_nothing(monkeypatch):
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url: {"artist": "A", "title": "T", "album": "Alb", "release_year": 2020})
|
||||
lambda url, *a: {"artist": "A", "title": "T", "album": "Alb", "release_year": 2020})
|
||||
audio = _FakeAudio({})
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
changed = mf.repair_file(f"X/youtube/T [{YT_ID}].opus", "youtube", dry_run=True)
|
||||
@@ -88,7 +88,7 @@ def test_repair_file_dry_run_writes_nothing(monkeypatch):
|
||||
def test_repair_file_skips_music_video(monkeypatch):
|
||||
# No album AND no valid release year -> treat as a video, leave tags alone.
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url: {"title": "Artist - Song (Official Music Video)",
|
||||
lambda url, *a: {"title": "Artist - Song (Official Music Video)",
|
||||
"uploader": "SomeVEVO", "upload_date": "20110101"})
|
||||
audio = _FakeAudio({"artist": ["Real Artist"], "title": ["Song"]})
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
@@ -100,7 +100,7 @@ def test_repair_file_skips_music_video(monkeypatch):
|
||||
def test_repair_file_fills_missing_but_never_clobbers(monkeypatch):
|
||||
# Source artist is a channel name; existing artist must be kept.
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url: {"artist": "SomeChannelVEVO", "title": "Channel Decorated Title",
|
||||
lambda url, *a: {"artist": "SomeChannelVEVO", "title": "Channel Decorated Title",
|
||||
"album": "Real Album", "release_year": 2019})
|
||||
audio = _FakeAudio({"artist": ["Correct Artist"], "title": ["Clean Title"]})
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
@@ -112,7 +112,7 @@ def test_repair_file_fills_missing_but_never_clobbers(monkeypatch):
|
||||
|
||||
def test_repair_file_fills_missing_artist_when_absent(monkeypatch):
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url: {"artist": "Real Artist", "title": "T",
|
||||
lambda url, *a: {"artist": "Real Artist", "title": "T",
|
||||
"album": "Alb", "release_year": 2020})
|
||||
audio = _FakeAudio({}) # nothing present -> fill artist + title too
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
@@ -123,7 +123,7 @@ def test_repair_file_fills_missing_artist_when_absent(monkeypatch):
|
||||
def test_repair_file_skips_bad_id(monkeypatch):
|
||||
called = {"meta": False}
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url: called.update(meta=True) or {})
|
||||
lambda url, *a: called.update(meta=True) or {})
|
||||
# last bracket is a descriptor, not a real id
|
||||
assert mf.repair_file("X/youtube/Song [Official Video].opus", "youtube", dry_run=False) == []
|
||||
assert called["meta"] is False # never hit the network
|
||||
@@ -132,11 +132,33 @@ def test_repair_file_skips_bad_id(monkeypatch):
|
||||
def test_repair_file_skips_unparseable(monkeypatch):
|
||||
called = {"meta": False}
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url: called.update(meta=True) or {})
|
||||
lambda url, *a: called.update(meta=True) or {})
|
||||
assert mf.repair_file("X/youtube/no-id.opus", "youtube", dry_run=False) == []
|
||||
assert called["meta"] is False
|
||||
|
||||
|
||||
def test_run_yt_dlp_get_metadata_passes_extra_args(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
class _R:
|
||||
stdout = '{"title": "x"}'
|
||||
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _R())
|
||||
mf.run_yt_dlp_get_metadata("http://u", ["--extractor-args", "youtube:player_skip=js"])
|
||||
assert "youtube:player_skip=js" in captured["cmd"]
|
||||
|
||||
|
||||
def test_repair_uses_player_skip_fast_args(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
def fake_meta(url, extra_args=None):
|
||||
captured["extra"] = extra_args
|
||||
return {"album": "A", "release_year": 2020, "artist": "X", "title": "T"}
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", fake_meta)
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda p: (_FakeAudio({}), None))
|
||||
mf.repair_file(f"X/youtube/T [{YT_ID}].opus", "youtube", dry_run=True)
|
||||
assert captured["extra"] == mf._REPAIR_META_ARGS
|
||||
|
||||
|
||||
# ---- repair_library (real temp tree, repair_file mocked) ----
|
||||
def test_repair_library_scans_only_source_dirs(tmp_path, monkeypatch):
|
||||
root = tmp_path
|
||||
@@ -159,6 +181,23 @@ def test_repair_library_missing_root():
|
||||
assert mf.repair_library("/no/such/dir", dry_run=False) == (0, 0)
|
||||
|
||||
|
||||
def test_repair_library_exclude_skips_folders(tmp_path, monkeypatch):
|
||||
root = tmp_path
|
||||
(root / "Daft Punk" / "youtube").mkdir(parents=True)
|
||||
(root / "Daft Punk" / "youtube" / f"A [{YT_ID}].opus").write_text("x")
|
||||
(root / "Unsorted" / "youtube").mkdir(parents=True) # excluded artist folder
|
||||
(root / "Unsorted" / "youtube" / f"B [{YT_ID}].opus").write_text("x")
|
||||
(root / "Ephixa" / "playlists").mkdir(parents=True) # excluded source folder
|
||||
(root / "Ephixa" / "playlists" / f"C [{YT_ID}].opus").write_text("x")
|
||||
|
||||
visited = []
|
||||
monkeypatch.setattr(mf, "repair_file",
|
||||
lambda path, source, dry_run: visited.append(path) or ["x"])
|
||||
scanned, _ = mf.repair_library(str(root), dry_run=False, exclude=["unsorted", "playlists"])
|
||||
assert scanned == 1
|
||||
assert visited and "Daft Punk" in visited[0]
|
||||
|
||||
|
||||
# ---- offline retag-from-path ----
|
||||
def test_title_from_filename():
|
||||
assert mf._title_from_filename(f"Song [{YT_ID}].opus") == "Song"
|
||||
@@ -221,3 +260,174 @@ def test_retag_library_walks_source_files(tmp_path, monkeypatch):
|
||||
scanned, changed = mf.retag_library_from_path(str(root), dry_run=False)
|
||||
assert (scanned, changed) == (1, 1)
|
||||
assert visited == ["Daft Punk"]
|
||||
|
||||
|
||||
# ---- bogus-tag recovery (old-code NA / Unknown breakage) ----
|
||||
def test_is_bogus():
|
||||
for v in ("", "NA", "na", "N/A", "Unknown", "Unknown Album", "unknown artist", " NA "):
|
||||
assert mf._is_bogus(v) is True, v
|
||||
for v in ("Cochise", "Solid", "Brother Stoon", "Discovery"):
|
||||
assert mf._is_bogus(v) is False, v
|
||||
|
||||
|
||||
def test_repair_file_overwrites_bogus_title(monkeypatch):
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
|
||||
"album": "Audioslave", "release_year": 2002})
|
||||
audio = _FakeAudio({"artist": ["Audioslave"], "title": ["NA"]}) # bogus title
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
changed = mf.repair_file(f"X/youtube/Brother Stoon [{YT_ID}].opus", "youtube", dry_run=False)
|
||||
assert "title=Cochise" in changed
|
||||
assert audio["title"] == ["Cochise"]
|
||||
|
||||
|
||||
def test_repair_file_overwrites_bogus_artist(monkeypatch):
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url, *a: {"artist": "Real Artist", "title": "Real Title",
|
||||
"album": "Alb", "release_year": 2020})
|
||||
audio = _FakeAudio({"artist": ["NA"], "title": ["Good Title"]}) # bogus artist, good title
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
changed = mf.repair_file(f"X/youtube/Good Title [{YT_ID}].opus", "youtube", dry_run=False)
|
||||
assert "artist=Real Artist" in changed
|
||||
assert audio["artist"] == ["Real Artist"]
|
||||
assert audio["title"] == ["Good Title"] # good title untouched
|
||||
|
||||
|
||||
def test_repair_file_normalizes_na_album_when_source_has_none(monkeypatch):
|
||||
# Music video: no source album/year, but album tag is the literal 'NA' -> Unknown Album.
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url, *a: {"title": "Some Live Thing", "uploader": "Chan"})
|
||||
audio = _FakeAudio({"artist": ["X"], "title": ["Y"], "album": ["NA"]})
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
changed = mf.repair_file(f"X/youtube/Y [{YT_ID}].opus", "youtube", dry_run=False)
|
||||
assert "album=Unknown Album" in changed
|
||||
assert audio["album"] == ["Unknown Album"]
|
||||
|
||||
|
||||
def test_repair_file_renames_bogus_filename(tmp_path, monkeypatch):
|
||||
d = tmp_path / "Audioslave" / "youtube"
|
||||
d.mkdir(parents=True)
|
||||
f = d / f"NA [{YT_ID}].opus"
|
||||
f.write_text("x")
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
|
||||
"album": "Audioslave", "release_year": 2002})
|
||||
audio = _FakeAudio({"artist": ["Audioslave"], "title": ["NA"]})
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
|
||||
changed = mf.repair_file(str(f), "youtube", dry_run=False)
|
||||
assert (d / f"Cochise [{YT_ID}].opus").exists()
|
||||
assert not f.exists()
|
||||
assert any("rename" in c.lower() or c.startswith("title=") for c in changed)
|
||||
|
||||
|
||||
def test_repair_file_dry_run_does_not_rename(tmp_path, monkeypatch):
|
||||
d = tmp_path / "Audioslave" / "youtube"
|
||||
d.mkdir(parents=True)
|
||||
f = d / f"NA [{YT_ID}].opus"
|
||||
f.write_text("x")
|
||||
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
|
||||
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
|
||||
"album": "Audioslave", "release_year": 2002})
|
||||
monkeypatch.setattr(mf, "_open_audio", lambda path: (_FakeAudio({"title": ["NA"]}), None))
|
||||
mf.repair_file(str(f), "youtube", dry_run=True)
|
||||
assert f.exists() # untouched in dry-run
|
||||
assert not (d / f"Cochise [{YT_ID}].opus").exists()
|
||||
|
||||
|
||||
def test_fs_safe_replaces_slash():
|
||||
assert "/" not in mf._fs_safe("AC/DC Live")
|
||||
|
||||
|
||||
# ---- parallel repair ----
|
||||
def test_repair_library_parallel_visits_all(tmp_path, monkeypatch):
|
||||
root = tmp_path
|
||||
n = 50
|
||||
for i in range(n):
|
||||
d = root / f"Artist{i}" / "youtube"
|
||||
d.mkdir(parents=True)
|
||||
(d / f"T{i} [{YT_ID}].opus").write_text("x")
|
||||
|
||||
import threading
|
||||
seen = set()
|
||||
lock = threading.Lock()
|
||||
|
||||
def fake(path, source, dry_run):
|
||||
with lock:
|
||||
seen.add(path)
|
||||
return ["album=X"]
|
||||
monkeypatch.setattr(mf, "repair_file", fake)
|
||||
scanned, changed = mf.repair_library(str(root), dry_run=False, workers=8)
|
||||
assert scanned == n and changed == n
|
||||
assert len(seen) == n
|
||||
|
||||
|
||||
def test_repair_library_default_workers_still_works(tmp_path, monkeypatch):
|
||||
root = tmp_path
|
||||
(root / "A" / "youtube").mkdir(parents=True)
|
||||
(root / "A" / "youtube" / f"T [{YT_ID}].opus").write_text("x")
|
||||
monkeypatch.setattr(mf, "repair_file", lambda p, s, d: ["x"])
|
||||
assert mf.repair_library(str(root), dry_run=False) == (1, 1)
|
||||
|
||||
|
||||
# ---- cookies + error visibility ----
|
||||
def test_cookie_args_file_takes_precedence(monkeypatch):
|
||||
monkeypatch.setattr(mf, "COOKIES_FILE", "/c.txt")
|
||||
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "firefox")
|
||||
assert mf._cookie_args() == ["--cookies", "/c.txt"]
|
||||
|
||||
|
||||
def test_cookie_args_browser(monkeypatch):
|
||||
monkeypatch.setattr(mf, "COOKIES_FILE", "")
|
||||
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "firefox")
|
||||
assert mf._cookie_args() == ["--cookies-from-browser", "firefox"]
|
||||
|
||||
|
||||
def test_cookie_args_none(monkeypatch):
|
||||
monkeypatch.setattr(mf, "COOKIES_FILE", "")
|
||||
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
|
||||
assert mf._cookie_args() == []
|
||||
|
||||
|
||||
def test_metadata_fetch_passes_cookies(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
class _R:
|
||||
stdout = '{"title": "x"}'
|
||||
monkeypatch.setattr(mf, "COOKIES_FILE", "/cookies.txt")
|
||||
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
|
||||
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _R())
|
||||
mf.run_yt_dlp_get_metadata("http://u")
|
||||
assert "--cookies" in captured["cmd"]
|
||||
assert "/cookies.txt" in captured["cmd"]
|
||||
|
||||
|
||||
def test_ejs_args_default(monkeypatch):
|
||||
monkeypatch.setattr(mf, "REMOTE_COMPONENTS", "ejs:github")
|
||||
assert mf._ejs_args() == ["--remote-components", "ejs:github"]
|
||||
|
||||
|
||||
def test_ejs_args_disabled(monkeypatch):
|
||||
monkeypatch.setattr(mf, "REMOTE_COMPONENTS", "")
|
||||
assert mf._ejs_args() == []
|
||||
|
||||
|
||||
def test_metadata_fetch_passes_ejs(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
class _R:
|
||||
stdout = '{"title": "x"}'
|
||||
monkeypatch.setattr(mf, "REMOTE_COMPONENTS", "ejs:github")
|
||||
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _R())
|
||||
mf.run_yt_dlp_get_metadata("http://u")
|
||||
assert "--remote-components" in captured["cmd"]
|
||||
assert "ejs:github" 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