# Music Search (Spotify + YouTube) 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:** 2 nuove tab in MusicTools ("🟢 Spotify" e "▶ YouTube") per cercare brani per titolo/artista e scaricarli. Spotify ha toggle "Solo artista" per discografia completa. YouTube è per mix DJ/unreleased/bootleg che Spotify non ha. **Architecture:** Estensione `core/spotify_client.py` + nuovo `core/youtube_search.py` + 6 metodi in `api/bridge.py` + 2 tab UI in `webui/`. Riuso completo del pipeline download esistente (`start_tracks_download` per Spotify, nuovo `start_urls_download` per YouTube). Zero cambi server. **Tech Stack:** Python 3.8+ (con `from __future__ import annotations`), `requests` (già in deps), `yt-dlp` via subprocess (già in bundle). `pytest` + `responses` + `unittest.mock` per test. Vanilla JS + HTML/CSS lato UI. **Spec:** `docs/superpowers/specs/2026-07-14-music-search-design.md` --- ## Note operative - **Branch:** `feat/music-search` (già creato, HEAD: `91aa1bd`) - **Git author:** ogni commit deve usare `git -c user.email=info@djluza.com commit ...` - **Test:** riusare pattern e infra dei test Beatport (`tests/`, `pytest`) - **Python 3.8 compat:** ogni nuovo file `.py` inizia con `from __future__ import annotations` - **Non toccare:** `server/`, `landing/`, `build_macos.py`, `build_windows.py` --- ## Task 1: `search_tracks(token, query, limit)` — TDD **Files:** - Modify: `core/spotify_client.py` - Modify: `tests/test_spotify_client.py` - [ ] **Step 1: Aggiungi test in `tests/test_spotify_client.py`** Aggiungi in fondo al file: ```python class TestSearchTracks: @responses.activate def test_returns_list_of_tracks(self): responses.add( responses.GET, "https://api.spotify.com/v1/search", json={ "tracks": { "items": [ { "id": f"id{i}", "name": f"Track {i}", "artists": [{"name": "Solomun"}], "album": {"name": "Album X"}, "duration_ms": 300000 + i * 1000, "external_urls": {"spotify": f"https://open.spotify.com/track/id{i}"}, } for i in range(50) ] } }, status=200, ) result = spotify_client.search_tracks("t", "solomun", limit=50) assert isinstance(result, list) assert len(result) == 50 first = result[0] assert first["id"] == "id0" assert first["name"] == "Track 0" assert first["artists"] == "Solomun" assert first["album"] == "Album X" assert first["duration_sec"] == 300 assert first["url"] == "https://open.spotify.com/track/id0" @responses.activate def test_multiple_artists_joined_with_comma(self): responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"tracks": {"items": [{ "id": "x", "name": "n", "artists": [{"name": "A"}, {"name": "B"}, {"name": "C"}], "album": {"name": "Alb"}, "duration_ms": 60000, "external_urls": {"spotify": "u"}, }]}}, status=200, ) result = spotify_client.search_tracks("t", "q", limit=1) assert result[0]["artists"] == "A, B, C" @responses.activate def test_empty_query_returns_empty_list(self): responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"tracks": {"items": []}}, status=200, ) result = spotify_client.search_tracks("t", "no-match", limit=50) assert result == [] @responses.activate def test_query_params_include_limit(self): responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"tracks": {"items": []}}, status=200, ) spotify_client.search_tracks("t", "q", limit=25) params = responses.calls[0].request.params assert params["q"] == "q" assert params["type"] == "track" assert params["limit"] == "25" ``` - [ ] **Step 2: Verifica fail** Run: `python3 -m pytest tests/test_spotify_client.py::TestSearchTracks -v` Expected: FAIL con `AttributeError: module 'core.spotify_client' has no attribute 'search_tracks'` - [ ] **Step 3: Aggiungi `search_tracks` a `core/spotify_client.py`** Aggiungi in fondo al file (dopo `search_track`): ```python def search_tracks(token: str, query: str, limit: int = 50) -> list: """Cerca brani su Spotify e ritorna una lista di dict. Args: token: access token Spotify query: query libera (titolo, artista, misto) limit: max risultati (Spotify cap = 50) Returns: list[dict] con {id, url, name, artists, album, duration_sec}. Vuota se nessun match. """ resp = requests.get( "https://api.spotify.com/v1/search", headers={"Authorization": f"Bearer {token}"}, params={"q": query, "type": "track", "limit": min(limit, 50)}, timeout=15, ) resp.raise_for_status() items = resp.json().get("tracks", {}).get("items", []) return [_track_to_dict(t) for t in items] def _track_to_dict(t: dict) -> dict: """Mappa il track object Spotify sul nostro schema uniforme.""" return { "id": t.get("id", ""), "url": t.get("external_urls", {}).get("spotify", ""), "name": t.get("name", ""), "artists": ", ".join(a.get("name", "") for a in t.get("artists", [])), "album": t.get("album", {}).get("name", ""), "duration_sec": int(t.get("duration_ms", 0)) // 1000, } ``` - [ ] **Step 4: Verifica** Run: `python3 -m pytest tests/test_spotify_client.py -v` Expected: 8 test PASS (4 existing + 4 new) - [ ] **Step 5: Commit** ```bash git add core/spotify_client.py tests/test_spotify_client.py git -c user.email=info@djluza.com commit -m "spotify: search_tracks() free-form con limit configurabile" ``` --- ## Task 2: `search_artist_discography(token, artist_name)` — TDD **Files:** - Modify: `core/spotify_client.py` - Modify: `tests/test_spotify_client.py` - [ ] **Step 1: Test in `tests/test_spotify_client.py`** Aggiungi in fondo: ```python class TestSearchArtistDiscography: @responses.activate def test_exact_name_match_beats_popular(self): # 3 candidati: uno esatto (case-insensitive), altri popolari responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"artists": {"items": [ {"id": "pop", "name": "Solomun Tribute", "popularity": 90}, {"id": "exact", "name": "SOLOMUN", "popularity": 60}, {"id": "unrelated", "name": "Other", "popularity": 70}, ]}}, status=200, ) # Mock top-tracks vuoto per non allungare il test responses.add( responses.GET, "https://api.spotify.com/v1/artists/exact/top-tracks", json={"tracks": []}, status=200, ) # Mock albums vuoto responses.add( responses.GET, "https://api.spotify.com/v1/artists/exact/albums", json={"items": []}, status=200, ) result = spotify_client.search_artist_discography("t", "Solomun") assert result == [] # Verifica che sia stato chiamato l'artista "exact", non "pop" top_tracks_calls = [c for c in responses.calls if "/top-tracks" in c.request.url] assert len(top_tracks_calls) == 1 assert "/exact/top-tracks" in top_tracks_calls[0].request.url @responses.activate def test_falls_back_to_most_popular_if_no_exact_match(self): responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"artists": {"items": [ {"id": "a1", "name": "Solomun Fanpage", "popularity": 30}, {"id": "a2", "name": "Solomun Live", "popularity": 80}, ]}}, status=200, ) responses.add( responses.GET, "https://api.spotify.com/v1/artists/a2/top-tracks", json={"tracks": []}, status=200, ) responses.add( responses.GET, "https://api.spotify.com/v1/artists/a2/albums", json={"items": []}, status=200, ) spotify_client.search_artist_discography("t", "solomun") top_tracks_calls = [c for c in responses.calls if "/top-tracks" in c.request.url] assert "/a2/top-tracks" in top_tracks_calls[0].request.url @responses.activate def test_raises_when_no_artist_found(self): responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"artists": {"items": []}}, status=200, ) with pytest.raises(ValueError, match="Artista"): spotify_client.search_artist_discography("t", "asdgjhkasdgj") @responses.activate def test_combines_top_tracks_and_album_tracks(self): # 1 artista esatto responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"artists": {"items": [{"id": "artX", "name": "artX", "popularity": 50}]}}, status=200, ) # 3 top-tracks top_tracks_data = {"tracks": [ { "id": f"t{i}", "name": f"Top{i}", "artists": [{"name": "artX"}], "album": {"name": "AlbTop"}, "duration_ms": 200000, "external_urls": {"spotify": f"u{i}"}, } for i in range(3) ]} responses.add( responses.GET, "https://api.spotify.com/v1/artists/artX/top-tracks", json=top_tracks_data, status=200, ) # 2 album responses.add( responses.GET, "https://api.spotify.com/v1/artists/artX/albums", json={"items": [ {"id": "alb1", "name": "Album 1"}, {"id": "alb2", "name": "Album 2"}, ]}, status=200, ) # album 1: 2 tracce responses.add( responses.GET, "https://api.spotify.com/v1/albums/alb1/tracks", json={"items": [ { "id": f"a1t{i}", "name": f"Alb1Track{i}", "artists": [{"name": "artX"}], "duration_ms": 180000, "external_urls": {"spotify": f"a1u{i}"}, } for i in range(2) ]}, status=200, ) # album 2: 1 traccia responses.add( responses.GET, "https://api.spotify.com/v1/albums/alb2/tracks", json={"items": [ { "id": "a2t0", "name": "Alb2Track0", "artists": [{"name": "artX"}], "duration_ms": 240000, "external_urls": {"spotify": "a2u0"}, } ]}, status=200, ) with patch("core.spotify_client.time.sleep"): # no wait result = spotify_client.search_artist_discography("t", "artX") # 3 top + 2 alb1 + 1 alb2 = 6 assert len(result) == 6 titles = {r["name"] for r in result} assert "Top0" in titles assert "Alb1Track0" in titles assert "Alb2Track0" in titles @responses.activate def test_dedupe_across_top_and_album(self): responses.add( responses.GET, "https://api.spotify.com/v1/search", json={"artists": {"items": [{"id": "aX", "name": "aX", "popularity": 50}]}}, status=200, ) # Stesso track name+artist in top-tracks e in album (id diverso) common = { "name": "Same Song", "artists": [{"name": "aX"}], "album": {"name": "OG Album"}, "duration_ms": 200000, "external_urls": {"spotify": "u"}, } responses.add( responses.GET, "https://api.spotify.com/v1/artists/aX/top-tracks", json={"tracks": [{**common, "id": "top-id"}]}, status=200, ) responses.add( responses.GET, "https://api.spotify.com/v1/artists/aX/albums", json={"items": [{"id": "alb", "name": "OG"}]}, status=200, ) responses.add( responses.GET, "https://api.spotify.com/v1/albums/alb/tracks", json={"items": [{**common, "id": "alb-id"}]}, status=200, ) with patch("core.spotify_client.time.sleep"): result = spotify_client.search_artist_discography("t", "aX") assert len(result) == 1 ``` Aggiungi import in cima al file test (se non presente): ```python from unittest.mock import patch ``` - [ ] **Step 2: Verifica fail** Run: `python3 -m pytest tests/test_spotify_client.py::TestSearchArtistDiscography -v` Expected: FAIL - [ ] **Step 3: Implementa `search_artist_discography` in `core/spotify_client.py`** Aggiungi in cima al file (se non presente): ```python import time ``` Poi in fondo al file: ```python def search_artist_discography(token: str, artist_name: str) -> list: """Trova l'artista esatto (o il più popolare tra i match) e ritorna tutti i suoi brani: top tracks + tracce di ogni album/single. Deduplica per (name.lower().strip(), first_artist.lower().strip()). Raises: ValueError: se nessun artista trovato per il nome dato. """ headers = {"Authorization": f"Bearer {token}"} # 1. Cerca artista resp = requests.get( "https://api.spotify.com/v1/search", headers=headers, params={"q": artist_name, "type": "artist", "limit": 5}, timeout=15, ) resp.raise_for_status() candidates = resp.json().get("artists", {}).get("items", []) if not candidates: raise ValueError(f"Artista '{artist_name}' non trovato") # Match esatto (case-insensitive) se possibile, altrimenti più popolare query_lower = artist_name.lower().strip() exact = [c for c in candidates if c.get("name", "").lower().strip() == query_lower] if exact: artist = exact[0] else: artist = max(candidates, key=lambda c: c.get("popularity", 0)) artist_id = artist["id"] collected: list = [] # 2. Top tracks r_top = requests.get( f"https://api.spotify.com/v1/artists/{artist_id}/top-tracks", headers=headers, params={"market": "IT"}, timeout=15, ) r_top.raise_for_status() for t in r_top.json().get("tracks", []): collected.append(_track_to_dict(t)) # 3. Albums (album + single) r_alb = requests.get( f"https://api.spotify.com/v1/artists/{artist_id}/albums", headers=headers, params={"include_groups": "album,single", "limit": 50, "market": "IT"}, timeout=15, ) r_alb.raise_for_status() albums = r_alb.json().get("items", []) # 4. Per ogni album, tracce (album/track object non ha "album" sub-field, # gliela aggiungiamo esplicitamente) for alb in albums: alb_id = alb.get("id") alb_name = alb.get("name", "") if not alb_id: continue time.sleep(0.1) # rate limit interno r_at = requests.get( f"https://api.spotify.com/v1/albums/{alb_id}/tracks", headers=headers, params={"limit": 50}, timeout=15, ) r_at.raise_for_status() for t in r_at.json().get("items", []): t = dict(t) # Album tracks non hanno "album" nested; iniettiamo il nome t.setdefault("album", {"name": alb_name}) collected.append(_track_to_dict(t)) # 5. Dedupe seen: set = set() unique: list = [] for t in collected: key = (t["name"].lower().strip(), t["artists"].split(",")[0].lower().strip()) if key in seen: continue seen.add(key) unique.append(t) return unique ``` - [ ] **Step 4: Verifica** Run: `python3 -m pytest tests/test_spotify_client.py -v` Expected: 13 test PASS (4 preesistenti + 4 search_tracks + 5 search_artist_discography) - [ ] **Step 5: Commit** ```bash git add core/spotify_client.py tests/test_spotify_client.py git -c user.email=info@djluza.com commit -m "spotify: search_artist_discography con dedupe top-tracks+album" ``` --- ## Task 3: `core/youtube_search.py` — TDD **Files:** - Create: `core/youtube_search.py` - Create: `tests/test_youtube_search.py` - [ ] **Step 1: Test in `tests/test_youtube_search.py`** Crea il file: ```python """Test per core.youtube_search.""" from __future__ import annotations import json from unittest.mock import patch, MagicMock import pytest from core import youtube_search def _mock_ytdlp_result(entries: list) -> MagicMock: """Mock subprocess.CompletedProcess con JSON stub.""" result = MagicMock() result.returncode = 0 result.stdout = json.dumps({"entries": entries}) result.stderr = "" return result class TestSearchYoutube: def test_parses_entries(self): entries = [ { "id": "abc123", "url": "https://www.youtube.com/watch?v=abc123", "title": "Kapuchon - Hot Sauce (Official Video)", "uploader": "Kapuchon Official", "duration": 336, }, { "id": "def456", "url": "https://www.youtube.com/watch?v=def456", "title": "Solomun @ Cocoricò 2024", "uploader": "Cocoricò", "duration": 3600, }, ] with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \ patch("core.youtube_search.subprocess.run", return_value=_mock_ytdlp_result(entries)): result = youtube_search.search_youtube("kapuchon hot sauce", limit=50) assert len(result) == 2 assert result[0]["title"] == "Kapuchon - Hot Sauce (Official Video)" assert result[0]["channel"] == "Kapuchon Official" assert result[0]["duration_sec"] == 336 assert result[0]["url"] == "https://www.youtube.com/watch?v=abc123" def test_empty_entries_returns_empty(self): with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \ patch("core.youtube_search.subprocess.run", return_value=_mock_ytdlp_result([])): result = youtube_search.search_youtube("no-match", limit=50) assert result == [] def test_raises_when_ytdlp_missing(self): with patch("core.youtube_search.find_ytdlp", return_value=None): with pytest.raises(RuntimeError, match="yt-dlp"): youtube_search.search_youtube("q", limit=50) def test_command_uses_ytsearch_with_limit(self): with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \ patch("core.youtube_search.subprocess.run", return_value=_mock_ytdlp_result([])) as mock_run: youtube_search.search_youtube("solomun", limit=30) args = mock_run.call_args.args[0] # positional list assert args[0] == "/fake/yt-dlp" # Trova l'argomento ytsearchN:query search_arg = [a for a in args if a.startswith("ytsearch")] assert search_arg == ["ytsearch30:solomun"] def test_timeout_raises_runtime_error(self): import subprocess as sp with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \ patch("core.youtube_search.subprocess.run", side_effect=sp.TimeoutExpired("yt-dlp", 30)): with pytest.raises(RuntimeError, match="[Tt]imeout"): youtube_search.search_youtube("q", limit=50) def test_nonzero_exit_raises(self): bad = MagicMock() bad.returncode = 1 bad.stdout = "" bad.stderr = "some yt-dlp error" with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \ patch("core.youtube_search.subprocess.run", return_value=bad): with pytest.raises(RuntimeError, match="yt-dlp"): youtube_search.search_youtube("q", limit=50) ``` - [ ] **Step 2: Verifica fail** Run: `python3 -m pytest tests/test_youtube_search.py -v` Expected: FAIL con `ModuleNotFoundError: No module named 'core.youtube_search'` - [ ] **Step 3: Implementa `core/youtube_search.py`** Crea: ```python """Ricerca su YouTube via yt-dlp (subprocess, flat playlist metadata).""" from __future__ import annotations import json import subprocess from core.paths import find_ytdlp, subprocess_flags _TIMEOUT_SEC = 30 def search_youtube(query: str, limit: int = 50) -> list: """Cerca su YouTube (ytsearchN:query) e ritorna metadata "flat" (senza scaricare). Args: query: testo di ricerca libero limit: numero massimo di risultati (default 50) Returns: list[dict] con {id, url, title, channel, duration_sec}. Vuota se nessun match. Raises: RuntimeError: se yt-dlp non disponibile, timeout, o exit non-zero. """ ytdlp = find_ytdlp() if not ytdlp: raise RuntimeError("yt-dlp non trovato nel bundle / PATH") cmd = [ ytdlp, "--flat-playlist", "--dump-single-json", "--no-warnings", f"ytsearch{limit}:{query}", ] try: proc = subprocess.run( cmd, capture_output=True, text=True, timeout=_TIMEOUT_SEC, **subprocess_flags(), ) except subprocess.TimeoutExpired as e: raise RuntimeError(f"Timeout ricerca YouTube ({_TIMEOUT_SEC}s)") from e if proc.returncode != 0: tail = (proc.stderr or "").strip().splitlines()[-3:] raise RuntimeError(f"yt-dlp error: {' | '.join(tail) or 'unknown'}") try: data = json.loads(proc.stdout) except json.JSONDecodeError as e: raise RuntimeError(f"yt-dlp output non-JSON: {e}") from e entries = data.get("entries") or [] result: list = [] for e in entries: if not isinstance(e, dict): continue video_id = e.get("id") or "" url = e.get("url") or (f"https://www.youtube.com/watch?v={video_id}" if video_id else "") result.append({ "id": video_id, "url": url, "title": e.get("title") or "", "channel": e.get("uploader") or e.get("channel") or "", "duration_sec": int(e.get("duration") or 0), }) return result ``` **Nota:** `subprocess_flags()` è un helper esistente in `core/paths.py` che restituisce kwargs come `creationflags` su Windows per nascondere la console. Verifica: `grep subprocess_flags core/paths.py`. Se firma diversa, adatta. - [ ] **Step 4: Verifica** Run: `python3 -m pytest tests/test_youtube_search.py -v` Expected: 6 test PASS - [ ] **Step 5: Commit** ```bash git add core/youtube_search.py tests/test_youtube_search.py git -c user.email=info@djluza.com commit -m "youtube: search_youtube via yt-dlp ytsearchN + parse JSON" ``` --- ## Task 4: Downloader supporto URL diretti + helper bridge **Files:** - Modify: `core/downloader.py` (potenzialmente — verifica prima) - Modify: `api/bridge.py` (aggiunta helper) - [ ] **Step 1: Verifica se `download_playlist` accetta già URL diretti** Ispeziona `core/downloader.py`: ```bash sed -n '118,200p' core/downloader.py ``` Cerca: la funzione fa `_search_youtube(query, ...)` per ogni track. Se accetta come input `[{url: "https://..."}]` e salta la search quando `url` è presente, riusiamo. Altrimenti dobbiamo aggiungere una funzione parallela. - [ ] **Step 2A (se `download_playlist` NON supporta URL diretti): aggiungi `download_urls`** In `core/downloader.py`, aggiungi in fondo: ```python def download_urls( urls: list, titles: list, output_dir: str, bitrate: str = "320K", cookies_path: str = None, progress_callback=None, ) -> None: """Scarica direttamente da URL YouTube (bypassa search). Args: urls: lista URL YouTube (parallela a titles) titles: lista titoli video (per logging/dedupe) output_dir: cartella destinazione bitrate: qualita audio (default 320K) cookies_path: file cookies opzionale progress_callback: callback(idx, total, title, status, pct) """ from pathlib import Path reset_stop() total = len(urls) out_path = Path(output_dir) out_path.mkdir(parents=True, exist_ok=True) done_file = out_path / ".downloaded_tracks" done_set = _load_done_set(done_file) ytdlp = find_ytdlp() for i, (url, title) in enumerate(zip(urls, titles)): if is_stopped(): if progress_callback: progress_callback(i, total, "", "stopped", 0) return key = title or url if key in done_set: if progress_callback: progress_callback(i, total, key, "skipped", 100) continue if progress_callback: progress_callback(i, total, key, "downloading", 0) # Riusa la stessa logica di download del pipeline esistente # (chiama il worker download interno o replica i flag yt-dlp). # Vedi `_download_single_url` in questo file oppure la funzione # equivalente usata da download_playlist. try: _download_single_url(ytdlp, url, str(out_path), bitrate, cookies_path) _mark_done(done_file, key) if progress_callback: progress_callback(i, total, key, "done", 100) except Exception as e: if progress_callback: progress_callback(i, total, key, f"error: {e}", 0) if progress_callback: progress_callback(total, total, "", "completed", 100) ``` **Nota:** `_download_single_url` non esiste ancora — devi identificare la funzione interna che `download_playlist` chiama per il singolo download e (a) esporla come helper riusabile, oppure (b) estrarla in una nuova funzione. Ispeziona il corpo di `download_playlist` (righe 143+) per vedere il codice che va estratto. **Alternativa più semplice:** invece di estrarre, chiama direttamente `subprocess.run([ytdlp, "-x", "--audio-format", "mp3", "--audio-quality", bitrate, "-o",