diff --git a/api/bridge.py b/api/bridge.py index fd47c6b..f3d1033 100644 --- a/api/bridge.py +++ b/api/bridge.py @@ -12,8 +12,10 @@ from dataclasses import asdict from pathlib import Path from typing import Any, Optional +import requests + from core.config import load_config, save_config, VERSION, LICENSE_API_URL -from core import beatport +from core import beatport, spotify_client from core import license as license_mod from core.downloader import ( download_playlist, @@ -544,6 +546,40 @@ class Api: self._download_thread.start() return {"ok": True} + def start_urls_download(self, payload: dict) -> dict: + """Analogo a start_tracks_download ma accetta URL YouTube gia noti + (bypass search). Usato dal flow del tab 'YouTube Search'.""" + if self._any_job_running(): + return {"ok": False, "error": "Un download gia in corso"} + + urls = payload.get("urls") or [] + titles = payload.get("titles") or [] + output_dir = (payload.get("output_dir") or "").strip() + subfolder = (payload.get("subfolder") or "").strip() + if not urls: + return {"ok": False, "error": "Nessuna URL fornita"} + if len(urls) != len(titles): + return {"ok": False, "error": "urls e titles devono avere stessa lunghezza"} + if not output_dir: + return {"ok": False, "error": "Cartella output non impostata"} + + if subfolder: + safe = subfolder.replace("/", "_").replace("\\", "_").strip() + if safe: + output_dir = os.path.join(output_dir, safe) + + gate = self._gate("audio") + if gate: + return gate + + self._download_thread = threading.Thread( + target=self._urls_worker, + args=(list(urls), list(titles), output_dir), + daemon=True, + ) + self._download_thread.start() + return {"ok": True} + def stop_download(self) -> dict: request_download_stop() self._log("download", "[INFO] Interruzione richiesta...") @@ -601,6 +637,53 @@ class Api: download_playlist(tracks, output_dir, bitrate, cookies_path, progress_cb) self._emit("download:done", {"ok": True}) + def _urls_worker(self, urls: list, titles: list, output_dir: str) -> None: + from core.downloader import download_urls + reset_download_stop() + cfg = load_config() + bitrate = cfg.get("bitrate", "320K") + cookies_path = cfg.get("cookies_path", "") + view = "download" + + self._log(view, f"[INFO] URL list: {len(urls)} da scaricare da YouTube") + self._log(view, f"[INFO] Destinazione: {output_dir}") + + _last = [0.0] + _THROTTLE = 0.10 + + def progress_cb(idx, total, title, status, pct): + if status == "downloading": + now = time.monotonic() + if now - _last[0] < _THROTTLE: + return + _last[0] = now + + payload_evt = { + "idx": idx, "total": total, "track": title, + "status": status, "pct": pct, + "url_idx": 0, "url_total": 1, + } + if status == "skipped": + self._log(view, f"[SKIP] {title} (gia scaricato)") + payload_evt["overall"] = min((idx + 1) / total, 1.0) if total else 1 + elif status == "downloading": + payload_evt["overall"] = min((idx / total) + (pct / 100 / total), 1.0) if total else 0 + elif status == "done": + self._log(view, f"[OK] {title}") + payload_evt["overall"] = min((idx + 1) / total, 1.0) if total else 1 + elif status == "stopped": + self._log(view, "[INFO] Download interrotto.") + elif status == "completed": + payload_evt["overall"] = 1.0 + self._log(view, "[INFO] Download completato!") + elif status.startswith("error"): + self._log(view, f"[ERRORE] {title}: {status}") + + self._emit("download:progress", payload_evt) + + download_urls(urls, titles, output_dir, bitrate, cookies_path, progress_cb) + self._emit("download:done", {"ok": True}) + def _download_worker(self, urls: list, output_dir: str) -> None: reset_download_stop() cfg = load_config() @@ -1006,3 +1089,184 @@ class Api: "output_dir": out_root, "subfolder": subfolder, }) + + # ================================================================ + # Music Search — Spotify + YouTube + # ================================================================ + + def _music_output_dir(self, out_root: str, source: str) -> Path: + """Cartella target per Spotify/YouTube search. Coerente col pattern + di start_tracks_download (subfolder singolo, no nested).""" + safe = source.replace("/", "_").replace("\\", "_").strip() + return Path(out_root) / safe + + # ---- Spotify ---- + + def spotify_search(self, query: str, artist_mode: bool = False) -> dict: + """Cerca su Spotify. Free-form (limit 50) o artist-mode (discografia).""" + query = (query or "").strip() + if not query: + return {"ok": False, "error": "empty_query", "message": "Query vuota"} + + # Salva stato + try: + cfg = load_config() + cfg["spotify_search_last_query"] = query + cfg["spotify_search_artist_mode"] = bool(artist_mode) + save_config(cfg) + except Exception: + pass + + cfg = load_config() + cid = (cfg.get("client_id") or "").strip() + secret = (cfg.get("client_secret") or "").strip() + if not cid or not secret: + return {"ok": False, "error": "no_creds", "message": "Credenziali Spotify mancanti"} + + try: + token = spotify_client.get_access_token(cid, secret) + except Exception as e: + return {"ok": False, "error": "auth", "message": str(e)} + + try: + if artist_mode: + tracks = spotify_client.search_artist_discography(token, query) + else: + tracks = spotify_client.search_tracks(token, query, limit=50) + except ValueError as e: + return {"ok": False, "error": "artist_not_found", "message": str(e)} + except requests.HTTPError as e: + code = e.response.status_code if e.response is not None else 0 + if code == 429: + return {"ok": False, "error": "rate_limit", "message": "Spotify limitante - attendi qualche secondo"} + return {"ok": False, "error": "server", "message": f"Spotify HTTP {code}"} + except Exception as e: + return {"ok": False, "error": "unknown", "message": str(e)} + + return {"ok": True, "tracks": tracks} + + def spotify_check_existing(self, tracks: list) -> list: + """True per ogni track già presente in output_dir/Spotify/.""" + cfg = load_config() + out_root = (cfg.get("output_dir") or "").strip() + if not out_root: + return [False] * len(tracks) + out_dir = self._music_output_dir(out_root, "Spotify") + if not out_dir.exists(): + return [False] * len(tracks) + existing_stems = [p.stem.lower() for p in out_dir.glob("*.mp3")] + result = [] + for t in tracks: + title = (t.get("name") or "").lower().strip() + artists = (t.get("artists") or "") + first_artist = artists.split(",")[0].strip().lower() + if not title or not first_artist: + result.append(False) + continue + result.append(any((title in stem and first_artist in stem) for stem in existing_stems)) + return result + + def spotify_search_download(self, tracks: list) -> dict: + """Scarica i track Spotify selezionati (name+artist → YouTube search).""" + cfg = load_config() + out_root = (cfg.get("output_dir") or "").strip() + if not out_root: + return {"ok": False, "error": "Cartella output non impostata"} + + target = self._music_output_dir(out_root, "Spotify") + subfolder = target.name + + converted = [] + for t in tracks: + title = (t.get("name") or "").strip() + artists = (t.get("artists") or "").strip() + if not title: + continue + converted.append({"name": title, "artist": artists}) + + if not converted: + return {"ok": False, "error": "Nessun brano valido"} + + return self.start_tracks_download({ + "tracks": converted, + "output_dir": out_root, + "subfolder": subfolder, + }) + + # ---- YouTube ---- + + def youtube_search(self, query: str) -> dict: + """Cerca 50 risultati su YouTube via yt-dlp.""" + from core import youtube_search as yts + + query = (query or "").strip() + if not query: + return {"ok": False, "error": "empty_query", "message": "Query vuota"} + + try: + cfg = load_config() + cfg["youtube_search_last_query"] = query + save_config(cfg) + except Exception: + pass + + try: + results = yts.search_youtube(query, limit=50) + except RuntimeError as e: + return {"ok": False, "error": "ytdlp", "message": str(e)} + except Exception as e: + return {"ok": False, "error": "unknown", "message": str(e)} + + return {"ok": True, "tracks": results} + + def youtube_check_existing(self, tracks: list) -> list: + """True per ogni track già presente in output_dir/YouTube/. Match sul titolo video.""" + cfg = load_config() + out_root = (cfg.get("output_dir") or "").strip() + if not out_root: + return [False] * len(tracks) + out_dir = self._music_output_dir(out_root, "YouTube") + if not out_dir.exists(): + return [False] * len(tracks) + existing_stems = [p.stem.lower() for p in out_dir.glob("*.mp3")] + result = [] + for t in tracks: + title = (t.get("title") or "").lower().strip() + if not title: + result.append(False) + continue + result.append(any( + (title in stem or stem in title) + for stem in existing_stems + )) + return result + + def youtube_search_download(self, tracks: list) -> dict: + """Scarica direttamente gli URL YouTube selezionati (no re-search).""" + cfg = load_config() + out_root = (cfg.get("output_dir") or "").strip() + if not out_root: + return {"ok": False, "error": "Cartella output non impostata"} + + target = self._music_output_dir(out_root, "YouTube") + subfolder = target.name + + urls: list = [] + titles: list = [] + for t in tracks: + url = (t.get("url") or "").strip() + title = (t.get("title") or "").strip() + if not url or not title: + continue + urls.append(url) + titles.append(title) + + if not urls: + return {"ok": False, "error": "Nessun URL valido"} + + return self.start_urls_download({ + "urls": urls, + "titles": titles, + "output_dir": out_root, + "subfolder": subfolder, + }) diff --git a/core/config.py b/core/config.py index 5dddc45..96a759f 100644 --- a/core/config.py +++ b/core/config.py @@ -5,7 +5,7 @@ import os import sys from pathlib import Path -VERSION = "v1.8.0" +VERSION = "v1.8.1" APP_NAME = "MusicTools" @@ -72,6 +72,10 @@ DEFAULTS = { "theme": "dark", # ---- Beatport ---- "beatport_last_genre": "melodic-house-techno", # ultimo genere Top 100 caricato + # ---- Music Search (Spotify + YouTube) ---- + "spotify_search_last_query": "", + "spotify_search_artist_mode": False, + "youtube_search_last_query": "", # ---- Licenza ---- "license_key": "", # chiave fornita all'utente via email "license_email": "", # email associata all'acquisto diff --git a/core/downloader.py b/core/downloader.py index e28151f..1aeaab6 100644 --- a/core/downloader.py +++ b/core/downloader.py @@ -399,6 +399,125 @@ def download_direct_url( progress_callback(total, total, "", "completed", 100) +def download_urls( + urls: list, + titles: list, + output_dir: str, + bitrate: str = "320K", + cookies_path: Optional[str] = None, + progress_callback: Optional[Callable] = None, +) -> None: + """Scarica direttamente da URL YouTube (bypassa search). + + Usa gli stessi flag yt-dlp di download_playlist. La differenza e che + salta lo step di ricerca perche l'URL e gia noto (usato dal flow del + tab "YouTube Search" dopo che l'utente ha selezionato i risultati). + + 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, percent) + """ + global _current_process + 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) + existing_files = _scan_existing_files(out_path) + 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 + + # Salta se gia scaricato (file di tracking) + if key in done_set: + if progress_callback: + progress_callback(i, total, key, "skipped", 100) + continue + + # Salta se esiste gia un file corrispondente nella cartella + if title and _file_exists_for_query(title, existing_files): + _mark_done(done_file, key) + done_set.add(key) + if progress_callback: + progress_callback(i, total, key, "skipped", 100) + continue + + if progress_callback: + progress_callback(i, total, key, "downloading", 0) + + cmd = [ + ytdlp, + "--extract-audio", + "--audio-format", "mp3", + "--audio-quality", bitrate.rstrip("Kk"), + "--embed-thumbnail", + "--add-metadata", + "--no-check-certificates", + "--newline", + "--output", str(Path(output_dir) / "%(title)s.%(ext)s"), + url, + ] + ffmpeg_dir = find_ffmpeg_dir() + if ffmpeg_dir: + cmd.extend(["--ffmpeg-location", ffmpeg_dir]) + if cookies_path and Path(cookies_path).exists(): + cmd.extend(["--cookies", cookies_path]) + + try: + with _process_lock: + _current_process = subprocess.Popen( + cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, + **subprocess_flags(), + ) + + for line in _current_process.stdout: + if is_stopped(): + _current_process.terminate() + if progress_callback: + progress_callback(i, total, key, "stopped", 0) + return + + pct_match = re.search(r"(\d+(?:\.\d+)?)%", line) + if pct_match and progress_callback: + pct = int(float(pct_match.group(1))) + progress_callback(i, total, key, "downloading", pct) + + return_code = _current_process.wait() + + with _process_lock: + _current_process = None + + if return_code == 0: + _mark_done(done_file, key) + done_set.add(key) + existing_files = _scan_existing_files(out_path) + if progress_callback: + progress_callback(i, total, key, "done", 100) + else: + if progress_callback: + progress_callback(i, total, key, f"error: yt-dlp exit {return_code}", 0) + + except Exception as e: + with _process_lock: + _current_process = None + if progress_callback: + progress_callback(i, total, key, f"error: {e}", 0) + + if progress_callback and not is_stopped(): + progress_callback(total, total, "", "completed", 100) + + def _video_format(quality: str) -> str: """Costruisce la format string yt-dlp per la qualita richiesta.""" if not quality or quality == "best": diff --git a/core/spotify_client.py b/core/spotify_client.py index 77570ed..34ea1c1 100644 --- a/core/spotify_client.py +++ b/core/spotify_client.py @@ -3,6 +3,8 @@ from __future__ import annotations import re +import time + import requests @@ -280,3 +282,123 @@ def search_track(token: str, query: str): "name": t.get("name", ""), "artists": ", ".join(a.get("name", "") for a in t.get("artists", [])), } + + +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, + } + + +def search_artist_discography(token: str, artist_name: str) -> list: + """Trova l'artista esatto (o il piu' 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 piu' 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, iniettiamola) + 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 diff --git a/core/youtube_search.py b/core/youtube_search.py new file mode 100644 index 0000000..30f2b10 --- /dev/null +++ b/core/youtube_search.py @@ -0,0 +1,72 @@ +"""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 diff --git a/docs/superpowers/plans/2026-07-14-music-search.md b/docs/superpowers/plans/2026-07-14-music-search.md new file mode 100644 index 0000000..f5dbc9c --- /dev/null +++ b/docs/superpowers/plans/2026-07-14-music-search.md @@ -0,0 +1,1747 @@ +# 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",