20 Commits

Author SHA1 Message Date
fc2d082a25 feat(server): supply YouTube cookies to the API container via .env
Set YTDLP_COOKIES_HOST in server/.env and compose mounts the host
cookies.txt read-only at /cookies.txt and wires YTDLP_COOKIES
automatically — no compose edits per host. Falls back to /dev/null when
unset so the container still starts. Add server/.env.example, gitignore
server/.env, and document the flow.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-05 19:33:36 -07:00
82bc524155 fix: enable yt-dlp EJS solver so YouTube formats resolve
YouTube gates media formats behind signature / "n" JS challenges that
yt-dlp's bundled solver can't reliably crack. Without the EJS
remote-component solver script yt-dlp warns "Signature/n challenge
solving failed: Some formats may be missing" and downloads break — even
with valid cookies, so it masquerades as a cookie/auth problem.

Pass --remote-components ejs:github by default (env/CLI overridable) on
all four yt-dlp call sites. Needs a recent yt-dlp + deno present; set
YTDLP_REMOTE_COMPONENTS / --remote-components '' to disable.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-05 19:33:28 -07:00
b26e321926 feat: throttle yt-dlp requests to dodge YouTube rate-limiting
Bulk metadata fetches trip YouTube's per-session rate limit ("This
content isn't available, try again later"), failing even single-worker
runs after a burst. Add --sleep-requests between extraction calls (and a
randomized --sleep-interval before downloads), default 1s, tunable via
--sleep / $YTDLP_SLEEP (0 disables). Applied to metadata, search, probe,
and download yt-dlp invocations.

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

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

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

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

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

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

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

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

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

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 11:09:27 -07:00
33ca743a34 removed useless portion of compose 2026-06-11 22:02:51 -07:00
16 changed files with 1847 additions and 83 deletions

4
.gitignore vendored
View File

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

View File

@@ -8,6 +8,10 @@ whatever you choose. It accepts:
- A **free-form query**: an artist, an album, a track title, or combos like
`"Artist - Title"` or `"Artist - Album"` (e.g. `"ODESZA - Bloom"`, `"Daft Punk"`, `"Discovery"`).
- A **URL** (e.g. `"https://music.youtube.com/watch?v=..."` or a regular YouTube URL).
- **Any streaming link** (Spotify, Apple Music, Tidal, Deezer, …): resolved to
metadata via [Odesli/song.link](https://odesli.co), then searched on Lidarr
first and downloaded from the matching YouTube track if no Lidarr release is
available. YouTube and SoundCloud links download directly.
Lidarr is tried first by default. If you pick a Lidarr album but **no indexer
release is available**, MusicFetch automatically falls through to the top
@@ -95,6 +99,10 @@ 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. |
@@ -131,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

View File

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

View File

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

View File

@@ -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,22 +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"]
# Always embed an album: the resolved/native album if present, else a
# placeholder so players (e.g. Plexamp) don't choke on a blank album.
cmd += ["--parse-metadata", "%(album|Unknown Album)s:%(meta_album)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
@@ -690,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:
@@ -721,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).
@@ -732,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:
@@ -771,12 +906,19 @@ def download_single(url: str, root: str, quality: str, dry_run: bool) -> dict:
def run_yt_dlp_get_metadata(url: str, extra_args=None) -> Optional[dict]:
cmd = ["yt-dlp", "-j", "--no-playlist", *(extra_args or []), url]
cmd = ["yt-dlp", *_cookie_args(), *_ejs_args(), *_sleep_args(), "-j", "--no-playlist", *(extra_args or []), url]
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
return json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
err(f"yt-dlp metadata extraction failed: {e}")
# Surface yt-dlp's own last stderr line (e.g. 429 / "not a bot") instead
# of a bare exit code — the actual reason is what you need to act on.
detail = ""
stderr = getattr(e, "stderr", "") or ""
lines = [ln for ln in stderr.strip().splitlines() if ln.strip()]
if lines:
detail = f" — {lines[-1]}"
err(f"yt-dlp metadata extraction failed for {url}{detail}")
return None
@@ -794,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":
@@ -881,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}")
@@ -904,12 +1127,6 @@ def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
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
@@ -918,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():
@@ -942,22 +1169,40 @@ 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, exclude=()) -> tuple[int, int]:
"""Walk <root>/<artist>/<source>/ and re-tag audio files. Returns (scanned, changed)."""
def repair_library(root: str, dry_run: bool, exclude=(), workers: int = 8) -> tuple[int, int]:
"""Walk <root>/<artist>/<source>/ and re-tag audio files. Returns (scanned, changed).
Each file is an independent yt-dlp network round-trip, so they run in a
thread pool (network-bound); `workers` caps concurrency. Each thread owns
its own file + request, so no shared state needs locking beyond the counts.
Lower `workers` if YouTube starts rate-limiting (HTTP 429/403)."""
if not os.path.isdir(root):
err(f"Root folder not found: {root}")
return 0, 0
scanned = changed = 0
for path, source, _artist in _iter_source_files(root, exclude):
scanned += 1
def _one(path, source):
try:
if repair_file(path, source, dry_run):
changed += 1
return bool(repair_file(path, source, dry_run))
except Exception as e: # noqa: BLE001 — one bad file shouldn't abort
err(f"repair failed ({os.path.basename(path)}): {e}")
return False
scanned = changed = 0
files = ((p, s) for p, s, _a in _iter_source_files(root, exclude))
with ThreadPoolExecutor(max_workers=max(1, workers)) as ex:
for ok in ex.map(lambda ps: _one(*ps), files):
scanned += 1
changed += int(ok)
if scanned % 100 == 0:
print(f"… {scanned} scanned, {changed} changed", flush=True)
verb = "Would repair" if dry_run else "Repaired"
print(f"{verb} {changed}/{scanned} files")
return scanned, changed
@@ -1096,6 +1341,23 @@ 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).")
@@ -1106,10 +1368,38 @@ def parse_args():
return p.parse_args()
def _dispatch_chosen(chosen: Hit, hits: list[Hit], root: str, quality: str,
dry_run: bool, lidarr_only: bool, search_all: bool) -> None:
"""Act on a picked Hit: Lidarr album (add+search, fall to top YouTube hit on
no release), Lidarr artist, or a YouTube track. Shared by main() and handle_link."""
if chosen.source == "lidarr":
if chosen.kind == "album":
handled = act_lidarr_album(chosen, root, search_all, dry_run)
if not handled and not lidarr_only:
yt_fallback = next((h for h in hits if h.source == "youtube"), None)
if yt_fallback:
print("Using top YouTube hit as fallback.")
act_youtube(yt_fallback, root, quality, dry_run)
else:
print("No YouTube fallback available.")
else:
act_lidarr_artist(chosen, root, search_all, dry_run)
else:
act_youtube(chosen, root, quality, dry_run)
def main():
global DEBUG
global DEBUG, COOKIES_FILE, COOKIES_FROM_BROWSER, SLEEP_REQUESTS, 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:
@@ -1117,7 +1407,7 @@ def main():
return
if args.repair:
repair_library(args.root, args.dry_run, args.exclude)
repair_library(args.root, args.dry_run, args.exclude, args.workers)
return
if not query:
@@ -1129,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,
@@ -1143,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
View 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

View File

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

View File

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

View File

@@ -13,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"

View File

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

View File

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

View File

@@ -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

View File

@@ -6,3 +6,11 @@ def test_url_helpers_reexported():
assert callable(smf.probe_url)
assert callable(smf.download_hits)
assert callable(smf.download_single)
def test_odesli_symbols_reexported():
import server.mf as smf
assert callable(smf._is_direct_url)
assert callable(smf.odesli_resolve)
assert callable(smf.resolve_link_hits)
assert callable(smf.handle_link)

169
tests/test_odesli.py Normal file
View File

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

View File

@@ -110,3 +110,35 @@ def test_yt_download_always_sets_album_default(monkeypatch):
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"]

View File

@@ -260,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