luzadev
6e96e8c6b7
Impostazioni: il bottone "Salva impostazioni" e' ora sticky in fondo al viewport, sempre visibile senza dover scrollare in coda alla pagina. Spotify: aggiornata guida in-app e landing per riflettere il nuovo requisito Premium per creare le API keys (fino a oggi era gratis con account Spotify Free; Spotify ha cambiato policy). Singoli brani e tutte le altre fonti continuano a funzionare senza Premium. Inclusi anche file gia deployati lato server in sessioni precedenti (webhook LS, landing icone PNG, download.html Gatekeeper instructions).
901 lines
35 KiB
Python
901 lines
35 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 typing import Any, Optional
|
|
|
|
from core.config import load_config, save_config, VERSION, LICENSE_API_URL
|
|
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.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
|
|
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."""
|
|
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()
|
|
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),
|
|
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) -> 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}")
|
|
|
|
_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)
|
|
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})
|