"""Ponte JS <-> Python: tutti i metodi che la UI invoca via pywebview.api.""" from __future__ import annotations import json import os import queue import re import threading import time import webbrowser from dataclasses import asdict from pathlib import Path from typing import Any, Callable, Optional import requests from core.config import load_config, save_config, VERSION, LICENSE_API_URL from core import beatport, spotify_client, traxsource from core import license as license_mod from core.downloader import ( download_playlist, download_direct_url, download_video, request_stop as request_download_stop, reset_stop as reset_download_stop, is_stopped as download_is_stopped, ) from core.spotify_client import ( get_access_token, resolve_spotify_url, resolve_spotify_track_no_auth, SpotifyAuthRequired, ) from core.metadata import read_metadata, write_metadata, SUPPORTED_EXTS from core import tagger from core.recorder import ( list_input_devices, start_recording, stop_recording as recorder_stop, is_recording, get_last_stderr as recorder_last_stderr, ) from core.upgrader import ( upgrade_folder, request_stop as request_upgrade_stop, count_files_info, ) from core import dedup from core import catalog from core import flatten from core import charts SPOTIFY_GUIDE_TEXT = """\ NOTE: Spotify oggi consente di creare le chiavi API SOLO con un account Spotify Premium attivo. Senza Premium puoi comunque scaricare: - YouTube, SoundCloud, TikTok, Instagram, Facebook (senza setup) - Brani Spotify SINGOLI (incolla il link e basta, l'app fa il resto) Le chiavi Spotify servono solo per playlist e album da Spotify. Procedura (se hai Spotify Premium): 1) Vai su https://developer.spotify.com/dashboard e accedi. 2) Clicca "Create app". 3) Compila: - App name: MusicTools (o quello che vuoi) - Description: a piacere - Redirect URI: http://localhost:8888/callback (obbligatorio ma non usato) - Seleziona "Web API" - Accetta i termini, "Save" 4) Apri la app appena creata > "Settings" e copia: - Client ID - Client Secret (clicca "View client secret") 5) Incollali qui sotto e premi "Salva". Limite tecnico Spotify: 100 richieste/minuto. """ def _js_safe(value: Any) -> str: """Serializza un valore per inserirlo in evaluate_js.""" return json.dumps(value, ensure_ascii=False) _NUM_PREFIX = re.compile(r"^\s*\d+\s*[\.\)\:\-]\s*") _DURATION_TAIL = re.compile(r"\s*\(\s*\d{1,2}:\d{2}\s*\)\s*$") # Tutti i tipi di trattini: hyphen, en-dash, em-dash, horizontal bar, minus _DASH_SPLIT = re.compile(r"\s+[‐‑‒–—―−\-]\s+") def _parse_track_line(line: str) -> Optional[dict]: """Parse 'N. Artista - Titolo (variant) (durata)' -> {name, artist}. Ritorna None se la riga non e parseabile.""" s = line.strip() if not s: return None s = _NUM_PREFIX.sub("", s) s = _DURATION_TAIL.sub("", s).strip() parts = _DASH_SPLIT.split(s, maxsplit=1) if len(parts) < 2: return None artist = parts[0].strip() name = parts[1].strip() if not artist or not name: return None return {"name": name, "artist": artist} class Api: """Espone i metodi a JavaScript tramite pywebview js_api.""" def __init__(self): self.window = None # impostato dopo create_window self._download_thread: Optional[threading.Thread] = None self._upgrade_thread: Optional[threading.Thread] = None self._video_thread: Optional[threading.Thread] = None self._dedup_thread: Optional[threading.Thread] = None self._catalog_thread: Optional[threading.Thread] = None self._flatten_thread: Optional[threading.Thread] = None # Coda usata dal resolve_callback per attendere la scelta utente # sul modal "match locali multipli" della tab Upgrade. self._upgrade_resolve_q: "queue.Queue[dict]" = queue.Queue(1) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _emit(self, channel: str, payload: dict | str | None = None) -> None: """Invia un evento al frontend.""" if not self.window: return try: self.window.evaluate_js( f"window.bridge && window.bridge.emit({_js_safe(channel)}, {_js_safe(payload)});" ) except Exception: pass def _log(self, view: str, msg: str) -> None: # Emit alla UI + mirror sul file di log persistente self._emit("log", {"view": view, "msg": msg}) try: import logging logging.getLogger(f"ui.{view}").info(msg) except Exception: pass # ------------------------------------------------------------------ # Log persistente # ------------------------------------------------------------------ def get_log_file_path(self) -> str: """Ritorna il path del file di log corrente (per apertura da UI).""" try: from core.applog import log_file_path return str(log_file_path()) except Exception: return "" def open_log_folder(self) -> dict: """Apre la cartella dei log nel file manager del sistema.""" try: from core.applog import log_file_path import subprocess folder = log_file_path().parent if sys.platform == "darwin": subprocess.Popen(["open", str(folder)]) elif sys.platform == "win32": os.startfile(str(folder)) # type: ignore[attr-defined] else: subprocess.Popen(["xdg-open", str(folder)]) return {"ok": True, "path": str(folder)} except Exception as e: return {"ok": False, "error": str(e)} # ------------------------------------------------------------------ # Bootstrap # ------------------------------------------------------------------ def get_init_data(self) -> dict: """Chiamato dal frontend all'avvio per popolare lo stato iniziale.""" cfg = load_config() # Non spediamo il token raw al frontend (e' sensibile e non serve in UI) safe_cfg = {k: v for k, v in cfg.items() if k != "license_token"} license_status = license_mod.get_status(cfg) return { "version": VERSION, "config": safe_cfg, "spotify_guide": SPOTIFY_GUIDE_TEXT, "license": license_status, "plan": license_mod.get_plan(cfg), "purchase_url": "https://musictools.djluza.com", } # ------------------------------------------------------------------ # Settings # ------------------------------------------------------------------ def save_settings(self, payload: dict) -> dict: try: threshold = int(payload.get("hq_threshold", 310)) except (TypeError, ValueError): threshold = 310 # Merge sul config esistente per preservare i campi licenza # (license_key/token/email/...), device_id, ecc. config = load_config() config.update({ "client_id": (payload.get("client_id") or "").strip(), "client_secret": (payload.get("client_secret") or "").strip(), "bitrate": payload.get("bitrate", "320K"), "hq_threshold": threshold, "cookies_path": (payload.get("cookies_path") or "").strip(), "cookies_browser": (payload.get("cookies_browser") or "").strip().lower(), "output_dir": (payload.get("output_dir") or "").strip(), "theme": payload.get("theme", "dark"), }) save_config(config) return {"ok": True} # ------------------------------------------------------------------ # Licenza # ------------------------------------------------------------------ def get_license_status(self) -> dict: return license_mod.get_status() def activate_license(self, payload: dict) -> dict: try: status = license_mod.activate( payload.get("key") or "", payload.get("email") or "", ) return {"ok": True, "license": status} except license_mod.LicenseNetworkError as e: return {"ok": False, "error": f"Impossibile contattare il server: {e}"} except license_mod.LicenseError as e: return {"ok": False, "error": str(e)} def deactivate_license(self) -> dict: status = license_mod.deactivate(release_remote=True) return {"ok": True, "license": status} def revalidate_license(self) -> dict: """Revalidate in foreground (chiamabile dalla UI 'Verifica ora').""" status = license_mod.validate() return {"ok": True, "license": status} def open_purchase_page(self) -> dict: webbrowser.open("https://musictools.djluza.com") return {"ok": True} # ------------------------------------------------------------------ # Quota giornaliera # ------------------------------------------------------------------ def get_quota_status(self) -> dict: """Ritorna lo stato quota corrente (lettura senza incremento).""" try: data = license_mod.get_quota_status() return {"ok": True, "quota": data} except license_mod.LicenseNetworkError as e: return {"ok": False, "offline": True, "error": str(e)} except license_mod.LicenseError as e: return {"ok": False, "error": str(e)} def _gate(self, feature: str) -> Optional[dict]: """Verifica feature + quota prima di un'azione. Ritorna None se l'azione e' permessa; altrimenti un dict di errore pronto per la UI (con eventuale info quota per il modal "limite raggiunto"). """ # 1) Feature gate locale (dai claims JWT in config) if not license_mod.has_feature(feature): plan = license_mod.get_plan() plan_name = plan.get("name") or "il tuo piano" return { "ok": False, "reason": "feature_not_in_plan", "feature": feature, "error": ( f"Questa funzione non e' inclusa in {plan_name}. " "Passa a un piano superiore per sbloccarla." ), "plan": plan, } # 2) Quota gate server-side: skippato per i piani senza limite # giornaliero (Annual) — evita una chiamata HTTP inutile e fa # funzionare l'app anche offline per quei piani. plan = license_mod.get_plan() if plan.get("daily_limit") is None: return None try: res = license_mod.consume_quota(feature) except license_mod.LicenseNetworkError as e: return { "ok": False, "reason": "offline", "offline": True, "error": f"Impossibile contattare il server: {e}", } except license_mod.LicenseError as e: return {"ok": False, "reason": "license_invalid", "error": str(e)} if not res.get("allowed", True): return { "ok": False, "reason": "quota_exceeded", "feature": feature, "error": res.get("error") or "Limite giornaliero raggiunto.", "quota": res, } # OK: notifico la UI dello stato quota aggiornato self._emit("quota:update", res) return None def check_update(self) -> dict: """Controlla aggiornamenti interrogando l'API djluza.com. Il server identifica la piattaforma e restituisce un URL di download firmato a tempo. Senza un token di licenza valido ritorna comunque la versione corrente ma con download_url vuoto. """ import platform import urllib.error import urllib.request cfg = load_config() token = (cfg.get("license_token") or "").strip() is_macos = platform.system() == "Darwin" plat = "macos" if is_macos else "windows" url = f"{LICENSE_API_URL.rstrip('/')}/api/latest?platform={plat}¤t={VERSION}" headers = { "User-Agent": f"MusicTools/{VERSION}", "Accept": "application/json", } if token: headers["Authorization"] = f"Bearer {token}" req = urllib.request.Request(url, headers=headers, method="GET") try: with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read().decode("utf-8")) remote = data.get("version", "") notes = data.get("notes", "") or "" download_url = data.get("download_url", "") or "" return { "ok": True, "current": VERSION, "remote": remote, "is_new": bool(remote and remote != VERSION), "download_url": download_url, "notes": notes[:500] if notes else "", "requires_license": not bool(download_url) and not token, } except urllib.error.HTTPError as e: return {"ok": False, "error": f"HTTP {e.code}: {e.reason}", "current": VERSION} except (urllib.error.URLError, TimeoutError, OSError) as e: return {"ok": False, "error": f"Connessione fallita: {e}", "current": VERSION} except Exception as e: return {"ok": False, "error": str(e), "current": VERSION} # ------------------------------------------------------------------ # File dialogs (delegati a pywebview) # ------------------------------------------------------------------ def _folder_dialog_type(self): import webview # API nuova (>= 5.x): webview.FileDialog.FOLDER # API vecchia: webview.FOLDER_DIALOG if hasattr(webview, "FileDialog"): return webview.FileDialog.FOLDER return webview.FOLDER_DIALOG def _open_dialog_type(self): import webview if hasattr(webview, "FileDialog"): return webview.FileDialog.OPEN return webview.OPEN_DIALOG def browse_directory(self) -> str: if not self.window: return "" result = self.window.create_file_dialog(self._folder_dialog_type()) if result: return result[0] if isinstance(result, (list, tuple)) else result return "" def browse_file(self, filetypes: Optional[list] = None) -> str: if not self.window: return "" if filetypes: ft = tuple(filetypes) else: ft = ("Text files (*.txt)", "All files (*.*)") result = self.window.create_file_dialog(self._open_dialog_type(), file_types=ft) if result: return result[0] if isinstance(result, (list, tuple)) else result return "" def load_url_list(self, path: str) -> dict: """Carica un file .txt. Riconosce automaticamente se contiene URL oppure una tracklist 'Artista - Titolo'.""" try: with open(path, "r", encoding="utf-8") as f: lines = [line.strip() for line in f if line.strip() and not line.strip().startswith("#")] if not lines: return {"ok": True, "kind": "urls", "urls": [], "count": 0} # Detect: se la prima riga e un URL trattiamo tutto come URL list if lines[0].lower().startswith(("http://", "https://")): return {"ok": True, "kind": "urls", "urls": lines, "count": len(lines)} # Altrimenti parse come tracklist tracks = [] for line in lines: t = _parse_track_line(line) if t: tracks.append(t) name = os.path.splitext(os.path.basename(path))[0] return { "ok": True, "kind": "tracks", "tracks": tracks, "count": len(tracks), "name": name, } except Exception as e: return {"ok": False, "error": str(e)} def open_external_url(self, url: str) -> None: webbrowser.open(url) # ------------------------------------------------------------------ # Metadata editor # ------------------------------------------------------------------ def pick_audio_file(self) -> str: return self.browse_file([ "Audio (*.mp3;*.m4a;*.mp4;*.aac;*.flac;*.wav)", "All files (*.*)", ]) def pick_image_file(self) -> str: return self.browse_file([ "Immagini (*.jpg;*.jpeg;*.png)", "All files (*.*)", ]) def read_metadata(self, path: str) -> dict: try: data = read_metadata(path) return {"ok": True, "data": data} except Exception as e: return {"ok": False, "error": str(e)} def save_metadata(self, payload: dict) -> dict: path = (payload.get("path") or "").strip() if not path: return {"ok": False, "error": "Path mancante"} if not os.path.exists(path): return {"ok": False, "error": "File non trovato"} gate = self._gate("metadata") if gate: return gate data = payload.get("data") or {} cover_path = (payload.get("cover_path") or "").strip() or None remove_cover = bool(payload.get("remove_cover", False)) try: write_metadata(path, data, cover_path, remove_cover) return {"ok": True} except Exception as e: return {"ok": False, "error": str(e)} # ------------------------------------------------------------------ # Recorder # ------------------------------------------------------------------ def list_audio_inputs(self) -> dict: try: return {"ok": True, "devices": list_input_devices()} except Exception as e: return {"ok": False, "error": str(e), "devices": []} def start_audio_recording(self, payload: dict) -> dict: device_id = str(payload.get("device_id", "")).strip() output_dir = (payload.get("output_dir") or "").strip() filename = (payload.get("filename") or "").strip() bitrate = (payload.get("bitrate") or "320k").strip() if not device_id: return {"ok": False, "error": "Dispositivo non selezionato"} if not output_dir: return {"ok": False, "error": "Cartella output non impostata"} if not filename: filename = f"Registrazione_{time.strftime('%Y%m%d_%H%M%S')}.mp3" if not filename.lower().endswith(".mp3"): filename += ".mp3" gate = self._gate("record") if gate: return gate out_path = os.path.join(output_dir, filename) def cb(status, payload_evt): payload_evt = dict(payload_evt or {}) payload_evt["status"] = status self._emit("recording:event", payload_evt) res = start_recording(device_id, out_path, bitrate=bitrate, progress_callback=cb) return res def stop_audio_recording(self) -> dict: return recorder_stop() def is_audio_recording(self) -> dict: return {"recording": is_recording()} def get_recorder_log(self) -> dict: """Ritorna le ultime righe di stderr di ffmpeg dell'ultima sessione di registrazione. Utile per diagnosticare un errore.""" return {"ok": True, "lines": recorder_last_stderr(80)} # ------------------------------------------------------------------ # Audio folder scan (Upgrade tab) # ------------------------------------------------------------------ def scan_audio_folder(self, directory: str, recursive: bool) -> dict: try: total, done = count_files_info(directory, bool(recursive)) return {"ok": True, "total": total, "done": done, "remaining": total - done} except Exception as e: return {"ok": False, "error": str(e)} # ------------------------------------------------------------------ # DOWNLOAD # ------------------------------------------------------------------ def _any_job_running(self) -> bool: return ( (self._download_thread and self._download_thread.is_alive()) or (self._video_thread and self._video_thread.is_alive()) ) def start_download(self, payload: dict) -> dict: if self._any_job_running(): return {"ok": False, "error": "Un download gia in corso"} urls = payload.get("urls") or [] output_dir = (payload.get("output_dir") or "").strip() if not urls: return {"ok": False, "error": "Nessun URL fornito"} if not output_dir: return {"ok": False, "error": "Cartella output non impostata"} gate = self._gate("audio") if gate: return gate self._download_thread = threading.Thread( target=self._download_worker, args=(list(urls), output_dir), daemon=True, ) self._download_thread.start() return {"ok": True} def start_tracks_download(self, payload: dict) -> dict: """Avvia il download di una tracklist gia parsata (lista di {name, artist}). Se 'subfolder' e presente, scarica in output_dir/subfolder. `payload['metadata']` opzionale: lista parallela a `tracks` con dict per il tagging ID3 (title, artist, album, date, genre, tracknumber, cover_url). Se presente, i file MP3 vengono nominati come "Artista - Titolo.mp3" e taggati post-download. """ if self._any_job_running(): return {"ok": False, "error": "Un download gia in corso"} tracks = payload.get("tracks") or [] output_dir = (payload.get("output_dir") or "").strip() subfolder = (payload.get("subfolder") or "").strip() metadata_list = payload.get("metadata") or None if not tracks: return {"ok": False, "error": "Nessuna traccia fornita"} if not output_dir: return {"ok": False, "error": "Cartella output non impostata"} if subfolder: # Sanitizza: rimuovi separatori di path 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._tracks_worker, args=(list(tracks), output_dir), kwargs={"metadata_list": list(metadata_list) if metadata_list else None}, daemon=True, ) 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'. `payload['metadata']` opzionale: lista parallela a `urls` con dict per il tagging ID3 (title, artist, album, date, genre, tracknumber, cover_url). Se presente, i file MP3 vengono nominati come "Artista - Titolo.mp3" e taggati post-download. """ 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() metadata_list = payload.get("metadata") or None 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), kwargs={"metadata_list": list(metadata_list) if metadata_list else None}, daemon=True, ) self._download_thread.start() return {"ok": True} def stop_download(self) -> dict: request_download_stop() self._log("download", "[INFO] Interruzione richiesta...") return {"ok": True} def _tracks_worker(self, tracks: list, output_dir: str, metadata_list: Optional[list] = None) -> None: reset_download_stop() cfg = load_config() bitrate = cfg.get("bitrate", "320K") cookies_path = cfg.get("cookies_path", "") view = "download" self._log(view, f"[INFO] Tracklist: {len(tracks)} brani da cercare su YouTube") self._log(view, f"[INFO] Destinazione: {output_dir}") # Pipeline tagging: se metadata_list fornito, calcola i filename # ("Artista - Titolo") e prepara callback per scrivere ID3 tag # + cover art dopo ogni download riuscito. output_filenames: Optional[list] = None post_cb: Optional[Callable] = None if metadata_list: output_filenames = [] for md in metadata_list: md = md or {} stem = tagger.build_filename_stem( md.get("artist", ""), md.get("title", ""), ) output_filenames.append(stem) def post_cb(idx: int, filepath: str, _md_list=metadata_list) -> None: if idx < 0 or idx >= len(_md_list): return md = _md_list[idx] or {} try: tagger.write_tags(filepath, md) except Exception: pass # tagging best-effort, non blocca il download _last = [0.0] _THROTTLE = 0.10 def progress_cb(idx, total, track_name, status, pct): if status == "downloading": now = time.monotonic() if now - _last[0] < _THROTTLE: return _last[0] = now payload_evt = { "idx": idx, "total": total, "track": track_name, "status": status, "pct": pct, "url_idx": 0, "url_total": 1, } if status == "searching": self._log(view, f"[CERCA] {track_name}") elif status == "skipped": self._log(view, f"[SKIP] {track_name} (gia scaricato)") track_progress = (idx + 1) / total if total > 0 else 1 payload_evt["overall"] = min(track_progress, 1.0) elif status == "downloading": track_progress = (idx / total) + (pct / 100 / total) if total > 0 else 0 payload_evt["overall"] = min(track_progress, 1.0) elif status == "done": self._log(view, f"[OK] {track_name}") track_progress = (idx + 1) / total if total > 0 else 1 payload_evt["overall"] = min(track_progress, 1.0) 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] {track_name}: {status}") self._emit("download:progress", payload_evt) download_playlist( tracks, output_dir, bitrate, cookies_path, progress_cb, output_filenames=output_filenames, post_download_callback=post_cb, ) self._emit("download:done", {"ok": True}) def _urls_worker(self, urls: list, titles: list, output_dir: str, metadata_list: Optional[list] = None) -> 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}") # Stessa pipeline tagging di _tracks_worker output_filenames: Optional[list] = None post_cb: Optional[Callable] = None if metadata_list: output_filenames = [] for md in metadata_list: md = md or {} stem = tagger.build_filename_stem( md.get("artist", ""), md.get("title", ""), ) output_filenames.append(stem) def post_cb(idx: int, filepath: str, _md_list=metadata_list) -> None: if idx < 0 or idx >= len(_md_list): return md = _md_list[idx] or {} try: tagger.write_tags(filepath, md) except Exception: pass # tagging best-effort, non blocca il download _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, output_filenames=output_filenames, post_download_callback=post_cb, ) self._emit("download:done", {"ok": True}) def _download_worker(self, urls: list, output_dir: str) -> None: reset_download_stop() cfg = load_config() bitrate = cfg.get("bitrate", "320K") cookies_path = cfg.get("cookies_path", "") total_urls = len(urls) multi = total_urls > 1 view = "download" token = None client_id = cfg.get("client_id", "") client_secret = cfg.get("client_secret", "") has_credentials = bool(client_id and client_secret) has_spotify = any("spotify.com" in u for u in urls) if has_spotify and has_credentials: self._log(view, "[INFO] Autenticazione Spotify...") try: token = get_access_token(client_id, client_secret) except Exception as e: self._log(view, f"[ERRORE] Autenticazione fallita: {e}") self._emit("download:done", {"ok": False}) return elif has_spotify: self._log( view, "[INFO] Spotify API non configurate: i singoli brani " "verranno comunque scaricati via metadata pubblico. " "Per album/playlist servono le credenziali (gratuite, 2 min)." ) for url_idx, url in enumerate(urls): if download_is_stopped(): self._log(view, "[INFO] Download interrotto.") break if multi: self._log(view, f"\n{'='*50}") self._log(view, f"[LISTA] URL {url_idx + 1} di {total_urls}") self._log(view, f"{'='*50}") is_spotify = "spotify.com" in url _last = [0.0] _THROTTLE = 0.10 url_base = url_idx / total_urls if multi else 0 url_weight = 1 / total_urls if multi else 1 def progress_cb(idx, total, track_name, status, pct, _base=url_base, _weight=url_weight, _uidx=url_idx): if status == "downloading": now = time.monotonic() if now - _last[0] < _THROTTLE: return _last[0] = now payload_evt = { "idx": idx, "total": total, "track": track_name, "status": status, "pct": pct, "url_idx": _uidx, "url_total": total_urls, } if status == "searching": self._log(view, f"[CERCA] {track_name}") elif status == "skipped": self._log(view, f"[SKIP] {track_name} (gia scaricato)") track_progress = (idx + 1) / total if total > 0 else 1 payload_evt["overall"] = min(_base + track_progress * _weight, 1.0) elif status == "downloading": track_progress = (idx / total) + (pct / 100 / total) if total > 0 else 0 payload_evt["overall"] = min(_base + track_progress * _weight, 1.0) elif status == "done": self._log(view, f"[OK] {track_name}") track_progress = (idx + 1) / total if total > 0 else 1 payload_evt["overall"] = min(_base + track_progress * _weight, 1.0) elif status == "stopped": self._log(view, "[INFO] Download interrotto.") elif status == "completed": if not multi: payload_evt["overall"] = 1.0 self._log(view, "[INFO] Download completato!") elif status.startswith("error"): self._log(view, f"[ERRORE] {track_name}: {status}") self._emit("download:progress", payload_evt) if is_spotify: self._log(view, "[INFO] Recupero informazioni da Spotify...") try: if token: label, name, tracks = resolve_spotify_url(token, url) else: # Senza API keys: fallback solo per singoli brani. label, name, tracks = resolve_spotify_track_no_auth(url) except SpotifyAuthRequired as e: self._log(view, f"[ERRORE] {e}") continue except Exception as e: self._log(view, f"[ERRORE] {e}") continue self._log(view, f"[INFO] {label}: {name} ({len(tracks)} brani)") dest_dir = os.path.join(output_dir, name) if len(tracks) > 1 else output_dir download_playlist(tracks, dest_dir, bitrate, cookies_path, progress_cb) else: self._log(view, f"[INFO] Download diretto via yt-dlp: {url}") download_direct_url(url, output_dir, bitrate, cookies_path, progress_cb) if multi and not download_is_stopped(): self._log(view, f"\n[INFO] Completate tutte le {total_urls} playlist/URL!") self._emit("download:progress", { "status": "completed", "overall": 1.0, "url_total": total_urls, }) self._emit("download:done", {"ok": True}) # ------------------------------------------------------------------ # UPGRADE # ------------------------------------------------------------------ def start_upgrade(self, payload: dict) -> dict: if self._upgrade_thread and self._upgrade_thread.is_alive(): return {"ok": False, "error": "Upgrade gia in corso"} directory = (payload.get("directory") or "").strip() archive_dir = (payload.get("archive_dir") or "").strip() or None archive_auto_pick = bool(payload.get("archive_auto_pick", False)) recursive = bool(payload.get("recursive", False)) try: threshold = int(payload.get("threshold", 310)) except (TypeError, ValueError): threshold = 310 if not directory: return {"ok": False, "error": "Cartella non impostata"} gate = self._gate("upgrade") if gate: return gate cfg = load_config() cookies_path = cfg.get("cookies_path", "") # Drain la queue di risoluzione da eventuali run precedenti (safety) try: while True: self._upgrade_resolve_q.get_nowait() except queue.Empty: pass self._upgrade_thread = threading.Thread( target=self._upgrade_worker, args=(directory, threshold, cookies_path, recursive, archive_dir, archive_auto_pick), daemon=True, ) self._upgrade_thread.start() return {"ok": True} def stop_upgrade(self) -> dict: request_upgrade_stop() # Sblocca eventuale resolve_callback in attesa: inviamo scelta 'skip' # cosi il worker esce pulito invece di restare bloccato in q.get(). try: self._upgrade_resolve_q.put_nowait({"action": "skip"}) except queue.Full: pass self._log("upgrade", "[INFO] Interruzione richiesta...") return {"ok": True} def read_audio_data_url(self, path: str) -> dict: """Legge un file audio locale e lo ritorna come data URL base64 per il preview HTML5 nel modal upgrade. WKWebView (macOS pywebview) blocca `file://` da HTML servito via file://, quindi passiamo dal bridge. Limite: 60MB per non esplodere la memoria JS.""" import base64 as _b64 MAX = 60 * 1024 * 1024 try: p = Path(path or "") if not p.exists() or not p.is_file(): return {"ok": False, "error": "File non trovato"} size = p.stat().st_size if size > MAX: return {"ok": False, "error": f"File troppo grande ({size // 1024 // 1024}MB, max 60MB)"} ext = p.suffix.lower().lstrip(".") mime_map = { "mp3": "audio/mpeg", "m4a": "audio/mp4", "aac": "audio/aac", "wav": "audio/wav", "flac": "audio/flac", "ogg": "audio/ogg", "opus": "audio/opus", "webm": "audio/webm", } mime = mime_map.get(ext, "audio/mpeg") b64 = _b64.b64encode(p.read_bytes()).decode("ascii") return {"ok": True, "data_url": f"data:{mime};base64,{b64}"} except Exception as e: return {"ok": False, "error": str(e)} def upgrade_resolve_candidates(self, choice: dict) -> dict: """Riceve la scelta utente dal modal 'match locali multipli' e la deposita nella queue attesa dal resolve_callback del worker. `choice`: {'action': 'use_local'|'use_youtube'|'skip', 'path': str?} """ if not isinstance(choice, dict): choice = {"action": "skip"} try: # Drain (nel caso rarissimo di doppio put) e poi metti la nuova try: self._upgrade_resolve_q.get_nowait() except queue.Empty: pass self._upgrade_resolve_q.put_nowait(choice) return {"ok": True} except Exception as e: return {"ok": False, "error": str(e)} def _upgrade_worker(self, directory, threshold, cookies_path, recursive, archive_dir: Optional[str] = None, archive_auto_pick: bool = False): view = "upgrade" def progress_cb(idx, total, filename, status, old_kbps, new_kbps): payload_evt = { "idx": idx, "total": total, "filename": filename, "status": status, "old_kbps": old_kbps, "new_kbps": new_kbps, } if total > 0: payload_evt["overall"] = min(idx / total, 1.0) if status == "searching": self._log(view, f"[CERCA] {filename} ({old_kbps}kbps)") elif status == "not_found": self._log(view, f"[NON TROVATO] {filename}") elif status == "cover_only": self._log(view, f"[COPERTINA] {filename} (gia {old_kbps}kbps)") elif status == "cover_done": self._log(view, f"[OK COPERTINA] {filename}") elif status == "cover_failed": self._log(view, f"[SKIP] {filename} - copertina non disponibile ({old_kbps}kbps ok)") elif status == "downloading": self._log(view, f"[DOWNLOAD] {filename} ({old_kbps}kbps)") elif status == "upgraded": diff = new_kbps - old_kbps diff_str = f" (+{diff}kbps)" if diff > 0 else "" self._log(view, f"[UPGRADE] {filename}: {old_kbps} -> {new_kbps}kbps{diff_str}") elif status == "local_copy": self._log(view, f"[LOCALE] {filename}: uso match dall'archivio") elif status == "local_upgraded": diff = new_kbps - old_kbps diff_str = f" (+{diff}kbps)" if diff > 0 else "" self._log(view, f"[LOCALE OK] {filename}: {old_kbps} -> {new_kbps}kbps{diff_str}") elif status == "skipped_by_user": self._log(view, f"[SKIP] {filename} (saltato manualmente)") elif status == "resolve_wait": self._log(view, f"[SCELTA] {filename}: match multipli, in attesa scelta utente") elif status == "scan_archive": # `new_kbps` in questo caso trasporta il count delle chiavi self._log(view, f"[ARCHIVIO] Indicizzati {new_kbps} gruppi audio") elif status == "download_error": self._log(view, f"[ERRORE] {filename}: download fallito") elif status == "stopped": self._log(view, "[INFO] Upgrade interrotto.") elif status == "completed": payload_evt["overall"] = 1.0 self._log(view, "[INFO] Upgrade completato!") self._emit("upgrade:progress", payload_evt) def resolve_cb(source_path: str, candidates: list) -> dict: """Bloccante: emette evento e attende scelta utente via queue. Se archive_auto_pick=True, salta il modal e sceglie il primo candidato (già ordinato per bitrate DESC nel core). `source_path` è il path assoluto del file sorgente (per il preview audio nel modal).""" src_name = Path(source_path).name if source_path else "" if archive_auto_pick and candidates: top = candidates[0] path = top.get("path") if isinstance(top, dict) else str(top) self._log(view, f"[AUTO] {src_name}: match multipli, uso {Path(path).name}") return {"action": "use_local", "path": path} self._emit("upgrade:candidates_needed", { "file": src_name, "source_path": source_path, "candidates": candidates, }) try: # Timeout 10 min — se l'utente sparisce, salta il brano choice = self._upgrade_resolve_q.get(timeout=600) except queue.Empty: choice = {"action": "skip"} return choice upgrade_folder( directory, threshold, cookies_path, recursive, progress_cb, archive_dir=archive_dir, resolve_callback=resolve_cb, ) self._emit("upgrade:done", {"ok": True}) # ------------------------------------------------------------------ # VIDEO (YouTube, TikTok, Instagram, Facebook) # ------------------------------------------------------------------ def start_video_download(self, payload: dict) -> dict: if self._any_job_running(): return {"ok": False, "error": "Un download gia in corso"} urls = payload.get("urls") or [] output_dir = (payload.get("output_dir") or "").strip() quality = payload.get("quality") or "1080p" audio_only = bool(payload.get("audio_only", False)) if not urls: return {"ok": False, "error": "Nessun URL fornito"} if not output_dir: return {"ok": False, "error": "Cartella output non impostata"} # Anche "audio_only" video count come video: usa codec/yt-dlp diversi gate = self._gate("video") if gate: return gate self._video_thread = threading.Thread( target=self._video_worker, args=(list(urls), output_dir, quality, audio_only), daemon=True, ) self._video_thread.start() return {"ok": True} def stop_video_download(self) -> dict: request_download_stop() self._log("video", "[INFO] Interruzione richiesta...") return {"ok": True} def _video_worker(self, urls: list, output_dir: str, quality: str, audio_only: bool = False) -> None: reset_download_stop() cfg = load_config() cookies_path = cfg.get("cookies_path", "") bitrate = cfg.get("bitrate", "320K") total_urls = len(urls) multi = total_urls > 1 view = "video" for url_idx, url in enumerate(urls): if download_is_stopped(): self._log(view, "[INFO] Download interrotto.") break if multi: self._log(view, f"\n{'='*50}") self._log(view, f"[LISTA] URL {url_idx + 1} di {total_urls}") self._log(view, f"{'='*50}") _last = [0.0] _THROTTLE = 0.10 url_base = url_idx / total_urls if multi else 0 url_weight = 1 / total_urls if multi else 1 def progress_cb(idx, total, track_name, status, pct, _base=url_base, _weight=url_weight, _uidx=url_idx): if status == "downloading": now = time.monotonic() if now - _last[0] < _THROTTLE: return _last[0] = now payload_evt = { "idx": idx, "total": total, "track": track_name, "status": status, "pct": pct, "url_idx": _uidx, "url_total": total_urls, } if status == "skipped": self._log(view, f"[SKIP] {track_name} (gia scaricato)") track_progress = (idx + 1) / total if total > 0 else 1 payload_evt["overall"] = min(_base + track_progress * _weight, 1.0) elif status == "downloading": track_progress = (idx / total) + (pct / 100 / total) if total > 0 else 0 payload_evt["overall"] = min(_base + track_progress * _weight, 1.0) elif status == "done": self._log(view, f"[OK] {track_name}") track_progress = (idx + 1) / total if total > 0 else 1 payload_evt["overall"] = min(_base + track_progress * _weight, 1.0) elif status == "stopped": self._log(view, "[INFO] Download interrotto.") elif status == "completed": if not multi: payload_evt["overall"] = 1.0 self._log(view, "[INFO] Download completato!") elif status.startswith("error"): self._log(view, f"[ERRORE] {track_name}: {status}") self._emit("video:progress", payload_evt) mode_lbl = f"Solo audio MP3 {bitrate}" if audio_only else f"Video {quality}" self._log(view, f"[INFO] {mode_lbl}: {url}") try: download_video(url, output_dir, quality, cookies_path, progress_cb, audio_only=audio_only, audio_bitrate=bitrate) except Exception as e: self._log(view, f"[ERRORE] {e}") continue if multi and not download_is_stopped(): self._log(view, f"\n[INFO] Completate tutte le {total_urls} URL!") self._emit("video:progress", { "status": "completed", "overall": 1.0, "url_total": total_urls, }) self._emit("video:done", {"ok": True}) # ================================================================ # Beatport charts # ================================================================ def beatport_genres(self) -> list: """Lista dei generi disponibili per il dropdown UI.""" return beatport.list_genres() def beatport_fetch_chart(self, slug: str, force_refresh: bool = False) -> dict: """Fetches la Top 100 per il genere. Salva anche l'ultimo genere in config. Ritorna {ok: True, tracks: [...]} oppure {ok: False, error, message}.""" # Memoria: ricorda l'ultimo genere usato try: cfg = load_config() cfg["beatport_last_genre"] = slug save_config(cfg) except Exception: pass try: tracks = beatport.fetch_top100(slug, force_refresh=force_refresh) except ValueError as e: return {"ok": False, "error": "invalid_genre", "message": str(e)} except beatport.BeatportUnreachableError as e: return {"ok": False, "error": "unreachable", "message": str(e)} except beatport.BeatportParseError as e: return {"ok": False, "error": "parse", "message": str(e)} return {"ok": True, "tracks": [asdict(t) for t in tracks]} def _beatport_output_dir(self, out_root: str, genre_name: str) -> Path: """Calcola la cartella dove finiscono i file Beatport. Coerente col comportamento di `start_tracks_download`, che sanitizza gli slash nel subfolder (`/` -> `_`), quindi `Beatport/` diventa flat: `Beatport_`.""" safe_genre = genre_name.replace("/", "_").replace("\\", "_").strip() subfolder = f"Beatport_{safe_genre}" if safe_genre else "Beatport" return Path(out_root) / subfolder def beatport_check_existing(self, tracks: list, genre_name: str) -> list: """Per ogni track ritorna True se il file esiste gia' nella cartella di output Beatport (stessa dove `beatport_download_selected` scrive). Match euristico: nel filename (senza extension) devono comparire sia il titolo che il primo artista (case-insensitive).""" cfg = load_config() out_root = (cfg.get("output_dir") or "").strip() if not out_root: return [False] * len(tracks) out_dir = self._beatport_output_dir(out_root, genre_name) 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() artists = (t.get("artists") or "") first_artist = artists.split(",")[0].split("&")[0].strip().lower() if not title or not first_artist: result.append(False) continue found = any( (title in stem and first_artist in stem) for stem in existing_stems ) result.append(found) return result def beatport_download_selected(self, tracks: list, genre_name: str) -> dict: """Converte i BeatportTrack in tracklist compatibile con start_tracks_download e lancia il download riusando l'infrastruttura esistente (license gate, skip file gia' scaricati, log su canale 'download', ecc.).""" cfg = load_config() out_root = (cfg.get("output_dir") or "").strip() if not out_root: return {"ok": False, "error": "Cartella output non impostata"} # Sfrutta lo stesso computed path del check_existing per garantire # che i due metodi restino allineati (senza affidarsi a start_tracks_download # per la sanitizzazione). target_dir = self._beatport_output_dir(out_root, genre_name) subfolder = target_dir.name # es. "Beatport_melodic-house-techno" converted = [] for t in tracks: title = (t.get("title") 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, }) # ================================================================ # Traxsource charts # ================================================================ def _traxsource_output_dir(self, out_root: str, genre_name: str) -> Path: """Cartella dove finiscono i file Traxsource. Coerente con la sanitizzazione di `start_tracks_download` (slash -> underscore).""" safe_genre = genre_name.replace("/", "_").replace("\\", "_").strip() subfolder = f"Traxsource_{safe_genre}" if safe_genre else "Traxsource" return Path(out_root) / subfolder def traxsource_genres(self) -> list: """Lista dei generi Traxsource per il dropdown UI.""" return traxsource.list_genres() def traxsource_fetch_chart(self, slug: str, force_refresh: bool = False) -> dict: """Fetches la Top 100 corrente per il genere. Salva l'ultimo genere in config.""" try: cfg = load_config() cfg["traxsource_last_genre"] = slug save_config(cfg) except Exception: pass try: tracks = traxsource.fetch_top100(slug, force_refresh=force_refresh) except ValueError as e: return {"ok": False, "error": "invalid_genre", "message": str(e)} except traxsource.TraxsourceUnreachableError as e: return {"ok": False, "error": "unreachable", "message": str(e)} except traxsource.TraxsourceParseError as e: return {"ok": False, "error": "parse", "message": str(e)} return {"ok": True, "tracks": [asdict(t) for t in tracks]} def traxsource_check_existing(self, tracks: list, genre_name: str) -> list: """True per ogni track già presente nella cartella output Traxsource. Match euristico: filename (senza extension) deve contenere sia il titolo che il primo artista (case-insensitive).""" cfg = load_config() out_root = (cfg.get("output_dir") or "").strip() if not out_root: return [False] * len(tracks) out_dir = self._traxsource_output_dir(out_root, genre_name) 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() artists = (t.get("artists") or "") first_artist = artists.split(",")[0].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 traxsource_download_selected(self, tracks: list, genre_name: str) -> dict: """Converte i TraxsourceTrack in tracklist compatibile con start_tracks_download + costruisce metadata paralleli per il tagger ID3 (album=label Traxsource, date=YYYY-MM corrente, genre=nome genere).""" 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._traxsource_output_dir(out_root, genre_name) subfolder = target.name # es. "Traxsource_Tech House" import datetime as _dt current_month = _dt.date.today().strftime("%Y-%m") converted: list = [] metadata: list = [] for t in tracks: title = (t.get("title") or "").strip() artists = (t.get("artists") or "").strip() if not title: continue converted.append({"name": title, "artist": artists}) metadata.append({ "title": title, "artist": artists, "album": (t.get("label") or "").strip() or f"Traxsource Top 100 {current_month}", "date": current_month, "genre": genre_name, "cover_url": (t.get("cover_url_large") or t.get("image_url") or "").strip(), }) if not converted: return {"ok": False, "error": "Nessun brano valido"} return self.start_tracks_download({ "tracks": converted, "output_dir": out_root, "subfolder": subfolder, "metadata": metadata, }) # ================================================================ # 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). Costruisce anche `metadata` (una entry per track) per il tagging ID3 + cover art post-download. Se le creds Spotify sono presenti, arricchisce con i generi dell'artista (cache locale per artist_id per evitare N+1 chiamate). """ 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 # Tentativo di token Spotify per fetch generi (best-effort, opzionale) cid = (cfg.get("client_id") or "").strip() secret = (cfg.get("client_secret") or "").strip() token: Optional[str] = None if cid and secret: try: token = spotify_client.get_access_token(cid, secret) except Exception: token = None genre_cache: dict = {} # artist_id -> [genres] converted: list = [] metadata: list = [] 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}) # Genere: primo genere dell'artista (se disponibile e token OK) genre = "" artist_id = (t.get("artist_id") or "").strip() if token and artist_id: if artist_id not in genre_cache: genre_cache[artist_id] = spotify_client.get_artist_genres(token, artist_id) genres = genre_cache.get(artist_id) or [] if genres: genre = genres[0] tn = t.get("track_number") or 0 metadata.append({ "title": title, "artist": artists, "album": (t.get("album") or "").strip(), "date": (t.get("release_date") or "").strip(), "genre": genre, "tracknumber": str(tn) if tn else "", "cover_url": (t.get("cover_url_large") or t.get("image_url") or "").strip(), }) if not converted: return {"ok": False, "error": "Nessun brano valido"} return self.start_tracks_download({ "tracks": converted, "output_dir": out_root, "subfolder": subfolder, "metadata": metadata, }) # ---- 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). Per ogni video prova a dedurre i metadati: 1) Se le creds Spotify sono presenti, cerca il titolo su Spotify e usa i tag Spotify se c'è un match (best-effort). 2) Altrimenti fallback: parsing "Artista - Titolo" dal titolo del video; uploader come fallback per l'album. """ 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 # Token Spotify opzionale per enrichment cid = (cfg.get("client_id") or "").strip() secret = (cfg.get("client_secret") or "").strip() token: Optional[str] = None if cid and secret: try: token = spotify_client.get_access_token(cid, secret) except Exception: token = None genre_cache: dict = {} # artist_id -> [genres] urls: list = [] titles: list = [] metadata: 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) md: dict = {} # 1) Enrichment via Spotify search (se token) if token: sp = spotify_client.enrich_from_youtube_title(token, title) if sp: # Genere: opzionale genre = "" artist_id = (sp.get("artist_id") or "").strip() if artist_id: if artist_id not in genre_cache: genre_cache[artist_id] = spotify_client.get_artist_genres(token, artist_id) genres = genre_cache.get(artist_id) or [] if genres: genre = genres[0] tn = sp.get("track_number") or 0 md = { "title": (sp.get("name") or "").strip(), "artist": (sp.get("artists") or "").strip(), "album": (sp.get("album") or "").strip(), "date": (sp.get("release_date") or "").strip(), "genre": genre, "tracknumber": str(tn) if tn else "", "cover_url": (sp.get("cover_url_large") or sp.get("image_url") or "").strip(), } # 2) Fallback: parsing del titolo YouTube if not md.get("title") or not md.get("artist"): parsed = _parse_track_line(title) if parsed: fb_artist = parsed["artist"] fb_title = parsed["name"] else: fb_artist = "" fb_title = title md = { "title": fb_title, "artist": fb_artist, "album": (t.get("channel") or "").strip(), "date": (t.get("release_date") or "").strip(), "genre": "", "tracknumber": "", "cover_url": (t.get("image_url") or "").strip(), } metadata.append(md) 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, "metadata": metadata, }) # ================================================================ # WAV -> MP3 converter # ================================================================ def convert_pick_wav_files(self) -> list: """Apre file picker multi-selezione per .wav. Ritorna list di path.""" if not self.window: return [] try: filetypes = ("WAV files (*.wav)",) result = self.window.create_file_dialog( self._open_dialog_type(), allow_multiple=True, file_types=filetypes, ) except Exception: return [] if not result: return [] return [str(p) for p in result if str(p).lower().endswith(".wav")] def convert_pick_wav_folder(self, recursive: bool = True) -> list: """Apre folder picker, scan .wav (opzionalmente ricorsivo).""" from core import converter if not self.window: return [] try: result = self.window.create_file_dialog(self._folder_dialog_type()) except Exception: return [] if not result: return [] folder = str(result[0]) if isinstance(result, (list, tuple)) else str(result) return converter.list_wav_files(folder, recursive=recursive) def convert_start(self, payload: dict) -> dict: """Avvia conversione batch. Payload: {files: [str], bitrate: int, vbr: bool, output_dir: str | None} Se output_dir e vuoto/None, il .mp3 va accanto al .wav. """ if self._any_job_running(): return {"ok": False, "error": "Un'operazione e' gia in corso"} files = payload.get("files") or [] if not files: return {"ok": False, "error": "Nessun file selezionato"} try: bitrate = int(payload.get("bitrate", 320)) except (TypeError, ValueError): bitrate = 320 if bitrate not in (128, 192, 256, 320): return {"ok": False, "error": "Bitrate non valido"} vbr = bool(payload.get("vbr", False)) output_dir = (payload.get("output_dir") or "").strip() or None gate = self._gate("audio") if gate: return gate self._download_thread = threading.Thread( target=self._convert_worker, args=(list(files), bitrate, vbr, output_dir), daemon=True, ) self._download_thread.start() return {"ok": True} def convert_stop(self) -> dict: from core import converter as conv conv.request_stop() self._log("convert", "[INFO] Interruzione richiesta...") return {"ok": True} def _convert_worker(self, files: list, bitrate: int, vbr: bool, output_dir: Optional[str]) -> None: from core import converter as conv conv.reset_stop() view = "convert" total = len(files) mode = "VBR" if vbr else "CBR" self._log(view, f"[INFO] Conversione di {total} file (MP3 {bitrate}k {mode})") if output_dir: self._log(view, f"[INFO] Destinazione: {output_dir}") else: self._log(view, "[INFO] Destinazione: accanto al file originale") _last = [0.0] def make_progress_cb(idx: int, filename: str): def cb(pct: int): import time as _t if pct == -1: # interruzione self._emit("convert:progress", { "idx": idx, "total": total, "file": filename, "status": "stopped", "pct": 0, }) return # throttle now = _t.monotonic() if pct < 100 and now - _last[0] < 0.10: return _last[0] = now self._emit("convert:progress", { "idx": idx, "total": total, "file": filename, "status": "converting" if pct < 100 else "done", "pct": pct, "overall": min(((idx + pct / 100) / total), 1.0) if total else 0, }) return cb for i, src_path in enumerate(files): if conv.is_stopped(): self._log(view, "[INFO] Conversione interrotta.") break src = Path(src_path) if not src.exists(): self._log(view, f"[ERRORE] File mancante: {src.name}") continue dst_dir = Path(output_dir) if output_dir else src.parent dst = dst_dir / (src.stem + ".mp3") # Skip se il .mp3 esiste gia (protezione contro overwrite) if dst.exists(): self._log(view, f"[SKIP] {dst.name} esiste gia'") self._emit("convert:progress", { "idx": i, "total": total, "file": src.name, "status": "skipped", "pct": 100, "overall": min(((i + 1) / total), 1.0) if total else 1, }) continue self._log(view, f"[CONVERTI] {src.name}") self._emit("convert:progress", { "idx": i, "total": total, "file": src.name, "status": "converting", "pct": 0, }) try: conv.convert_wav_to_mp3( str(src), str(dst), bitrate=bitrate, vbr=vbr, progress_callback=make_progress_cb(i, src.name), ) self._log(view, f"[OK] {dst.name}") except Exception as e: self._log(view, f"[ERRORE] {src.name}: {e}") self._emit("convert:progress", { "idx": i, "total": total, "file": src.name, "status": f"error: {e}", "pct": 0, }) self._emit("convert:done", {"ok": True}) # ================================================================ # DEDUP — audio duplicati via Chromaprint fingerprinting # ================================================================ def dedup_pick_folder(self) -> str: """Folder picker per la cartella da scansionare.""" return self.browse_directory() def dedup_start_scan(self, payload: dict) -> dict: """Avvia worker di scansione. Payload: {directory, recursive}. Salva `dedup_last_folder` e `dedup_recursive` in config per la prossima apertura del tab. """ if self._dedup_thread and self._dedup_thread.is_alive(): return {"ok": False, "error": "Scansione dedup gia in corso"} directory = (payload.get("directory") or "").strip() recursive = bool(payload.get("recursive", True)) method = (payload.get("method") or "fingerprint").strip() if method not in ("fingerprint", "filename"): method = "fingerprint" if not directory: return {"ok": False, "error": "Cartella non impostata"} if not os.path.isdir(directory): return {"ok": False, "error": "Cartella non trovata"} # Persist last folder/recursive/method try: cfg = load_config() cfg["dedup_last_folder"] = directory cfg["dedup_recursive"] = recursive cfg["dedup_method"] = method save_config(cfg) except Exception: pass self._dedup_thread = threading.Thread( target=self._dedup_worker, args=(directory, recursive, method), daemon=True, ) self._dedup_thread.start() return {"ok": True} def dedup_stop_scan(self) -> dict: dedup.request_stop() self._log("dedup", "[INFO] Interruzione richiesta...") return {"ok": True} def dedup_move_to_trash(self, paths: list) -> dict: """Sposta i file in cestino via send2trash. Non consuma quota.""" if not isinstance(paths, list): return {"ok": False, "error": "paths deve essere una lista"} # Sanitize: solo str non vuote clean = [str(p).strip() for p in paths if p and str(p).strip()] if not clean: return {"ok": False, "error": "Nessun file da cancellare"} result = dedup.move_to_trash(clean) moved_n = len(result.get("moved", [])) failed_n = len(result.get("failed", [])) if moved_n: self._log("dedup", f"[OK] {moved_n} file spostati nel cestino") for f in result.get("failed", []): self._log("dedup", f"[ERRORE] {f.get('path')}: {f.get('error')}") return {"ok": True, "moved": result.get("moved", []), "failed": result.get("failed", []), "moved_count": moved_n, "failed_count": failed_n} def _dedup_worker(self, directory: str, recursive: bool, method: str = "fingerprint") -> None: """Esegue la scansione in background e emette progress/done.""" view = "dedup" dedup.reset_stop() method_label = "audio fingerprint" if method == "fingerprint" else "nome file" self._log(view, f"[INFO] Scansione: {directory} (metodo: {method_label}, recursive={recursive})") _last = [0.0] _THROTTLE = 0.05 def progress_cb(idx: int, total: int, filename: str, status: str, err_msg: str = "") -> None: # Throttle solo eventi 'computing'/'cached' (che possono essere migliaia) if status in ("computing", "cached"): now = time.monotonic() if now - _last[0] < _THROTTLE and idx != total: return _last[0] = now payload_evt = { "idx": idx, "total": total, "filename": filename, "status": status, } if err_msg: payload_evt["error_msg"] = err_msg if total > 0: payload_evt["overall"] = min(idx / total, 1.0) if status == "completed": payload_evt["overall"] = 1.0 self._log(view, f"[INFO] Scansione completata ({total} file).") elif status == "stopped": self._log(view, "[INFO] Scansione interrotta.") elif status == "error" and filename: detail = f": {err_msg}" if err_msg else "" self._log(view, f"[ERRORE] {filename}{detail}") self._emit("dedup:progress", payload_evt) def group_cb(group: dict) -> None: """Emit streaming: appena un gruppo raggiunge/aggiorna >=2 file.""" try: self._emit("dedup:group", group) except Exception: pass try: groups = dedup.scan_folder(directory, recursive=recursive, progress_callback=progress_cb, method=method, group_callback=group_cb) except Exception as e: self._log(view, f"[ERRORE] {e}") self._emit("dedup:done", {"ok": False, "error": str(e), "groups": []}) return n_groups = len(groups) n_dupes = sum(max(0, len(g) - 1) for g in groups) total_bytes = sum(sum(int(e.get("size") or 0) for e in g[1:]) for g in groups) self._log(view, f"[INFO] Gruppi: {n_groups} — duplicati: {n_dupes} — " f"spazio recuperabile: ~{total_bytes // 1024 // 1024} MB") self._emit("dedup:done", { "ok": True, "groups": groups, "n_groups": n_groups, "n_dupes": n_dupes, "reclaimable_bytes": total_bytes, }) # ================================================================ # CATALOGA — sposta file audio in /// # via Chromaprint + AcoustID. # ================================================================ def catalog_pick_source_folder(self) -> str: """Folder picker per la cartella sorgente da catalogare.""" return self.browse_directory() def catalog_pick_target_folder(self) -> str: """Folder picker per la cartella target (dove finiscono i file organizzati). Se l'utente lascia vuoto → si usa la source.""" return self.browse_directory() def catalog_start_scan(self, payload: dict) -> dict: """Avvia worker di scansione + lookup AcoustID. Payload: {source, target?, recursive}. Persiste last source/target/recursive. """ if self._catalog_thread and self._catalog_thread.is_alive(): return {"ok": False, "error": "Scansione catalogo gia in corso"} source = (payload.get("source") or "").strip() target = (payload.get("target") or "").strip() recursive = bool(payload.get("recursive", True)) if not source: return {"ok": False, "error": "Cartella input non impostata"} if not os.path.isdir(source): return {"ok": False, "error": "Cartella input non trovata"} # target vuoto → sposta in-place nella source if not target: target = source # Persist last source/target/recursive try: cfg = load_config() cfg["catalog_last_source"] = source cfg["catalog_last_target"] = target cfg["catalog_recursive"] = recursive save_config(cfg) except Exception: pass self._catalog_thread = threading.Thread( target=self._catalog_worker, args=(source, recursive), daemon=True, ) self._catalog_thread.start() return {"ok": True} def catalog_stop_scan(self) -> dict: catalog.request_stop() self._log("catalog", "[INFO] Interruzione richiesta...") return {"ok": True} def catalog_move_files(self, entries: list, target_root: str, layout: str = "year_genre") -> dict: """Sposta le entry selezionate in ///. Non consuma quota (e' una riorganizzazione di file locali, non un download). Ritorna il summary con moved/failed/operations. """ if not isinstance(entries, list): return {"ok": False, "error": "entries deve essere una lista"} target = (target_root or "").strip() if not target: return {"ok": False, "error": "Cartella destinazione mancante"} def log_cb(op: dict) -> None: src = op.get("src", "") dst = op.get("dst", "") self._log("catalog", f"[MOVE] {src} → {dst}") try: self._emit("catalog:log", op) except Exception: pass try: safe_layout = layout if layout in ("year_genre", "genre_year") else "year_genre" result = catalog.move_files(entries, target, log_callback=log_cb, layout=safe_layout) except Exception as e: self._log("catalog", f"[ERRORE] move: {e}") return {"ok": False, "error": str(e)} moved_n = result.get("moved", 0) failed = result.get("failed", []) or [] if moved_n: self._log("catalog", f"[OK] {moved_n} file organizzati") for f in failed: self._log("catalog", f"[ERRORE] {f.get('path')}: {f.get('error')}") return { "ok": True, "moved": moved_n, "failed_count": len(failed), "failed": failed, "operations": result.get("operations", []), } # ------------------------------------------------------------------ # FLATTEN — sposta tutti i file audio delle sotto-cartelle nella root # ------------------------------------------------------------------ def flatten_pick_folder(self) -> str: if not self.window: return "" try: r = self.window.create_file_dialog(self._folder_dialog_type()) except Exception: return "" if not r: return "" return str(r[0]) if isinstance(r, (list, tuple)) else str(r) def flatten_start(self, payload: dict) -> dict: """Payload: {directory, remove_empty}.""" if self._flatten_thread and self._flatten_thread.is_alive(): return {"ok": False, "error": "Operazione gia' in corso"} directory = (payload.get("directory") or "").strip() if not directory: return {"ok": False, "error": "Cartella non impostata"} if not os.path.isdir(directory): return {"ok": False, "error": "Cartella non trovata"} remove_empty = bool(payload.get("remove_empty", True)) self._flatten_thread = threading.Thread( target=self._flatten_worker, args=(directory, remove_empty), daemon=True, ) self._flatten_thread.start() return {"ok": True} def flatten_stop(self) -> dict: flatten.request_stop() return {"ok": True} def _flatten_worker(self, directory: str, remove_empty: bool) -> None: view = "flatten" self._log(view, f"[INFO] Appiattisco: {directory} (remove_empty={remove_empty})") def progress_cb(idx, total, name, status, err=""): payload = {"idx": idx, "total": total, "filename": name, "status": status} if err: payload["error_msg"] = err if total > 0: payload["overall"] = min(idx / total, 1.0) if status == "completed": payload["overall"] = 1.0 self._log(view, f"[INFO] Completato: {total} file processati.") elif status == "stopped": self._log(view, "[INFO] Operazione interrotta.") elif status == "error" and name: self._log(view, f"[ERRORE] {name}: {err}") self._emit("flatten:progress", payload) def log_cb(op): self._log(view, f"[MOVE] {op.get('src','')} → {op.get('dst','')}") try: result = flatten.flatten_folder( directory, remove_empty=remove_empty, progress_callback=progress_cb, log_callback=log_cb, ) except Exception as e: self._log(view, f"[ERRORE] {e}") self._emit("flatten:done", {"ok": False, "error": str(e)}) return moved = result.get("moved", 0) failed = result.get("failed", []) or [] dirs_removed = result.get("dirs_removed", 0) self._log(view, f"[OK] Spostati {moved} file. Cartelle rimosse: {dirs_removed}. Falliti: {len(failed)}.") self._emit("flatten:done", { "ok": True, "moved": moved, "failed_count": len(failed), "failed": failed, "dirs_removed": dirs_removed, }) def _catalog_worker(self, directory: str, recursive: bool) -> None: """Esegue scan + lookup AcoustID in background, emette streaming.""" view = "catalog" catalog.reset_stop() self._log(view, f"[INFO] Scansione: {directory} (recursive={recursive})") _last = [0.0] _THROTTLE = 0.05 def progress_cb(idx: int, total: int, filename: str, status: str, err_msg: str = "") -> None: # Throttle solo eventi "computing"/"lookup" (possono essere tanti) if status in ("computing", "lookup"): now = time.monotonic() if now - _last[0] < _THROTTLE and idx != total: return _last[0] = now payload_evt = { "idx": idx, "total": total, "filename": filename, "status": status, } if err_msg: payload_evt["error_msg"] = err_msg if total > 0: payload_evt["overall"] = min(idx / total, 1.0) if status == "completed": payload_evt["overall"] = 1.0 self._log(view, f"[INFO] Scansione completata ({total} file).") elif status == "stopped": self._log(view, "[INFO] Scansione interrotta.") elif status == "error" and filename: detail = f": {err_msg}" if err_msg else "" self._log(view, f"[ERRORE] {filename}{detail}") self._emit("catalog:progress", payload_evt) def entry_cb(entry: dict) -> None: """Streaming: appena un file è processato, emit alla UI.""" try: self._emit("catalog:entry", entry) except Exception: pass # Prova a ottenere un token Spotify per il fallback (facoltativo). # Se non ci sono creds o l'auth fallisce, procediamo senza fallback. spotify_token = None try: cfg = load_config() cid = (cfg.get("client_id") or "").strip() csecret = (cfg.get("client_secret") or "").strip() if cid and csecret: spotify_token = get_access_token(cid, csecret) except Exception: spotify_token = None try: entries = catalog.scan_folder( directory, recursive=recursive, progress_callback=progress_cb, entry_callback=entry_cb, spotify_token=spotify_token, ) except Exception as e: self._log(view, f"[ERRORE] {e}") self._emit("catalog:done", {"ok": False, "error": str(e), "entries": []}) return n_matched = sum(1 for e in entries if e.get("matched")) n_unmatched = len(entries) - n_matched self._log(view, f"[INFO] Match AcoustID: {n_matched} · non identificati: " f"{n_unmatched} (totale {len(entries)})") self._emit("catalog:done", { "ok": True, "entries": entries, "n_total": len(entries), "n_matched": n_matched, "n_unmatched": n_unmatched, }) # ================================================================ # CHARTS — classifiche musicali (iTunes / Spotify / M2O / Last.fm) # ================================================================ def charts_fetch(self, payload: dict) -> dict: """Ritorna le entry di una chart per la sorgente/params richiesti. Payload: { source: "itunes" | "spotify" | "m2o" | "lastfm", country?: "it"|"us"|"gb"|"ww", # solo itunes playlist_key?: str, # solo spotify (chiave _SPOTIFY_CHARTS) decade?: str, # solo lastfm ("" | "70s"|"80s"|...) genre?: str, # solo lastfm ("" | pop|rock|dance|...) force?: bool, # bypass cache } Persiste anche la scelta corrente in config. """ payload = payload or {} source = (payload.get("source") or "itunes").strip().lower() force = bool(payload.get("force", False)) # Persist scelta corrente (best-effort, non blocca la fetch) try: cfg = load_config() cfg["charts_last_source"] = source if source == "itunes": cfg["charts_itunes_country"] = ( payload.get("country") or cfg.get("charts_itunes_country") or "it" ) elif source == "lastfm": cfg["charts_lastfm_decade"] = ( payload.get("decade") if payload.get("decade") is not None else cfg.get("charts_lastfm_decade", "90s") ) cfg["charts_lastfm_genre"] = ( payload.get("genre") if payload.get("genre") is not None else cfg.get("charts_lastfm_genre", "dance") ) elif source == "spotify": cfg["charts_spotify_playlist"] = ( payload.get("playlist_key") or cfg.get("charts_spotify_playlist") or "top50_global" ) save_config(cfg) except Exception: pass try: if source == "itunes": country = (payload.get("country") or "it").strip().lower() or "it" entries = charts.fetch_itunes(country, force=force) elif source == "lastfm": decade = payload.get("decade") or "" genre = payload.get("genre") or "" entries = charts.fetch_lastfm(decade, genre, force=force) elif source == "spotify": playlist_key = (payload.get("playlist_key") or "top50_global").strip() entries = charts.fetch_spotify(playlist_key, force=force) elif source == "m2o": entries = charts.fetch_m2o(force=force) else: return {"ok": False, "error": f"Sorgente sconosciuta: {source}"} except Exception as e: self._log("charts", f"[ERRORE] fetch {source}: {e}") return {"ok": False, "error": str(e)} self._log("charts", f"[INFO] {source}: {len(entries)} brani ricevuti") return {"ok": True, "source": source, "entries": entries} def charts_add_to_queue(self, entries: list) -> dict: """Riceve una lista di entry `{artist, title}` selezionate dalla UI e le converte in una tracklist compatibile con la tab Scarica. Emette l'evento `charts:add_to_queue` col testo formattato (`Artista - Titolo` una per riga) e la lista strutturata. La UI popola l'input della tab Scarica e commuta view. """ if not isinstance(entries, list): return {"ok": False, "error": "entries deve essere una lista"} tracks = [] lines = [] for e in entries: if not isinstance(e, dict): continue artist = (e.get("artist") or "").strip() title = (e.get("title") or "").strip() if not title: continue tracks.append({"artist": artist, "name": title}) if artist: lines.append(f"{artist} - {title}") else: lines.append(title) if not tracks: return {"ok": False, "error": "Nessun brano valido selezionato"} text = "\n".join(lines) payload = {"text": text, "count": len(tracks), "tracks": tracks} self._emit("charts:add_to_queue", payload) self._log("charts", f"[INFO] {len(tracks)} brani inviati alla coda Scarica") return {"ok": True, "count": len(tracks)}