Initial commit: MusicDownload con UI pywebview
- Frontend HTML/CSS/JS in webui/ (stile Spotify-like, hero gradient, modal blur) - API bridge JS<->Python in api/bridge.py - Core invariato: downloader yt-dlp, upgrader, spotify client, config - Build scripts macOS (WebKit) e Windows (EdgeChromium) - .gitignore esclude credenziali (config.json, cookies.txt) e artefatti build Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
luzadevandClaude Opus 4.7 committed
commit
8c06a26ddb
25 files changed
+4316
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,64 @@
|
||||
"""Gestione configurazione persistente (JSON)."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
VERSION = "v1.0"
|
||||
|
||||
|
||||
def _get_config_dir() -> Path:
|
||||
"""Ritorna la directory per config.json.
|
||||
|
||||
Frozen (bundle PyInstaller):
|
||||
- macOS: ~/Library/Application Support/MusicDownload/
|
||||
- Windows: %APPDATA%/MusicDownload/
|
||||
Dev (non frozen):
|
||||
- directory del progetto
|
||||
"""
|
||||
if getattr(sys, "frozen", False):
|
||||
if sys.platform == "win32":
|
||||
base = Path(os.environ.get("APPDATA", Path.home()))
|
||||
else:
|
||||
base = Path.home() / "Library" / "Application Support"
|
||||
config_dir = base / "MusicDownload"
|
||||
config_dir.mkdir(parents=True, exist_ok=True)
|
||||
return config_dir
|
||||
return Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
CONFIG_PATH = _get_config_dir() / "config.json"
|
||||
|
||||
# Default: i path relativi al progetto (se non frozen)
|
||||
_project_dir = Path(__file__).resolve().parent.parent
|
||||
|
||||
DEFAULTS = {
|
||||
"client_id": "",
|
||||
"client_secret": "",
|
||||
"bitrate": "320K",
|
||||
"hq_threshold": 310,
|
||||
"cookies_path": str(_project_dir / "cookies.txt"),
|
||||
"output_dir": str(_project_dir / "MUSICA"),
|
||||
"theme": "dark",
|
||||
}
|
||||
|
||||
|
||||
def load_config() -> dict:
|
||||
"""Carica la configurazione da config.json, con fallback ai default."""
|
||||
config = dict(DEFAULTS)
|
||||
if CONFIG_PATH.exists():
|
||||
try:
|
||||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||||
stored = json.load(f)
|
||||
config.update(stored)
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
return config
|
||||
|
||||
|
||||
def save_config(config: dict) -> None:
|
||||
"""Salva la configurazione su config.json."""
|
||||
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
|
||||
json.dump(config, f, indent=2, ensure_ascii=False)
|
||||
@@ -0,0 +1,395 @@
|
||||
"""Download da YouTube via yt-dlp (subprocess per usare la versione piu aggiornata)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
from core.paths import find_ytdlp, find_ffmpeg_dir
|
||||
|
||||
|
||||
# Flag globale per interruzione
|
||||
_stop_event = threading.Event()
|
||||
_current_process: Optional[subprocess.Popen] = None
|
||||
_process_lock = threading.Lock()
|
||||
|
||||
|
||||
def request_stop():
|
||||
"""Richiede l'interruzione del download in corso."""
|
||||
_stop_event.set()
|
||||
with _process_lock:
|
||||
if _current_process and _current_process.poll() is None:
|
||||
_current_process.terminate()
|
||||
|
||||
|
||||
def reset_stop():
|
||||
_stop_event.clear()
|
||||
|
||||
|
||||
def is_stopped() -> bool:
|
||||
return _stop_event.is_set()
|
||||
|
||||
|
||||
def _load_done_set(done_file: Path) -> set[str]:
|
||||
"""Carica l'elenco delle tracce gia scaricate."""
|
||||
if done_file.exists():
|
||||
try:
|
||||
return set(done_file.read_text(encoding="utf-8").splitlines())
|
||||
except OSError:
|
||||
return set()
|
||||
return set()
|
||||
|
||||
|
||||
def _mark_done(done_file: Path, key: str) -> None:
|
||||
"""Aggiunge una traccia al file delle tracce scaricate."""
|
||||
try:
|
||||
with open(done_file, "a", encoding="utf-8") as f:
|
||||
f.write(key + "\n")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
_STOPWORDS = {
|
||||
"official", "video", "audio", "music", "lyric", "lyrics", "hd", "4k",
|
||||
"remix", "edit", "extended", "feat", "ft", "featuring", "with",
|
||||
"from", "the", "and", "of", "a", "an", "to", "in",
|
||||
}
|
||||
|
||||
|
||||
def _tokenize(s: str) -> set[str]:
|
||||
"""Estrae token alfanumerici significativi da una stringa."""
|
||||
s = s.lower()
|
||||
tokens = re.findall(r"[a-z0-9]+", s)
|
||||
return {t for t in tokens if len(t) >= 2 and t not in _STOPWORDS}
|
||||
|
||||
|
||||
def _scan_existing_files(directory: Path, extensions: tuple[str, ...] = (".mp3", ".m4a", ".opus", ".webm", ".wav", ".flac")) -> list[set[str]]:
|
||||
"""Ritorna lista di set di token dei filename presenti nella directory."""
|
||||
if not directory.exists():
|
||||
return []
|
||||
return [_tokenize(f.stem) for f in directory.iterdir()
|
||||
if f.is_file() and f.suffix.lower() in extensions]
|
||||
|
||||
|
||||
def _file_exists_for_query(query: str, existing_tokens: list[set[str]]) -> bool:
|
||||
"""Verifica se esiste gia un file il cui titolo contiene tutte le parole del query.
|
||||
Match basato sui token: tutte le parole significative del query devono essere
|
||||
presenti nel filename (l'ordine non conta, eventuali parole extra nel filename ok)."""
|
||||
query_tokens = _tokenize(query)
|
||||
if not query_tokens:
|
||||
return False
|
||||
for file_tokens in existing_tokens:
|
||||
if query_tokens.issubset(file_tokens):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _search_youtube(query: str, cookies_path: Optional[str] = None) -> tuple[str, str]:
|
||||
"""Cerca un video su YouTube. Ritorna (video_url, video_title)."""
|
||||
ytdlp = find_ytdlp()
|
||||
cmd = [
|
||||
ytdlp,
|
||||
f"ytsearch1:{query}",
|
||||
"--dump-json",
|
||||
"--no-download",
|
||||
"--no-warnings",
|
||||
"--flat-playlist",
|
||||
]
|
||||
if cookies_path and Path(cookies_path).exists():
|
||||
cmd.extend(["--cookies", cookies_path])
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||
if result.returncode != 0:
|
||||
raise ValueError(f"Ricerca fallita: {result.stderr.strip()}")
|
||||
|
||||
data = json.loads(result.stdout)
|
||||
video_id = data.get("id", "")
|
||||
title = data.get("title", query)
|
||||
if not video_id:
|
||||
raise ValueError("Nessun risultato trovato")
|
||||
|
||||
return f"https://www.youtube.com/watch?v={video_id}", title
|
||||
|
||||
|
||||
def download_playlist(
|
||||
tracks: list[dict],
|
||||
output_dir: str,
|
||||
bitrate: str = "320K",
|
||||
cookies_path: Optional[str] = None,
|
||||
progress_callback: Optional[Callable] = None,
|
||||
) -> None:
|
||||
"""Scarica tutti i brani dalla lista di tracce.
|
||||
|
||||
Args:
|
||||
tracks: lista di {"name": ..., "artist": ...}
|
||||
output_dir: directory di destinazione
|
||||
bitrate: qualita audio (es. "320K")
|
||||
cookies_path: percorso file cookies (opzionale)
|
||||
progress_callback: callback(track_index, total, track_name, status, percent)
|
||||
"""
|
||||
reset_stop()
|
||||
total = len(tracks)
|
||||
out_path = Path(output_dir)
|
||||
out_path.mkdir(parents=True, exist_ok=True)
|
||||
done_file = out_path / ".downloaded_tracks"
|
||||
done_set = _load_done_set(done_file)
|
||||
existing_files = _scan_existing_files(out_path)
|
||||
ytdlp = find_ytdlp()
|
||||
|
||||
for i, track in enumerate(tracks):
|
||||
if is_stopped():
|
||||
if progress_callback:
|
||||
progress_callback(i, total, "", "stopped", 0)
|
||||
return
|
||||
|
||||
query = f"{track['name']} - {track['artist']}"
|
||||
track_only = track['name']
|
||||
|
||||
# Salta brani gia scaricati in precedenza (file di tracking)
|
||||
if query in done_set:
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, "skipped", 100)
|
||||
continue
|
||||
|
||||
# Salta se esiste gia un file corrispondente nella cartella
|
||||
if (_file_exists_for_query(query, existing_files)
|
||||
or _file_exists_for_query(track_only, existing_files)):
|
||||
_mark_done(done_file, query)
|
||||
done_set.add(query)
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, "skipped", 100)
|
||||
continue
|
||||
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, "searching", 0)
|
||||
|
||||
# Cerca il video su YouTube
|
||||
try:
|
||||
video_url, _ = _search_youtube(query, cookies_path)
|
||||
except Exception as e:
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, f"error: {e}", 0)
|
||||
continue
|
||||
|
||||
if is_stopped():
|
||||
return
|
||||
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, "downloading", 0)
|
||||
|
||||
# Scarica con yt-dlp subprocess
|
||||
cmd = [
|
||||
ytdlp,
|
||||
"--extract-audio",
|
||||
"--audio-format", "mp3",
|
||||
"--audio-quality", bitrate.rstrip("Kk"),
|
||||
"--embed-thumbnail",
|
||||
"--add-metadata",
|
||||
"--no-check-certificates",
|
||||
"--newline",
|
||||
"--output", str(Path(output_dir) / "%(title)s.%(ext)s"),
|
||||
video_url,
|
||||
]
|
||||
ffmpeg_dir = find_ffmpeg_dir()
|
||||
if ffmpeg_dir:
|
||||
cmd.extend(["--ffmpeg-location", ffmpeg_dir])
|
||||
if cookies_path and Path(cookies_path).exists():
|
||||
cmd.extend(["--cookies", cookies_path])
|
||||
|
||||
try:
|
||||
with _process_lock:
|
||||
_current_process = subprocess.Popen(
|
||||
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
|
||||
)
|
||||
|
||||
for line in _current_process.stdout:
|
||||
if is_stopped():
|
||||
_current_process.terminate()
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, "stopped", 0)
|
||||
return
|
||||
|
||||
# Parse progress da output yt-dlp
|
||||
pct_match = re.search(r"(\d+(?:\.\d+)?)%", line)
|
||||
if pct_match and progress_callback:
|
||||
pct = int(float(pct_match.group(1)))
|
||||
progress_callback(i, total, query, "downloading", pct)
|
||||
|
||||
return_code = _current_process.wait()
|
||||
|
||||
with _process_lock:
|
||||
_current_process = None
|
||||
|
||||
if return_code == 0:
|
||||
_mark_done(done_file, query)
|
||||
done_set.add(query)
|
||||
existing_files = _scan_existing_files(out_path)
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, "done", 100)
|
||||
else:
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, f"error: yt-dlp exit {return_code}", 0)
|
||||
|
||||
except Exception as e:
|
||||
with _process_lock:
|
||||
_current_process = None
|
||||
if progress_callback:
|
||||
progress_callback(i, total, query, f"error: {e}", 0)
|
||||
|
||||
if progress_callback and not is_stopped():
|
||||
progress_callback(total, total, "", "completed", 100)
|
||||
|
||||
|
||||
def download_direct_url(
|
||||
url: str,
|
||||
output_dir: str,
|
||||
bitrate: str = "320K",
|
||||
cookies_path: Optional[str] = None,
|
||||
progress_callback: Optional[Callable] = None,
|
||||
) -> None:
|
||||
"""Scarica direttamente da un URL supportato da yt-dlp (YouTube, SoundCloud, ecc.).
|
||||
|
||||
Supporta sia singoli video che playlist.
|
||||
Callback signature: progress_callback(idx, total, track_name, status, pct)
|
||||
"""
|
||||
global _current_process
|
||||
reset_stop()
|
||||
out_path = Path(output_dir)
|
||||
out_path.mkdir(parents=True, exist_ok=True)
|
||||
done_file = out_path / ".downloaded_tracks"
|
||||
done_set = _load_done_set(done_file)
|
||||
existing_files = _scan_existing_files(out_path)
|
||||
ytdlp = find_ytdlp()
|
||||
|
||||
# 1. Ottieni info tramite yt-dlp --dump-json
|
||||
probe_cmd = [
|
||||
ytdlp,
|
||||
"--dump-json",
|
||||
"--flat-playlist",
|
||||
"--no-download",
|
||||
"--no-warnings",
|
||||
url,
|
||||
]
|
||||
if cookies_path and Path(cookies_path).exists():
|
||||
probe_cmd.extend(["--cookies", cookies_path])
|
||||
|
||||
try:
|
||||
result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=60)
|
||||
if result.returncode != 0:
|
||||
raise ValueError(result.stderr.strip())
|
||||
except Exception as e:
|
||||
if progress_callback:
|
||||
progress_callback(0, 0, url, f"error: {e}", 0)
|
||||
return
|
||||
|
||||
# Ogni riga stdout e un JSON (una entry per playlist, una per singolo)
|
||||
entries = []
|
||||
for line in result.stdout.strip().splitlines():
|
||||
if line.strip():
|
||||
try:
|
||||
entries.append(json.loads(line))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
if not entries:
|
||||
if progress_callback:
|
||||
progress_callback(0, 0, url, "error: Nessun contenuto trovato", 0)
|
||||
return
|
||||
|
||||
total = len(entries)
|
||||
|
||||
# 2. Scarica ogni entry
|
||||
for i, entry in enumerate(entries):
|
||||
if is_stopped():
|
||||
if progress_callback:
|
||||
progress_callback(i, total, "", "stopped", 0)
|
||||
return
|
||||
|
||||
title = entry.get("title", f"Track {i + 1}")
|
||||
entry_url = entry.get("url") or entry.get("webpage_url") or entry.get("original_url") or url
|
||||
|
||||
# Per singoli video (non flat-playlist), usa l'URL originale
|
||||
if total == 1:
|
||||
entry_url = url
|
||||
|
||||
# Salta se gia scaricato (file di tracking)
|
||||
track_key = entry.get("id") or title
|
||||
if track_key in done_set:
|
||||
if progress_callback:
|
||||
progress_callback(i, total, title, "skipped", 100)
|
||||
continue
|
||||
|
||||
# Salta se esiste gia un file corrispondente nella cartella
|
||||
if _file_exists_for_query(title, existing_files):
|
||||
_mark_done(done_file, track_key)
|
||||
done_set.add(track_key)
|
||||
if progress_callback:
|
||||
progress_callback(i, total, title, "skipped", 100)
|
||||
continue
|
||||
|
||||
if progress_callback:
|
||||
progress_callback(i, total, title, "downloading", 0)
|
||||
|
||||
cmd = [
|
||||
ytdlp,
|
||||
"--extract-audio",
|
||||
"--audio-format", "mp3",
|
||||
"--audio-quality", bitrate.rstrip("Kk"),
|
||||
"--embed-thumbnail",
|
||||
"--add-metadata",
|
||||
"--no-check-certificates",
|
||||
"--newline",
|
||||
"--output", str(Path(output_dir) / "%(title)s.%(ext)s"),
|
||||
entry_url,
|
||||
]
|
||||
ffmpeg_dir = find_ffmpeg_dir()
|
||||
if ffmpeg_dir:
|
||||
cmd.extend(["--ffmpeg-location", ffmpeg_dir])
|
||||
if cookies_path and Path(cookies_path).exists():
|
||||
cmd.extend(["--cookies", cookies_path])
|
||||
|
||||
try:
|
||||
with _process_lock:
|
||||
_current_process = subprocess.Popen(
|
||||
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
|
||||
)
|
||||
|
||||
for line in _current_process.stdout:
|
||||
if is_stopped():
|
||||
_current_process.terminate()
|
||||
if progress_callback:
|
||||
progress_callback(i, total, title, "stopped", 0)
|
||||
return
|
||||
|
||||
pct_match = re.search(r"(\d+(?:\.\d+)?)%", line)
|
||||
if pct_match and progress_callback:
|
||||
pct = int(float(pct_match.group(1)))
|
||||
progress_callback(i, total, title, "downloading", pct)
|
||||
|
||||
return_code = _current_process.wait()
|
||||
|
||||
with _process_lock:
|
||||
_current_process = None
|
||||
|
||||
if return_code == 0:
|
||||
_mark_done(done_file, track_key)
|
||||
done_set.add(track_key)
|
||||
existing_files = _scan_existing_files(out_path)
|
||||
if progress_callback:
|
||||
progress_callback(i, total, title, "done", 100)
|
||||
else:
|
||||
if progress_callback:
|
||||
progress_callback(i, total, title, f"error: yt-dlp exit {return_code}", 0)
|
||||
|
||||
except Exception as e:
|
||||
with _process_lock:
|
||||
_current_process = None
|
||||
if progress_callback:
|
||||
progress_callback(i, total, title, f"error: {e}", 0)
|
||||
|
||||
if progress_callback and not is_stopped():
|
||||
progress_callback(total, total, "", "completed", 100)
|
||||
+137
@@ -0,0 +1,137 @@
|
||||
"""Risoluzione path per binari esterni (yt-dlp, ffmpeg).
|
||||
|
||||
Quando l'app gira come bundle PyInstaller, i binari sono inclusi
|
||||
nella cartella del bundle. Altrimenti cerca nel sistema.
|
||||
|
||||
Cross-platform: macOS e Windows.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
_IS_WINDOWS = sys.platform == "win32"
|
||||
|
||||
|
||||
def _is_frozen() -> bool:
|
||||
"""True se stiamo girando come bundle PyInstaller."""
|
||||
return getattr(sys, "frozen", False)
|
||||
|
||||
|
||||
def _bundle_dirs() -> list[Path]:
|
||||
"""Directories dove cercare i binari bundled (PyInstaller)."""
|
||||
dirs = []
|
||||
if _is_frozen():
|
||||
# _MEIPASS: directory dove PyInstaller estrae i file
|
||||
meipass = Path(getattr(sys, "_MEIPASS", ""))
|
||||
if meipass.exists():
|
||||
dirs.append(meipass)
|
||||
# Directory dell'eseguibile
|
||||
exe_dir = Path(sys.executable).parent
|
||||
dirs.append(exe_dir)
|
||||
# macOS: Contents/Frameworks/ (PyInstaller onedir .app)
|
||||
if not _IS_WINDOWS:
|
||||
frameworks = exe_dir.parent / "Frameworks"
|
||||
if frameworks.exists():
|
||||
dirs.append(frameworks)
|
||||
return dirs
|
||||
|
||||
|
||||
def _exe(name: str) -> str:
|
||||
"""Aggiunge .exe su Windows."""
|
||||
if _IS_WINDOWS and not name.endswith(".exe"):
|
||||
return name + ".exe"
|
||||
return name
|
||||
|
||||
|
||||
def find_ytdlp() -> str:
|
||||
"""Trova il path di yt-dlp.
|
||||
|
||||
Ordine di ricerca:
|
||||
1. Bundle PyInstaller
|
||||
2. Homebrew (macOS) / PATH comune (Windows)
|
||||
3. Qualsiasi posizione nel PATH
|
||||
"""
|
||||
name = _exe("yt-dlp")
|
||||
|
||||
# 1. Bundle
|
||||
for d in _bundle_dirs():
|
||||
candidate = d / name
|
||||
if candidate.exists():
|
||||
return str(candidate)
|
||||
|
||||
# 2. Percorsi noti per OS
|
||||
if not _IS_WINDOWS:
|
||||
brew = shutil.which("yt-dlp", path="/opt/homebrew/bin:/usr/local/bin")
|
||||
if brew:
|
||||
return brew
|
||||
|
||||
# 3. PATH generico
|
||||
system = shutil.which("yt-dlp")
|
||||
if system:
|
||||
return system
|
||||
|
||||
msg = "yt-dlp non trovato. "
|
||||
if _IS_WINDOWS:
|
||||
msg += "Scaricalo da https://github.com/yt-dlp/yt-dlp/releases"
|
||||
else:
|
||||
msg += "Installalo con: brew install yt-dlp"
|
||||
raise FileNotFoundError(msg)
|
||||
|
||||
|
||||
def find_ffmpeg_dir() -> Optional[str]:
|
||||
"""Trova la directory contenente ffmpeg e ffprobe.
|
||||
|
||||
Ordine di ricerca:
|
||||
1. Bundle PyInstaller
|
||||
2. Percorsi noti per OS
|
||||
3. Qualsiasi posizione nel PATH
|
||||
"""
|
||||
ffmpeg_name = _exe("ffmpeg")
|
||||
|
||||
# 1. Bundle
|
||||
for d in _bundle_dirs():
|
||||
if (d / ffmpeg_name).exists():
|
||||
return str(d)
|
||||
|
||||
# 2. Percorsi noti
|
||||
if _IS_WINDOWS:
|
||||
# Posizioni comuni su Windows
|
||||
for search in (
|
||||
Path(os.environ.get("LOCALAPPDATA", "")) / "Programs" / "ffmpeg" / "bin",
|
||||
Path("C:/ffmpeg/bin"),
|
||||
Path(os.environ.get("ProgramFiles", "")) / "ffmpeg" / "bin",
|
||||
):
|
||||
if (search / "ffmpeg.exe").exists():
|
||||
return str(search)
|
||||
else:
|
||||
for search_path in ("/opt/homebrew/bin", "/usr/local/bin"):
|
||||
if (Path(search_path) / "ffmpeg").exists():
|
||||
return search_path
|
||||
|
||||
# 3. PATH generico
|
||||
ffmpeg = shutil.which("ffmpeg")
|
||||
if ffmpeg:
|
||||
return str(Path(ffmpeg).parent)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def find_ffmpeg() -> Optional[str]:
|
||||
"""Ritorna il path completo di ffmpeg."""
|
||||
d = find_ffmpeg_dir()
|
||||
if d:
|
||||
return str(Path(d) / _exe("ffmpeg"))
|
||||
return None
|
||||
|
||||
|
||||
def find_ffprobe() -> Optional[str]:
|
||||
"""Ritorna il path completo di ffprobe."""
|
||||
d = find_ffmpeg_dir()
|
||||
if d:
|
||||
return str(Path(d) / _exe("ffprobe"))
|
||||
return None
|
||||
@@ -0,0 +1,177 @@
|
||||
"""Client API Spotify — autenticazione e fetch playlist/track/album."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import requests
|
||||
|
||||
|
||||
def get_access_token(client_id: str, client_secret: str) -> str:
|
||||
"""Ottiene un access token tramite Client Credentials Flow."""
|
||||
resp = requests.post(
|
||||
"https://accounts.spotify.com/api/token",
|
||||
data={"grant_type": "client_credentials"},
|
||||
auth=(client_id, client_secret),
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
token = resp.json().get("access_token")
|
||||
if not token:
|
||||
raise ValueError("Token non ricevuto da Spotify")
|
||||
return token
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Riconoscimento tipo URL
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def detect_url_type(url: str) -> tuple[str, str]:
|
||||
"""Rileva il tipo di URL Spotify e ne estrae l'ID.
|
||||
|
||||
Ritorna (tipo, id) dove tipo e "track", "album" o "playlist".
|
||||
Solleva ValueError se l'URL non e riconosciuto.
|
||||
"""
|
||||
for kind in ("track", "album", "playlist"):
|
||||
match = re.search(rf"{kind}/([a-zA-Z0-9]+)", url)
|
||||
if match:
|
||||
return kind, match.group(1)
|
||||
raise ValueError(
|
||||
"URL non riconosciuto. Sono supportati: playlist, album e singoli brani."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Singolo brano
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def get_track_info(token: str, track_id: str) -> tuple[str, list[dict]]:
|
||||
"""Ritorna (track_name, [{"name": ..., "artist": ...}]) per un singolo brano."""
|
||||
resp = requests.get(
|
||||
f"https://api.spotify.com/v1/tracks/{track_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
name = data.get("name", "Unknown")
|
||||
artists = data.get("artists", [])
|
||||
artist = artists[0].get("name", "Unknown") if artists else "Unknown"
|
||||
return name, [{"name": name, "artist": artist}]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Album
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def get_album_info(token: str, album_id: str) -> tuple[str, list[dict]]:
|
||||
"""Ritorna (album_name, lista tracce) per un album."""
|
||||
resp = requests.get(
|
||||
f"https://api.spotify.com/v1/albums/{album_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
album_name = data.get("name", "Unknown Album")
|
||||
|
||||
tracks = []
|
||||
# Prima pagina inclusa nella risposta
|
||||
tracks_data = data.get("tracks", {})
|
||||
_collect_album_tracks(tracks_data, tracks)
|
||||
|
||||
# Paginazione se ci sono piu di 50 tracce
|
||||
next_url = tracks_data.get("next")
|
||||
while next_url:
|
||||
resp = requests.get(
|
||||
next_url,
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
page = resp.json()
|
||||
_collect_album_tracks(page, tracks)
|
||||
next_url = page.get("next")
|
||||
|
||||
return album_name, tracks
|
||||
|
||||
|
||||
def _collect_album_tracks(data: dict, out: list[dict]) -> None:
|
||||
for item in data.get("items", []):
|
||||
name = item.get("name", "Unknown")
|
||||
artists = item.get("artists", [])
|
||||
artist = artists[0].get("name", "Unknown") if artists else "Unknown"
|
||||
out.append({"name": name, "artist": artist})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Playlist
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def get_playlist_info(token: str, playlist_id: str) -> tuple[str, list[dict]]:
|
||||
"""Ritorna (playlist_name, lista tracce) per una playlist."""
|
||||
resp = requests.get(
|
||||
f"https://api.spotify.com/v1/playlists/{playlist_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
name = data.get("name")
|
||||
if not name:
|
||||
raise ValueError("Impossibile leggere il nome della playlist")
|
||||
|
||||
tracks = get_all_tracks(token, playlist_id)
|
||||
return name, tracks
|
||||
|
||||
|
||||
def get_all_tracks(token: str, playlist_id: str) -> list[dict]:
|
||||
"""Scarica tutte le tracce con paginazione (100 per pagina).
|
||||
|
||||
Ritorna lista di {"name": ..., "artist": ...}.
|
||||
"""
|
||||
tracks: list[dict] = []
|
||||
url: str | None = f"https://api.spotify.com/v1/playlists/{playlist_id}/tracks?limit=100&offset=0"
|
||||
|
||||
while url:
|
||||
resp = requests.get(
|
||||
url,
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
|
||||
for item in data.get("items", []):
|
||||
track = item.get("track")
|
||||
if not track:
|
||||
continue
|
||||
name = track.get("name", "Unknown")
|
||||
artists = track.get("artists", [])
|
||||
artist = artists[0].get("name", "Unknown") if artists else "Unknown"
|
||||
tracks.append({"name": name, "artist": artist})
|
||||
|
||||
url = data.get("next") # None quando non ci sono altre pagine
|
||||
|
||||
return tracks
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Funzione unificata
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def resolve_spotify_url(token: str, url: str) -> tuple[str, str, list[dict]]:
|
||||
"""Risolve qualsiasi URL Spotify supportato.
|
||||
|
||||
Ritorna (label, nome, lista_tracce) dove label e "Brano"/"Album"/"Playlist".
|
||||
"""
|
||||
kind, item_id = detect_url_type(url)
|
||||
|
||||
if kind == "track":
|
||||
name, tracks = get_track_info(token, item_id)
|
||||
return "Brano", name, tracks
|
||||
elif kind == "album":
|
||||
name, tracks = get_album_info(token, item_id)
|
||||
return "Album", name, tracks
|
||||
else:
|
||||
name, tracks = get_playlist_info(token, item_id)
|
||||
return "Playlist", name, tracks
|
||||
@@ -0,0 +1,359 @@
|
||||
"""Logica upgrade qualita audio — riscrittura di upgrade_quality.sh."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
from core.paths import find_ytdlp, find_ffmpeg_dir, find_ffprobe
|
||||
|
||||
AUDIO_EXTENSIONS = {".mp3", ".m4a", ".wav", ".flac"}
|
||||
|
||||
# Flag globale per interruzione
|
||||
_stop_event = threading.Event()
|
||||
_current_process: Optional[subprocess.Popen] = None
|
||||
_process_lock = threading.Lock()
|
||||
|
||||
|
||||
def request_stop():
|
||||
_stop_event.set()
|
||||
with _process_lock:
|
||||
if _current_process and _current_process.poll() is None:
|
||||
_current_process.terminate()
|
||||
|
||||
|
||||
def reset_stop():
|
||||
_stop_event.clear()
|
||||
|
||||
|
||||
def is_stopped() -> bool:
|
||||
return _stop_event.is_set()
|
||||
|
||||
|
||||
def scan_audio_files(directory: str, recursive: bool = False) -> list[Path]:
|
||||
"""Trova file audio nella directory."""
|
||||
base = Path(directory)
|
||||
files = []
|
||||
if recursive:
|
||||
for ext in AUDIO_EXTENSIONS:
|
||||
files.extend(base.rglob(f"*{ext}"))
|
||||
else:
|
||||
for ext in AUDIO_EXTENSIONS:
|
||||
files.extend(base.glob(f"*{ext}"))
|
||||
return sorted(files)
|
||||
|
||||
|
||||
def get_bitrate(filepath: str | Path) -> int:
|
||||
"""Ritorna il bitrate in kbps tramite ffprobe. Ritorna 0 in caso di errore."""
|
||||
try:
|
||||
ffprobe_bin = find_ffprobe() or "ffprobe"
|
||||
result = subprocess.run(
|
||||
[
|
||||
ffprobe_bin, "-v", "quiet",
|
||||
"-print_format", "json",
|
||||
"-show_format",
|
||||
str(filepath),
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
data = json.loads(result.stdout)
|
||||
bit_rate = int(data.get("format", {}).get("bit_rate", 0))
|
||||
return bit_rate // 1000
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
|
||||
def _load_done_set(done_file: Path) -> set[str]:
|
||||
if done_file.exists():
|
||||
return set(done_file.read_text(encoding="utf-8").splitlines())
|
||||
return set()
|
||||
|
||||
|
||||
def _mark_done(done_file: Path, filename: str) -> None:
|
||||
with open(done_file, "a", encoding="utf-8") as f:
|
||||
f.write(filename + "\n")
|
||||
|
||||
|
||||
def _search_youtube(query: str, cookies_path: Optional[str] = None) -> str:
|
||||
"""Cerca un video su YouTube e ritorna l'URL."""
|
||||
ytdlp = find_ytdlp()
|
||||
cmd = [
|
||||
ytdlp,
|
||||
f"ytsearch1:{query}",
|
||||
"--dump-json",
|
||||
"--no-download",
|
||||
"--no-warnings",
|
||||
"--flat-playlist",
|
||||
]
|
||||
if cookies_path and Path(cookies_path).exists():
|
||||
cmd.extend(["--cookies", cookies_path])
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||
if result.returncode != 0:
|
||||
raise ValueError("Ricerca fallita")
|
||||
|
||||
data = json.loads(result.stdout)
|
||||
video_id = data.get("id", "")
|
||||
if not video_id:
|
||||
raise ValueError("Nessun risultato")
|
||||
return f"https://www.youtube.com/watch?v={video_id}"
|
||||
|
||||
|
||||
def update_cover_only(
|
||||
filepath: Path,
|
||||
video_url: str,
|
||||
temp_dir: Path,
|
||||
cookies_path: Optional[str] = None,
|
||||
) -> bool:
|
||||
"""Aggiorna solo la copertina di un file audio. Ritorna True se riuscito."""
|
||||
ytdlp = find_ytdlp()
|
||||
cmd = [
|
||||
ytdlp,
|
||||
"--skip-download",
|
||||
"--write-thumbnail",
|
||||
"--convert-thumbnails", "jpg",
|
||||
"--no-warnings",
|
||||
"--output", str(temp_dir / "cover"),
|
||||
video_url,
|
||||
]
|
||||
ffmpeg_dir = _find_ffmpeg_dir()
|
||||
if ffmpeg_dir:
|
||||
cmd.extend(["--ffmpeg-location", ffmpeg_dir])
|
||||
if cookies_path and Path(cookies_path).exists():
|
||||
cmd.extend(["--cookies", cookies_path])
|
||||
|
||||
subprocess.run(cmd, capture_output=True, timeout=60)
|
||||
|
||||
cover_files = list(temp_dir.glob("cover*.jpg"))
|
||||
if not cover_files:
|
||||
return False
|
||||
|
||||
cover_file = cover_files[0]
|
||||
temp_output = temp_dir / "temp_output.mp3"
|
||||
|
||||
try:
|
||||
subprocess.run(
|
||||
[
|
||||
"ffmpeg", "-y",
|
||||
"-i", str(filepath),
|
||||
"-i", str(cover_file),
|
||||
"-map", "0:a", "-map", "1:0",
|
||||
"-c:a", "copy",
|
||||
"-id3v2_version", "3",
|
||||
"-metadata:s:v", "title=Album cover",
|
||||
"-metadata:s:v", "comment=Cover (front)",
|
||||
str(temp_output),
|
||||
],
|
||||
capture_output=True,
|
||||
timeout=60,
|
||||
)
|
||||
except Exception:
|
||||
_cleanup_temp(temp_dir)
|
||||
return False
|
||||
|
||||
if temp_output.exists():
|
||||
temp_output.replace(filepath)
|
||||
_cleanup_temp(temp_dir)
|
||||
return True
|
||||
|
||||
_cleanup_temp(temp_dir)
|
||||
return False
|
||||
|
||||
|
||||
def _cleanup_temp(temp_dir: Path) -> None:
|
||||
for pattern in ("*.jpg", "*.webp", "*.mp3", "*.webm", "*.m4a"):
|
||||
for f in temp_dir.glob(pattern):
|
||||
f.unlink(missing_ok=True)
|
||||
|
||||
|
||||
def upgrade_folder(
|
||||
directory: str,
|
||||
threshold: int = 310,
|
||||
cookies_path: Optional[str] = None,
|
||||
recursive: bool = False,
|
||||
progress_callback: Optional[Callable] = None,
|
||||
) -> None:
|
||||
"""Logica principale di upgrade qualita."""
|
||||
reset_stop()
|
||||
ytdlp = find_ytdlp()
|
||||
|
||||
if recursive:
|
||||
base = Path(directory)
|
||||
folders = set()
|
||||
for ext in AUDIO_EXTENSIONS:
|
||||
for f in base.rglob(f"*{ext}"):
|
||||
folders.add(f.parent)
|
||||
folders = sorted(folders)
|
||||
else:
|
||||
folders = [Path(directory)]
|
||||
|
||||
all_items: list[tuple[Path, Path]] = []
|
||||
for folder in folders:
|
||||
for ext in AUDIO_EXTENSIONS:
|
||||
for f in sorted(folder.glob(f"*{ext}")):
|
||||
all_items.append((f, folder))
|
||||
|
||||
total = len(all_items)
|
||||
if total == 0:
|
||||
if progress_callback:
|
||||
progress_callback(0, 0, "", "no_files", 0, 0)
|
||||
return
|
||||
|
||||
processed = 0
|
||||
|
||||
for filepath, folder in all_items:
|
||||
if is_stopped():
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, "", "stopped", 0, 0)
|
||||
return
|
||||
|
||||
done_file = folder / ".upgraded_tracks"
|
||||
temp_dir = folder / ".temp_download"
|
||||
temp_dir.mkdir(exist_ok=True)
|
||||
|
||||
done_set = _load_done_set(done_file)
|
||||
filename = filepath.stem
|
||||
|
||||
if filename in done_set:
|
||||
processed += 1
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "skipped", 0, 0)
|
||||
continue
|
||||
|
||||
current_kbps = get_bitrate(filepath)
|
||||
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "searching", current_kbps, 0)
|
||||
|
||||
query = filename
|
||||
try:
|
||||
video_url = _search_youtube(query, cookies_path)
|
||||
except Exception:
|
||||
_mark_done(done_file, filename)
|
||||
processed += 1
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "not_found", current_kbps, 0)
|
||||
continue
|
||||
|
||||
if is_stopped():
|
||||
return
|
||||
|
||||
# Se qualita gia alta, aggiorna solo copertina
|
||||
if current_kbps >= threshold:
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "cover_only", current_kbps, 0)
|
||||
|
||||
success = update_cover_only(filepath, video_url, temp_dir, cookies_path)
|
||||
_mark_done(done_file, filename)
|
||||
processed += 1
|
||||
|
||||
status = "cover_done" if success else "cover_failed"
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, status, current_kbps, current_kbps)
|
||||
continue
|
||||
|
||||
# Scarica versione HQ
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "downloading", current_kbps, 0)
|
||||
|
||||
cmd = [
|
||||
ytdlp,
|
||||
"--extract-audio",
|
||||
"--audio-format", "mp3",
|
||||
"--audio-quality", "0",
|
||||
"--embed-thumbnail",
|
||||
"--add-metadata",
|
||||
"--no-warnings",
|
||||
"--newline",
|
||||
"--output", str(temp_dir / "%(title)s.%(ext)s"),
|
||||
video_url,
|
||||
]
|
||||
ffmpeg_dir = find_ffmpeg_dir()
|
||||
if ffmpeg_dir:
|
||||
cmd.extend(["--ffmpeg-location", ffmpeg_dir])
|
||||
if cookies_path and Path(cookies_path).exists():
|
||||
cmd.extend(["--cookies", cookies_path])
|
||||
|
||||
try:
|
||||
with _process_lock:
|
||||
_current_process = subprocess.Popen(
|
||||
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
|
||||
)
|
||||
|
||||
for line in _current_process.stdout:
|
||||
if is_stopped():
|
||||
_current_process.terminate()
|
||||
return
|
||||
pct_match = re.search(r"(\d+(?:\.\d+)?)%", line)
|
||||
if pct_match and progress_callback:
|
||||
pct = int(float(pct_match.group(1)))
|
||||
progress_callback(processed, total, filepath.name, "downloading", current_kbps, pct)
|
||||
|
||||
_current_process.wait()
|
||||
with _process_lock:
|
||||
_current_process = None
|
||||
|
||||
except Exception:
|
||||
with _process_lock:
|
||||
_current_process = None
|
||||
_mark_done(done_file, filename)
|
||||
processed += 1
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "download_error", current_kbps, 0)
|
||||
_cleanup_temp(temp_dir)
|
||||
continue
|
||||
|
||||
# Trova il file scaricato
|
||||
new_files = list(temp_dir.glob("*.mp3"))
|
||||
if not new_files:
|
||||
_mark_done(done_file, filename)
|
||||
processed += 1
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "download_error", current_kbps, 0)
|
||||
_cleanup_temp(temp_dir)
|
||||
continue
|
||||
|
||||
newest = max(new_files, key=lambda p: p.stat().st_mtime)
|
||||
new_kbps = get_bitrate(newest)
|
||||
|
||||
filepath.unlink()
|
||||
newest.replace(filepath.parent / f"{filename}.mp3")
|
||||
|
||||
_mark_done(done_file, filename)
|
||||
_cleanup_temp(temp_dir)
|
||||
processed += 1
|
||||
|
||||
if progress_callback:
|
||||
progress_callback(processed, total, filepath.name, "upgraded", current_kbps, new_kbps)
|
||||
|
||||
for folder in folders:
|
||||
td = folder / ".temp_download"
|
||||
if td.exists():
|
||||
try:
|
||||
td.rmdir()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
if progress_callback and not is_stopped():
|
||||
progress_callback(total, total, "", "completed", 0, 0)
|
||||
|
||||
|
||||
def count_files_info(directory: str, recursive: bool = False) -> tuple[int, int]:
|
||||
"""Ritorna (total_audio_files, already_upgraded)."""
|
||||
files = scan_audio_files(directory, recursive)
|
||||
total = len(files)
|
||||
|
||||
already_done = 0
|
||||
for f in files:
|
||||
done_file = f.parent / ".upgraded_tracks"
|
||||
done_set = _load_done_set(done_file)
|
||||
if f.stem in done_set:
|
||||
already_done += 1
|
||||
|
||||
return total, already_done
|
||||
Reference in new issue
Block a user