1455 lines
57 KiB
Python
1455 lines
57 KiB
Python
"""Ponte JS <-> Python: tutti i metodi che la UI invoca via pywebview.api."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
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
|
|
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,
|
|
)
|
|
|
|
|
|
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
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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:
|
|
self._emit("log", {"view": view, "msg": msg})
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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(),
|
|
"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()
|
|
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", "")
|
|
|
|
self._upgrade_thread = threading.Thread(
|
|
target=self._upgrade_worker,
|
|
args=(directory, threshold, cookies_path, recursive),
|
|
daemon=True,
|
|
)
|
|
self._upgrade_thread.start()
|
|
return {"ok": True}
|
|
|
|
def stop_upgrade(self) -> dict:
|
|
request_upgrade_stop()
|
|
self._log("upgrade", "[INFO] Interruzione richiesta...")
|
|
return {"ok": True}
|
|
|
|
def _upgrade_worker(self, directory, threshold, cookies_path, recursive):
|
|
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 == "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)
|
|
|
|
upgrade_folder(directory, threshold, cookies_path, recursive, progress_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/<genre>`
|
|
diventa flat: `Beatport_<genre>`."""
|
|
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,
|
|
})
|
|
|
|
# ================================================================
|
|
# 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,
|
|
})
|