luzadevandClaude Opus 4.7
f339197861
Sul sistema dell'utente il comando ffmpeg che usa l'app funziona correttamente (testato come subprocess identico), ma il vecchio reporting era confuso quando l'errore era transitorio (BlackHole non pronto, clock drift, ecc.) o quando la parola 'abort' non compariva nello stderr. Cambiamenti: core/recorder.py: - _extract_friendly_error riconosce ora piu casi: input/output error (BlackHole non pronto -> suggerisce 'sudo killall coreaudiod'), device not configured, permission denied, no such device - rc=-6 su macOS ritorna SEMPRE il messaggio sul permesso Microfono (prima richiedeva 'abort' nello stderr, non sempre presente) - Fallback finale mostra anche le ultime 3 righe di stderr - Nuova funzione get_last_stderr() per debugging api/bridge.py: - Nuovo metodo get_recorder_log() che ritorna stderr per la UI webui: - Nuova card di errore persistente con bottoni 'Mostra log tecnici' / 'Copia log' - Si resetta automaticamente all'avvio di una nuova registrazione Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
794 lines
31 KiB
Python
794 lines
31 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
|
|
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 = """\
|
|
1) Vai su https://developer.spotify.com/dashboard e accedi (anche con account free).
|
|
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".
|
|
|
|
Note: credenziali gratuite, no Premium. Limite 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,
|
|
"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}
|
|
|
|
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"}
|
|
|
|
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"
|
|
|
|
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"}
|
|
|
|
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)
|
|
|
|
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
|
|
has_spotify = any("spotify.com" in u for u in urls)
|
|
if has_spotify:
|
|
client_id = cfg.get("client_id", "")
|
|
client_secret = cfg.get("client_secret", "")
|
|
if not client_id or not client_secret:
|
|
self._log(view, "[ERRORE] Configura Client ID e Client Secret nelle Impostazioni.")
|
|
self._emit("download:done", {"ok": False})
|
|
return
|
|
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
|
|
|
|
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:
|
|
label, name, tracks = resolve_spotify_url(token, url)
|
|
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)
|
|
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"}
|
|
|
|
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"}
|
|
|
|
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})
|