30 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
7951c436dd Merge feat/unknown-album: default blank album to 'Unknown Album' 2026-06-11 21:53:11 -07:00
8b881c14bf feat: always embed an album tag (default 'Unknown Album')
Downloads with no album (regular YouTube videos, fan edits, etc) left a blank
album, which trips up players like Plexamp. yt_download now appends
--parse-metadata "%(album|Unknown Album)s:%(meta_album)s" so the native/resolved
album is kept when present, else 'Unknown Album' is embedded. Applies to all
download paths (search, playlist, single URL).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-11 21:53:11 -07:00
530b5b0406 typo in mutagen. fixed 2026-06-11 21:51:16 -07:00
0a4e6d474a added mutagen to requirements.txt 2026-06-11 21:48:33 -07:00
c0503187c5 Merge feat/repair-fast-meta: faster --repair via player_skip=js 2026-06-10 22:52:39 -07:00
a6aa469084 perf(repair): skip YouTube JS signature step when fetching tags
--repair only reads metadata (never downloads), so pass
--extractor-args youtube:player_skip=js to yt-dlp. Keeps album/artist/year/title
but avoids the slow, throttle-prone nsig JS step (which crawls without a JS
runtime and trips YouTube rate-limiting during bulk runs). run_yt_dlp_get_metadata
gains an optional extra_args param; the download path is unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 22:52:39 -07:00
f071158c10 Merge feat/repair-exclude: -x/--exclude for repair/retag 2026-06-10 22:41:16 -07:00
c6bde6958a feat: -x/--exclude to skip folders during --repair/--retag-from-path
Repeatable -x/--exclude NAME skips any artist- or source-level folder whose name
matches (case-insensitive) when walking the library, so hand-curated folders like
/media/music/Unsorted or .../playlists are left untouched. Threaded through
_iter_source_files -> repair_library / retag_library_from_path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 22:41:16 -07:00
7ea3ad2538 Merge feat/retag-from-path: offline tag recovery 2026-06-10 22:30:25 -07:00
9af7f91a25 feat: --retag-from-path to recover tags damaged by a prior --repair
Offline re-tag of artist/title from the artist folder + filename: strips
(Official Video)/(Lyrics)-style decorations and trailing [id], and treats an
'Artist - Title' filename as authoritative (recovering the real artist for
music videos filed under a channel name). Overwrites artist/title only; leaves
album/year. Honors --dry-run.

Refactors the source-folder walk into _iter_source_files, shared by --repair
and --retag-from-path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 22:30:25 -07:00
17 changed files with 2099 additions and 105 deletions

4
.gitignore vendored
View File

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

107
README.md
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 - 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"`). `"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). - 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 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 release is available**, MusicFetch automatically falls through to the top
@@ -95,6 +99,12 @@ export LIDARR_API_KEY="your-lidarr-api-key"
| `-o`, `--root PATH` | Output root folder (default `/media/music`). | | `-o`, `--root PATH` | Output root folder (default `/media/music`). |
| `--search-all` | Search all albums when adding an artist to Lidarr. | | `--search-all` | Search all albums when adding an artist to Lidarr. |
| `--repair` | Re-tag existing downloads under `--root` from source metadata (see below). | | `--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. | | `--debug` | Verbose output. |
### Examples ### Examples
@@ -129,13 +139,88 @@ file using the `[id]` in its filename, and fixes tags. Useful when downloads lan
missing album or wrong year. missing album or wrong year.
It is deliberately **conservative**: it overwrites **album** and **year** (the usual It is deliberately **conservative**: it overwrites **album** and **year** (the usual
breakage), but only *fills in* a missing **artist**/**title** — it never overwrites existing breakage), and fills in **artist**/**title** when they are missing *or* a known-bogus
artist/title with channel names or decorated video titles. Files whose source returns no real placeholder (`NA`, `Unknown Album`, `Unknown Artist` — left behind by older buggy tagging) —
music metadata (no album and no release year — e.g. plain music videos) are left untouched. 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. `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 ```bash
# Preview what would change (writes nothing) # Preview what would change (writes nothing)
./musicfetch --repair -d ./musicfetch --repair -d
@@ -144,6 +229,20 @@ It re-queries the source over the network, so run it occasionally, not constantl
./musicfetch --repair -o /media/music ./musicfetch --repair -o /media/music
``` ```
**`--retag-from-path`** is an offline companion: it derives **artist** and **title** purely
from the folder name + filename (stripping `(Official Video)` / `(Lyrics)`-style decorations,
and treating an `Artist - Title` filename correctly), with no network. Use it to undo bad
tags — e.g. titles/artists clobbered by an earlier `--repair` on music videos. It overwrites
artist/title and leaves album/year alone.
```bash
./musicfetch --retag-from-path -d # preview
./musicfetch --retag-from-path -o /media/music
# Skip folders (e.g. hand-curated playlists you don't want re-tagged)
./musicfetch --repair -x Unsorted -x playlists
```
### 📁 Output Structure ### 📁 Output Structure
```text ```text

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 from typing import Optional
import requests 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 from urllib.parse import urlparse, parse_qs
# Optional deps — degrade gracefully if missing. # 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(). # Runtime flags, populated in main().
DEBUG = False 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 for --quality.
QUALITY_CHOICES = ["best", "320", "m4a", "opus", "flac"] QUALITY_CHOICES = ["best", "320", "m4a", "opus", "flac"]
@@ -78,6 +133,21 @@ class Hit:
return self.title or self.album or self.artist 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 # Helpers
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -236,11 +306,22 @@ def lidarr_search(query: str, limit: int) -> list[Hit]:
return _fallback_lookup(query, limit, artist_first=True) 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]: def _lidarr_album_candidates(term: str) -> list[Hit]:
try: try:
return [_album_to_hit(a) for a in lidarr_get("/api/v1/album/lookup", params={"term": term})] return [_album_to_hit(a) for a in lidarr_get("/api/v1/album/lookup", params={"term": term})]
except RequestException as e: except RequestException as e:
dbg(f"album/lookup failed: {e}") _log_lidarr_failure("album/lookup", e)
return [] return []
@@ -248,7 +329,7 @@ def _lidarr_artist_candidates(term: str) -> list[Hit]:
try: try:
return [_artist_to_hit(a) for a in lidarr_get("/api/v1/artist/lookup", params={"term": term})] return [_artist_to_hit(a) for a in lidarr_get("/api/v1/artist/lookup", params={"term": term})]
except RequestException as e: except RequestException as e:
dbg(f"artist/lookup failed: {e}") _log_lidarr_failure("artist/lookup", e)
return [] return []
@@ -325,7 +406,7 @@ def _ytmusic_search(query: str, limit: int) -> list[Hit]:
def _ytdlp_search(query: str, limit: int) -> list[Hit]: def _ytdlp_search(query: str, limit: int) -> list[Hit]:
try: try:
result = subprocess.run( 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, capture_output=True, text=True, check=True,
) )
data = json.loads(result.stdout) 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, def yt_download(url_or_query: str, target_folder: Optional[str], quality: str, dry_run: bool,
hit: Optional[Hit] = None, outtmpl: Optional[str] = None): hit: Optional[Hit] = None, outtmpl: Optional[str] = None):
cmd = ["yt-dlp", cmd = ["yt-dlp",
*_cookie_args(),
*_ejs_args(),
*_sleep_args(download=True),
*_quality_args(quality), *_quality_args(quality),
"--embed-metadata", "--embed-metadata",
"--embed-thumbnail", "--embed-thumbnail",
@@ -605,19 +689,34 @@ def yt_download(url_or_query: str, target_folder: Optional[str], quality: str, d
cmd += ["-o", outtmpl] cmd += ["-o", outtmpl]
else: else:
cmd += ["-P", target_folder] 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:
if hit.artist: primary_artist = hit.artist.split(",")[0].strip() if hit.artist else ""
# First artist only; anchored ^.*$ replaces the whole field exactly once if primary_artist:
# (a bare .* matches twice and doubles the value). cmd += _force_tag("artist", primary_artist)
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"]
if hit.title: 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: if hit.year:
cmd += ["--parse-metadata", f"{hit.year}:%(release_year)s"] 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) cmd.append(url_or_query)
dest = outtmpl or target_folder 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 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]]: def _ytmusic_playlist(pid: str) -> tuple[str, list[Hit]]:
"""Expand a YouTube Music playlist via ytmusicapi. Returns ("", []) on failure.""" """Expand a YouTube Music playlist via ytmusicapi. Returns ("", []) on failure."""
try: try:
@@ -718,6 +830,32 @@ def _entry_to_hit(entry: dict) -> Hit:
"extractor": source}) "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]]: def probe_url(url: str) -> tuple[str, str, list[Hit]]:
"""Classify a URL via yt-dlp. Returns (kind, title, hits) where kind is """Classify a URL via yt-dlp. Returns (kind, title, hits) where kind is
'playlist' (hits populated) or 'track' (hits empty; caller downloads the URL). '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: if hits:
return "playlist", title, hits return "playlist", title, hits
try: 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) capture_output=True, text=True, check=True)
data = json.loads(result.stdout) data = json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e: 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} 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: try:
result = subprocess.run( result = subprocess.run(cmd, capture_output=True, text=True, check=True)
["yt-dlp", "-j", "--no-playlist", url],
capture_output=True, text=True, check=True,
)
return json.loads(result.stdout) return json.loads(result.stdout)
except (subprocess.CalledProcessError, json.JSONDecodeError) as e: 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 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: def get_artist_from_metadata(meta: dict) -> str:
for key in ("artist", "creator", "uploader", "channel"): for key in ("artist", "creator", "uploader", "channel"):
if meta.get(key): if meta.get(key):
@@ -788,6 +936,46 @@ def get_artist_from_metadata(meta: dict) -> str:
return "Unknown Artist" 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): def handle_url(url: str, root: str, quality: str, dry_run: bool):
kind, title, hits = probe_url(url) kind, title, hits = probe_url(url)
if kind == "playlist": 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) 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]: def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
"""Re-tag one file from source metadata. Conservative: fixes album/year """Re-tag one file from source metadata. album/year are authoritative
(the common breakage) and only fills MISSING artist/title — never clobbers (overwrite); artist/title are filled when MISSING *or* a known-bogus
existing tags with channel names or decorated music-video titles. Files whose placeholder ('NA', 'Unknown …') — the old tagging bug wrote those — but a
source has no real music metadata (no album/year, e.g. music videos) are left genuine existing tag is never clobbered with a channel name or decorated
untouched. Returns the list of changed fields.""" 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)) parsed = _parse_track_file(os.path.basename(path))
if not parsed: if not parsed:
dbg(f"skip (no id): {path}") 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: if not url:
dbg(f"skip (source '{source}' not re-queryable): {path}") dbg(f"skip (source '{source}' not re-queryable): {path}")
return [] return []
meta = run_yt_dlp_get_metadata(url) meta = run_yt_dlp_get_metadata(url, _REPAIR_META_ARGS)
if not meta: if not meta:
dbg(f"skip (no metadata): {path}") dbg(f"skip (no metadata): {path}")
return [] 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: try:
audio, key_map = _open_audio(path) audio, key_map = _open_audio(path)
except Exception as e: # noqa: BLE001 except Exception as e: # noqa: BLE001
@@ -912,20 +1135,144 @@ def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
if audio is None: if audio is None:
return [] 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 = {} updates = {}
if album: if album:
updates["album"] = 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: if year:
updates["date"] = year updates["date"] = year
artist = get_artist_from_metadata(meta) # Fill artist/title when missing OR bogus; never overwrite a genuine value.
if artist and artist != "Unknown Artist" and not _read_tag(audio, key_map, "artist"): if meta_artist and not _is_bogus(meta_artist) and _is_bogus(cur_artist):
updates["artist"] = artist updates["artist"] = meta_artist
title = meta.get("title", "") if meta_title and not _is_bogus(meta_title) and _is_bogus(cur_title):
if title and not _read_tag(audio, key_map, "title"): updates["title"] = meta_title
updates["title"] = title
changed = [] changed = []
for field, value in updates.items():
if _read_tag(audio, key_map, field) != value:
changed.append(f"{field}={value}")
if not dry_run:
audio[key_map[field] if key_map else field] = [value]
if changed and not dry_run:
audio.save()
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=(), 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
def _one(path, source):
try:
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, 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).
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
for fname in sorted(os.listdir(sdir)):
if fname.lower().endswith(_AUDIO_EXTS):
yield os.path.join(sdir, fname), source, artist
# --- Offline retag-from-path (recover from tags damaged by a prior --repair) ---
_DECORATION_RE = re.compile(
r"\s*[\(\[][^)\]]*\b(?:official|lyric[s]?|audio|visuali[sz]er|"
r"music\s+video|m/?v|hd|hq|4k|explicit|remaster(?:ed)?)\b[^)\]]*[\)\]]",
re.IGNORECASE)
def _title_from_filename(filename: str) -> str:
"""Filename minus extension and a trailing ' [<id>]'."""
stem = re.sub(r"\.(?:" + "|".join(_AUDIO_EXTS) + r")$", "", filename, flags=re.IGNORECASE)
return re.sub(r"\s*\[[^\]]+\]$", "", stem).strip()
def _strip_decorations(title: str) -> str:
return re.sub(r"\s{2,}", " ", _DECORATION_RE.sub("", title)).strip(" -–—")
def _derive_from_filename(filename: str, folder_artist: str) -> tuple[str, str]:
"""Best-effort (artist, title) from the filename. A 'Artist - Title' name wins
over the folder (handles music-video downloads filed under a channel name)."""
title = _strip_decorations(_title_from_filename(filename))
if " - " in title:
left, right = title.split(" - ", 1)
return left.strip(), right.strip()
return folder_artist, title
def retag_file_from_path(path: str, folder_artist: str, dry_run: bool) -> list[str]:
"""Overwrite artist/title from the folder + cleaned filename. Leaves album/date."""
artist, title = _derive_from_filename(os.path.basename(path), folder_artist)
try:
audio, key_map = _open_audio(path)
except Exception as e: # noqa: BLE001
err(f"cannot open {path}: {e}")
return []
if audio is None:
return []
updates = {}
if artist:
updates["artist"] = artist
if title:
updates["title"] = title
changed = []
for field, value in updates.items(): for field, value in updates.items():
if _read_tag(audio, key_map, field) != value: if _read_tag(audio, key_map, field) != value:
changed.append(f"{field}={value}") changed.append(f"{field}={value}")
@@ -939,30 +1286,20 @@ def repair_file(path: str, source: str, dry_run: bool) -> list[str]:
return changed return changed
def repair_library(root: str, dry_run: bool) -> tuple[int, int]: def retag_library_from_path(root: str, dry_run: bool, exclude=()) -> tuple[int, int]:
"""Walk <root>/<artist>/<source>/ and re-tag audio files. Returns (scanned, changed).""" """Re-tag artist/title offline from folder+filename for every source file."""
if not os.path.isdir(root): if not os.path.isdir(root):
err(f"Root folder not found: {root}") err(f"Root folder not found: {root}")
return 0, 0 return 0, 0
scanned = changed = 0 scanned = changed = 0
for artist in sorted(os.listdir(root)): for path, _source, artist in _iter_source_files(root, exclude):
adir = os.path.join(root, artist) scanned += 1
if not os.path.isdir(adir): try:
continue if retag_file_from_path(path, artist, dry_run):
for source in sorted(os.listdir(adir)): changed += 1
sdir = os.path.join(adir, source) except Exception as e: # noqa: BLE001
if not os.path.isdir(sdir) or not _is_source_dir(source): err(f"retag failed ({os.path.basename(path)}): {e}")
continue verb = "Would retag" if dry_run else "Retagged"
for fname in sorted(os.listdir(sdir)):
if not fname.lower().endswith(_AUDIO_EXTS):
continue
scanned += 1
try:
if repair_file(os.path.join(sdir, fname), source, dry_run):
changed += 1
except Exception as e: # noqa: BLE001 — one bad file shouldn't abort
err(f"repair failed ({fname}): {e}")
verb = "Would repair" if dry_run else "Repaired"
print(f"{verb} {changed}/{scanned} files") print(f"{verb} {changed}/{scanned} files")
return scanned, changed return scanned, changed
@@ -1004,18 +1341,73 @@ def parse_args():
help="Search all albums when adding an artist to Lidarr.") help="Search all albums when adding an artist to Lidarr.")
p.add_argument("--repair", action="store_true", p.add_argument("--repair", action="store_true",
help="Re-tag existing downloads under --root from source metadata.") 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.") p.add_argument("--debug", action="store_true", help="Verbose output.")
return p.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(): def main():
global DEBUG global DEBUG, COOKIES_FILE, COOKIES_FROM_BROWSER, SLEEP_REQUESTS, REMOTE_COMPONENTS
args = parse_args() args = parse_args()
DEBUG = args.debug 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() query = " ".join(args.query).strip()
if args.retag_from_path:
retag_library_from_path(args.root, args.dry_run, args.exclude)
return
if args.repair: if args.repair:
repair_library(args.root, args.dry_run) repair_library(args.root, args.dry_run, args.exclude, args.workers)
return return
if not query: if not query:
@@ -1027,7 +1419,11 @@ def main():
sys.exit(1) sys.exit(1)
if is_url(query): 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 return
hits = build_combined_hits(query, args.limit, args.ytsearch, hits = build_combined_hits(query, args.limit, args.ytsearch,
@@ -1041,21 +1437,8 @@ def main():
print("Nothing selected.") print("Nothing selected.")
return return
if chosen.source == "lidarr": _dispatch_chosen(chosen, hits, args.root, args.quality, args.dry_run,
if chosen.kind == "album": args.lidarr_only, args.search_all)
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)
if __name__ == "__main__": 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 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 \ 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/* && rm -rf /var/lib/apt/lists/*
WORKDIR /app 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}'.") raise HTTPException(status_code=422, detail=f"Invalid quality '{quality}'.")
if mf.is_url(q): if mf.is_url(q):
kind, title, hits = mf.probe_url(q) if mf._is_direct_url(q):
syn = mf.Hit(source="youtube", kind=kind, title=title, artist="") kind, title, hits = mf.probe_url(q)
job = jobs.create_job(hit=syn, message=actions.url_started_message(kind, title)) 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) response = _job_public(job)
done_msg = actions.playlist_done_message if kind == "playlist" else actions.url_done_message
jobs.run_job( jobs.run_job(
job.id, job.id,
lambda: actions.perform_url_fetch(q, kind, title, hits, quality, ROOT), lambda: actions.perform_fetch(chosen, hits, quality, ROOT),
done_message=done_msg, done_message=actions.done_message(chosen),
fail_message="Download failed.", fail_message=actions.failed_message(chosen),
) )
return response return response

View File

@@ -13,13 +13,14 @@ services:
MUSICFETCH_API_KEY: "${MUSICFETCH_API_KEY}" MUSICFETCH_API_KEY: "${MUSICFETCH_API_KEY}"
MUSICFETCH_ROOT: "/media/music" MUSICFETCH_ROOT: "/media/music"
MUSICFETCH_PORT: "6769" 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: volumes:
- /media/music:/media/music - /media/music:/media/music
networks: # Cookies bind: maps to the host cookies.txt when YTDLP_COOKIES_HOST is set,
- lidarr_net # otherwise /dev/null (harmless no-op so the container still starts).
- "${YTDLP_COOKIES_HOST:-/dev/null}:/cookies.txt:ro"
networks:
lidarr_net:
external: true
# Set to the actual network name of your existing Lidarr stack, e.g.:
# name: media_default

View File

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

View File

@@ -4,3 +4,4 @@ requests
ytmusicapi ytmusicapi
rich rich
yt-dlp yt-dlp
mutagen

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) r = client.post("/fetch", params={"q": "Daft Punk - Discovery"}, headers=auth)
assert r.status_code == 200 assert r.status_code == 200
assert r.json()["status"] == "queued" 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) monkeypatch.setattr(mf, "lidarr_get", fake_get)
hits = mf.lidarr_search("Daft Punk - Discovery", 10) hits = mf.lidarr_search("Daft Punk - Discovery", 10)
assert hits and hits[0].album == "Discovery" 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.probe_url)
assert callable(smf.download_hits) assert callable(smf.download_hits)
assert callable(smf.download_single) 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

@@ -102,3 +102,43 @@ def test_yt_download_returns_true_on_zero_exit(monkeypatch):
def test_yt_download_dry_run_returns_true(): def test_yt_download_dry_run_returns_true():
assert mf.yt_download("u", "/tmp/x", "best", True) is 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"]

View File

@@ -63,7 +63,7 @@ class _FakeAudio(dict):
def test_repair_file_fixes_album_and_year(monkeypatch): def test_repair_file_fixes_album_and_year(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", 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}) "album": "Discovery", "release_year": 2001})
audio = _FakeAudio({"artist": ["Daft Punk"], "title": ["Aerodynamic"]}) # album/date missing audio = _FakeAudio({"artist": ["Daft Punk"], "title": ["Aerodynamic"]}) # album/date missing
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None)) 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): def test_repair_file_dry_run_writes_nothing(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", 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({}) audio = _FakeAudio({})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None)) monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/T [{YT_ID}].opus", "youtube", dry_run=True) 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): def test_repair_file_skips_music_video(monkeypatch):
# No album AND no valid release year -> treat as a video, leave tags alone. # No album AND no valid release year -> treat as a video, leave tags alone.
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", 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"}) "uploader": "SomeVEVO", "upload_date": "20110101"})
audio = _FakeAudio({"artist": ["Real Artist"], "title": ["Song"]}) audio = _FakeAudio({"artist": ["Real Artist"], "title": ["Song"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None)) 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): def test_repair_file_fills_missing_but_never_clobbers(monkeypatch):
# Source artist is a channel name; existing artist must be kept. # Source artist is a channel name; existing artist must be kept.
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", 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}) "album": "Real Album", "release_year": 2019})
audio = _FakeAudio({"artist": ["Correct Artist"], "title": ["Clean Title"]}) audio = _FakeAudio({"artist": ["Correct Artist"], "title": ["Clean Title"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None)) 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): def test_repair_file_fills_missing_artist_when_absent(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", 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}) "album": "Alb", "release_year": 2020})
audio = _FakeAudio({}) # nothing present -> fill artist + title too audio = _FakeAudio({}) # nothing present -> fill artist + title too
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None)) 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): def test_repair_file_skips_bad_id(monkeypatch):
called = {"meta": False} called = {"meta": False}
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", 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 # last bracket is a descriptor, not a real id
assert mf.repair_file("X/youtube/Song [Official Video].opus", "youtube", dry_run=False) == [] assert mf.repair_file("X/youtube/Song [Official Video].opus", "youtube", dry_run=False) == []
assert called["meta"] is False # never hit the network 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): def test_repair_file_skips_unparseable(monkeypatch):
called = {"meta": False} called = {"meta": False}
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata", 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 mf.repair_file("X/youtube/no-id.opus", "youtube", dry_run=False) == []
assert called["meta"] is 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) ---- # ---- repair_library (real temp tree, repair_file mocked) ----
def test_repair_library_scans_only_source_dirs(tmp_path, monkeypatch): def test_repair_library_scans_only_source_dirs(tmp_path, monkeypatch):
root = tmp_path root = tmp_path
@@ -157,3 +179,255 @@ def test_repair_library_scans_only_source_dirs(tmp_path, monkeypatch):
def test_repair_library_missing_root(): def test_repair_library_missing_root():
assert mf.repair_library("/no/such/dir", dry_run=False) == (0, 0) assert mf.repair_library("/no/such/dir", dry_run=False) == (0, 0)
def test_repair_library_exclude_skips_folders(tmp_path, monkeypatch):
root = tmp_path
(root / "Daft Punk" / "youtube").mkdir(parents=True)
(root / "Daft Punk" / "youtube" / f"A [{YT_ID}].opus").write_text("x")
(root / "Unsorted" / "youtube").mkdir(parents=True) # excluded artist folder
(root / "Unsorted" / "youtube" / f"B [{YT_ID}].opus").write_text("x")
(root / "Ephixa" / "playlists").mkdir(parents=True) # excluded source folder
(root / "Ephixa" / "playlists" / f"C [{YT_ID}].opus").write_text("x")
visited = []
monkeypatch.setattr(mf, "repair_file",
lambda path, source, dry_run: visited.append(path) or ["x"])
scanned, _ = mf.repair_library(str(root), dry_run=False, exclude=["unsorted", "playlists"])
assert scanned == 1
assert visited and "Daft Punk" in visited[0]
# ---- offline retag-from-path ----
def test_title_from_filename():
assert mf._title_from_filename(f"Song [{YT_ID}].opus") == "Song"
assert mf._title_from_filename("STARDUST (Official Music Video) [3nsYNXtALhA].opus") \
== "STARDUST (Official Music Video)"
assert mf._title_from_filename("no brackets.mp3") == "no brackets"
def test_strip_decorations():
assert mf._strip_decorations("STARDUST (Official Music Video)") == "STARDUST"
assert mf._strip_decorations("Away From You (Lyrics)") == "Away From You"
assert mf._strip_decorations("More Than a Feeling (Official HD Video)") == "More Than a Feeling"
# real info like a feature credit is kept
assert mf._strip_decorations("WHO GON' SLIDE (Feat. Shakewell) [Official Music Video]") \
== "WHO GON' SLIDE (Feat. Shakewell)"
def test_derive_from_filename():
# plain title -> folder is the artist
assert mf._derive_from_filename(f"Aerodynamic [{YT_ID}].opus", "Daft Punk") == ("Daft Punk", "Aerodynamic")
# decorated music video filed under the artist
assert mf._derive_from_filename("STARDUST (Official Music Video) [3nsYNXtALhA].opus", "1nonly") \
== ("1nonly", "STARDUST")
# 'Artist - Title' name wins over a channel folder
assert mf._derive_from_filename("BLCKLGHT - Away From You (Lyrics) [QapF4b1jYw8].opus", "7clouds Techno") \
== ("BLCKLGHT", "Away From You")
def test_retag_file_from_path_fixes_clobbered_tags(monkeypatch):
audio = _FakeAudio({"artist": ["7clouds Techno"], "title": ["BLCKLGHT - Away From You (Lyrics)"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.retag_file_from_path(
"X/7clouds Techno/youtube/BLCKLGHT - Away From You (Lyrics) [QapF4b1jYw8].opus",
"7clouds Techno", dry_run=False)
assert set(changed) == {"artist=BLCKLGHT", "title=Away From You"}
assert audio["artist"] == ["BLCKLGHT"]
assert audio["title"] == ["Away From You"]
assert audio.saved is True
def test_retag_file_from_path_dry_run(monkeypatch):
audio = _FakeAudio({"artist": ["wrong"], "title": ["wrong"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.retag_file_from_path(f"X/Daft Punk/youtube/Aerodynamic [{YT_ID}].opus",
"Daft Punk", dry_run=True)
assert changed
assert audio == {"artist": ["wrong"], "title": ["wrong"]}
assert audio.saved is False
def test_retag_library_walks_source_files(tmp_path, monkeypatch):
root = tmp_path
(root / "Daft Punk" / "youtube").mkdir(parents=True)
(root / "Daft Punk" / "youtube" / f"Aerodynamic [{YT_ID}].opus").write_text("x")
(root / "Daft Punk" / "Discovery").mkdir(parents=True) # album folder -> skip
(root / "Daft Punk" / "Discovery" / "x.flac").write_text("x")
visited = []
monkeypatch.setattr(mf, "retag_file_from_path",
lambda path, artist, dry_run: visited.append(artist) or ["artist=x"])
scanned, changed = mf.retag_library_from_path(str(root), dry_run=False)
assert (scanned, changed) == (1, 1)
assert visited == ["Daft Punk"]
# ---- bogus-tag recovery (old-code NA / Unknown breakage) ----
def test_is_bogus():
for v in ("", "NA", "na", "N/A", "Unknown", "Unknown Album", "unknown artist", " NA "):
assert mf._is_bogus(v) is True, v
for v in ("Cochise", "Solid", "Brother Stoon", "Discovery"):
assert mf._is_bogus(v) is False, v
def test_repair_file_overwrites_bogus_title(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
"album": "Audioslave", "release_year": 2002})
audio = _FakeAudio({"artist": ["Audioslave"], "title": ["NA"]}) # bogus title
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Brother Stoon [{YT_ID}].opus", "youtube", dry_run=False)
assert "title=Cochise" in changed
assert audio["title"] == ["Cochise"]
def test_repair_file_overwrites_bogus_artist(monkeypatch):
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Real Artist", "title": "Real Title",
"album": "Alb", "release_year": 2020})
audio = _FakeAudio({"artist": ["NA"], "title": ["Good Title"]}) # bogus artist, good title
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Good Title [{YT_ID}].opus", "youtube", dry_run=False)
assert "artist=Real Artist" in changed
assert audio["artist"] == ["Real Artist"]
assert audio["title"] == ["Good Title"] # good title untouched
def test_repair_file_normalizes_na_album_when_source_has_none(monkeypatch):
# Music video: no source album/year, but album tag is the literal 'NA' -> Unknown Album.
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"title": "Some Live Thing", "uploader": "Chan"})
audio = _FakeAudio({"artist": ["X"], "title": ["Y"], "album": ["NA"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(f"X/youtube/Y [{YT_ID}].opus", "youtube", dry_run=False)
assert "album=Unknown Album" in changed
assert audio["album"] == ["Unknown Album"]
def test_repair_file_renames_bogus_filename(tmp_path, monkeypatch):
d = tmp_path / "Audioslave" / "youtube"
d.mkdir(parents=True)
f = d / f"NA [{YT_ID}].opus"
f.write_text("x")
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
"album": "Audioslave", "release_year": 2002})
audio = _FakeAudio({"artist": ["Audioslave"], "title": ["NA"]})
monkeypatch.setattr(mf, "_open_audio", lambda path: (audio, None))
changed = mf.repair_file(str(f), "youtube", dry_run=False)
assert (d / f"Cochise [{YT_ID}].opus").exists()
assert not f.exists()
assert any("rename" in c.lower() or c.startswith("title=") for c in changed)
def test_repair_file_dry_run_does_not_rename(tmp_path, monkeypatch):
d = tmp_path / "Audioslave" / "youtube"
d.mkdir(parents=True)
f = d / f"NA [{YT_ID}].opus"
f.write_text("x")
monkeypatch.setattr(mf, "run_yt_dlp_get_metadata",
lambda url, *a: {"artist": "Audioslave", "title": "Cochise",
"album": "Audioslave", "release_year": 2002})
monkeypatch.setattr(mf, "_open_audio", lambda path: (_FakeAudio({"title": ["NA"]}), None))
mf.repair_file(str(f), "youtube", dry_run=True)
assert f.exists() # untouched in dry-run
assert not (d / f"Cochise [{YT_ID}].opus").exists()
def test_fs_safe_replaces_slash():
assert "/" not in mf._fs_safe("AC/DC Live")
# ---- parallel repair ----
def test_repair_library_parallel_visits_all(tmp_path, monkeypatch):
root = tmp_path
n = 50
for i in range(n):
d = root / f"Artist{i}" / "youtube"
d.mkdir(parents=True)
(d / f"T{i} [{YT_ID}].opus").write_text("x")
import threading
seen = set()
lock = threading.Lock()
def fake(path, source, dry_run):
with lock:
seen.add(path)
return ["album=X"]
monkeypatch.setattr(mf, "repair_file", fake)
scanned, changed = mf.repair_library(str(root), dry_run=False, workers=8)
assert scanned == n and changed == n
assert len(seen) == n
def test_repair_library_default_workers_still_works(tmp_path, monkeypatch):
root = tmp_path
(root / "A" / "youtube").mkdir(parents=True)
(root / "A" / "youtube" / f"T [{YT_ID}].opus").write_text("x")
monkeypatch.setattr(mf, "repair_file", lambda p, s, d: ["x"])
assert mf.repair_library(str(root), dry_run=False) == (1, 1)
# ---- cookies + error visibility ----
def test_cookie_args_file_takes_precedence(monkeypatch):
monkeypatch.setattr(mf, "COOKIES_FILE", "/c.txt")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "firefox")
assert mf._cookie_args() == ["--cookies", "/c.txt"]
def test_cookie_args_browser(monkeypatch):
monkeypatch.setattr(mf, "COOKIES_FILE", "")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "firefox")
assert mf._cookie_args() == ["--cookies-from-browser", "firefox"]
def test_cookie_args_none(monkeypatch):
monkeypatch.setattr(mf, "COOKIES_FILE", "")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
assert mf._cookie_args() == []
def test_metadata_fetch_passes_cookies(monkeypatch):
captured = {}
class _R:
stdout = '{"title": "x"}'
monkeypatch.setattr(mf, "COOKIES_FILE", "/cookies.txt")
monkeypatch.setattr(mf, "COOKIES_FROM_BROWSER", "")
monkeypatch.setattr(mf.subprocess, "run", lambda cmd, **k: captured.update(cmd=cmd) or _R())
mf.run_yt_dlp_get_metadata("http://u")
assert "--cookies" in captured["cmd"]
assert "/cookies.txt" in captured["cmd"]
def test_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