Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c7dfe95890 | ||
|
|
25b646d754 | ||
|
|
73b4480890 | ||
|
|
d53b8f20ed | ||
|
|
0f7cd24703 |
No files matched your search
+192
-10
@@ -10,7 +10,7 @@ import time
|
|||||||
import webbrowser
|
import webbrowser
|
||||||
from dataclasses import asdict
|
from dataclasses import asdict
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Optional
|
from typing import Any, Callable, Optional
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
@@ -32,6 +32,7 @@ from core.spotify_client import (
|
|||||||
SpotifyAuthRequired,
|
SpotifyAuthRequired,
|
||||||
)
|
)
|
||||||
from core.metadata import read_metadata, write_metadata, SUPPORTED_EXTS
|
from core.metadata import read_metadata, write_metadata, SUPPORTED_EXTS
|
||||||
|
from core import tagger
|
||||||
from core.recorder import (
|
from core.recorder import (
|
||||||
list_input_devices,
|
list_input_devices,
|
||||||
start_recording,
|
start_recording,
|
||||||
@@ -516,13 +517,20 @@ class Api:
|
|||||||
def start_tracks_download(self, payload: dict) -> dict:
|
def start_tracks_download(self, payload: dict) -> dict:
|
||||||
"""Avvia il download di una tracklist gia parsata
|
"""Avvia il download di una tracklist gia parsata
|
||||||
(lista di {name, artist}). Se 'subfolder' e presente,
|
(lista di {name, artist}). Se 'subfolder' e presente,
|
||||||
scarica in output_dir/subfolder."""
|
scarica in output_dir/subfolder.
|
||||||
|
|
||||||
|
`payload['metadata']` opzionale: lista parallela a `tracks` con dict
|
||||||
|
per il tagging ID3 (title, artist, album, date, genre, tracknumber,
|
||||||
|
cover_url). Se presente, i file MP3 vengono nominati come
|
||||||
|
"Artista - Titolo.mp3" e taggati post-download.
|
||||||
|
"""
|
||||||
if self._any_job_running():
|
if self._any_job_running():
|
||||||
return {"ok": False, "error": "Un download gia in corso"}
|
return {"ok": False, "error": "Un download gia in corso"}
|
||||||
|
|
||||||
tracks = payload.get("tracks") or []
|
tracks = payload.get("tracks") or []
|
||||||
output_dir = (payload.get("output_dir") or "").strip()
|
output_dir = (payload.get("output_dir") or "").strip()
|
||||||
subfolder = (payload.get("subfolder") or "").strip()
|
subfolder = (payload.get("subfolder") or "").strip()
|
||||||
|
metadata_list = payload.get("metadata") or None
|
||||||
if not tracks:
|
if not tracks:
|
||||||
return {"ok": False, "error": "Nessuna traccia fornita"}
|
return {"ok": False, "error": "Nessuna traccia fornita"}
|
||||||
if not output_dir:
|
if not output_dir:
|
||||||
@@ -541,6 +549,7 @@ class Api:
|
|||||||
self._download_thread = threading.Thread(
|
self._download_thread = threading.Thread(
|
||||||
target=self._tracks_worker,
|
target=self._tracks_worker,
|
||||||
args=(list(tracks), output_dir),
|
args=(list(tracks), output_dir),
|
||||||
|
kwargs={"metadata_list": list(metadata_list) if metadata_list else None},
|
||||||
daemon=True,
|
daemon=True,
|
||||||
)
|
)
|
||||||
self._download_thread.start()
|
self._download_thread.start()
|
||||||
@@ -548,7 +557,13 @@ class Api:
|
|||||||
|
|
||||||
def start_urls_download(self, payload: dict) -> dict:
|
def start_urls_download(self, payload: dict) -> dict:
|
||||||
"""Analogo a start_tracks_download ma accetta URL YouTube gia noti
|
"""Analogo a start_tracks_download ma accetta URL YouTube gia noti
|
||||||
(bypass search). Usato dal flow del tab 'YouTube Search'."""
|
(bypass search). Usato dal flow del tab 'YouTube Search'.
|
||||||
|
|
||||||
|
`payload['metadata']` opzionale: lista parallela a `urls` con dict
|
||||||
|
per il tagging ID3 (title, artist, album, date, genre, tracknumber,
|
||||||
|
cover_url). Se presente, i file MP3 vengono nominati come
|
||||||
|
"Artista - Titolo.mp3" e taggati post-download.
|
||||||
|
"""
|
||||||
if self._any_job_running():
|
if self._any_job_running():
|
||||||
return {"ok": False, "error": "Un download gia in corso"}
|
return {"ok": False, "error": "Un download gia in corso"}
|
||||||
|
|
||||||
@@ -556,6 +571,7 @@ class Api:
|
|||||||
titles = payload.get("titles") or []
|
titles = payload.get("titles") or []
|
||||||
output_dir = (payload.get("output_dir") or "").strip()
|
output_dir = (payload.get("output_dir") or "").strip()
|
||||||
subfolder = (payload.get("subfolder") or "").strip()
|
subfolder = (payload.get("subfolder") or "").strip()
|
||||||
|
metadata_list = payload.get("metadata") or None
|
||||||
if not urls:
|
if not urls:
|
||||||
return {"ok": False, "error": "Nessuna URL fornita"}
|
return {"ok": False, "error": "Nessuna URL fornita"}
|
||||||
if len(urls) != len(titles):
|
if len(urls) != len(titles):
|
||||||
@@ -575,6 +591,7 @@ class Api:
|
|||||||
self._download_thread = threading.Thread(
|
self._download_thread = threading.Thread(
|
||||||
target=self._urls_worker,
|
target=self._urls_worker,
|
||||||
args=(list(urls), list(titles), output_dir),
|
args=(list(urls), list(titles), output_dir),
|
||||||
|
kwargs={"metadata_list": list(metadata_list) if metadata_list else None},
|
||||||
daemon=True,
|
daemon=True,
|
||||||
)
|
)
|
||||||
self._download_thread.start()
|
self._download_thread.start()
|
||||||
@@ -585,7 +602,8 @@ class Api:
|
|||||||
self._log("download", "[INFO] Interruzione richiesta...")
|
self._log("download", "[INFO] Interruzione richiesta...")
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|
||||||
def _tracks_worker(self, tracks: list, output_dir: str) -> None:
|
def _tracks_worker(self, tracks: list, output_dir: str,
|
||||||
|
metadata_list: Optional[list] = None) -> None:
|
||||||
reset_download_stop()
|
reset_download_stop()
|
||||||
cfg = load_config()
|
cfg = load_config()
|
||||||
bitrate = cfg.get("bitrate", "320K")
|
bitrate = cfg.get("bitrate", "320K")
|
||||||
@@ -595,6 +613,31 @@ class Api:
|
|||||||
self._log(view, f"[INFO] Tracklist: {len(tracks)} brani da cercare su YouTube")
|
self._log(view, f"[INFO] Tracklist: {len(tracks)} brani da cercare su YouTube")
|
||||||
self._log(view, f"[INFO] Destinazione: {output_dir}")
|
self._log(view, f"[INFO] Destinazione: {output_dir}")
|
||||||
|
|
||||||
|
# Pipeline tagging: se metadata_list fornito, calcola i filename
|
||||||
|
# ("Artista - Titolo") e prepara callback per scrivere ID3 tag
|
||||||
|
# + cover art dopo ogni download riuscito.
|
||||||
|
output_filenames: Optional[list] = None
|
||||||
|
post_cb: Optional[Callable] = None
|
||||||
|
if metadata_list:
|
||||||
|
output_filenames = []
|
||||||
|
for md in metadata_list:
|
||||||
|
md = md or {}
|
||||||
|
stem = tagger.build_filename_stem(
|
||||||
|
md.get("artist", ""),
|
||||||
|
md.get("title", ""),
|
||||||
|
)
|
||||||
|
output_filenames.append(stem)
|
||||||
|
|
||||||
|
def post_cb(idx: int, filepath: str,
|
||||||
|
_md_list=metadata_list) -> None:
|
||||||
|
if idx < 0 or idx >= len(_md_list):
|
||||||
|
return
|
||||||
|
md = _md_list[idx] or {}
|
||||||
|
try:
|
||||||
|
tagger.write_tags(filepath, md)
|
||||||
|
except Exception:
|
||||||
|
pass # tagging best-effort, non blocca il download
|
||||||
|
|
||||||
_last = [0.0]
|
_last = [0.0]
|
||||||
_THROTTLE = 0.10
|
_THROTTLE = 0.10
|
||||||
|
|
||||||
@@ -634,10 +677,15 @@ class Api:
|
|||||||
|
|
||||||
self._emit("download:progress", payload_evt)
|
self._emit("download:progress", payload_evt)
|
||||||
|
|
||||||
download_playlist(tracks, output_dir, bitrate, cookies_path, progress_cb)
|
download_playlist(
|
||||||
|
tracks, output_dir, bitrate, cookies_path, progress_cb,
|
||||||
|
output_filenames=output_filenames,
|
||||||
|
post_download_callback=post_cb,
|
||||||
|
)
|
||||||
self._emit("download:done", {"ok": True})
|
self._emit("download:done", {"ok": True})
|
||||||
|
|
||||||
def _urls_worker(self, urls: list, titles: list, output_dir: str) -> None:
|
def _urls_worker(self, urls: list, titles: list, output_dir: str,
|
||||||
|
metadata_list: Optional[list] = None) -> None:
|
||||||
from core.downloader import download_urls
|
from core.downloader import download_urls
|
||||||
reset_download_stop()
|
reset_download_stop()
|
||||||
cfg = load_config()
|
cfg = load_config()
|
||||||
@@ -648,6 +696,29 @@ class Api:
|
|||||||
self._log(view, f"[INFO] URL list: {len(urls)} da scaricare da YouTube")
|
self._log(view, f"[INFO] URL list: {len(urls)} da scaricare da YouTube")
|
||||||
self._log(view, f"[INFO] Destinazione: {output_dir}")
|
self._log(view, f"[INFO] Destinazione: {output_dir}")
|
||||||
|
|
||||||
|
# Stessa pipeline tagging di _tracks_worker
|
||||||
|
output_filenames: Optional[list] = None
|
||||||
|
post_cb: Optional[Callable] = None
|
||||||
|
if metadata_list:
|
||||||
|
output_filenames = []
|
||||||
|
for md in metadata_list:
|
||||||
|
md = md or {}
|
||||||
|
stem = tagger.build_filename_stem(
|
||||||
|
md.get("artist", ""),
|
||||||
|
md.get("title", ""),
|
||||||
|
)
|
||||||
|
output_filenames.append(stem)
|
||||||
|
|
||||||
|
def post_cb(idx: int, filepath: str,
|
||||||
|
_md_list=metadata_list) -> None:
|
||||||
|
if idx < 0 or idx >= len(_md_list):
|
||||||
|
return
|
||||||
|
md = _md_list[idx] or {}
|
||||||
|
try:
|
||||||
|
tagger.write_tags(filepath, md)
|
||||||
|
except Exception:
|
||||||
|
pass # tagging best-effort, non blocca il download
|
||||||
|
|
||||||
_last = [0.0]
|
_last = [0.0]
|
||||||
_THROTTLE = 0.10
|
_THROTTLE = 0.10
|
||||||
|
|
||||||
@@ -681,7 +752,11 @@ class Api:
|
|||||||
|
|
||||||
self._emit("download:progress", payload_evt)
|
self._emit("download:progress", payload_evt)
|
||||||
|
|
||||||
download_urls(urls, titles, output_dir, bitrate, cookies_path, progress_cb)
|
download_urls(
|
||||||
|
urls, titles, output_dir, bitrate, cookies_path, progress_cb,
|
||||||
|
output_filenames=output_filenames,
|
||||||
|
post_download_callback=post_cb,
|
||||||
|
)
|
||||||
self._emit("download:done", {"ok": True})
|
self._emit("download:done", {"ok": True})
|
||||||
|
|
||||||
def _download_worker(self, urls: list, output_dir: str) -> None:
|
def _download_worker(self, urls: list, output_dir: str) -> None:
|
||||||
@@ -1167,7 +1242,13 @@ class Api:
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
def spotify_search_download(self, tracks: list) -> dict:
|
def spotify_search_download(self, tracks: list) -> dict:
|
||||||
"""Scarica i track Spotify selezionati (name+artist → YouTube search)."""
|
"""Scarica i track Spotify selezionati (name+artist → YouTube search).
|
||||||
|
|
||||||
|
Costruisce anche `metadata` (una entry per track) per il tagging ID3
|
||||||
|
+ cover art post-download. Se le creds Spotify sono presenti,
|
||||||
|
arricchisce con i generi dell'artista (cache locale per artist_id
|
||||||
|
per evitare N+1 chiamate).
|
||||||
|
"""
|
||||||
cfg = load_config()
|
cfg = load_config()
|
||||||
out_root = (cfg.get("output_dir") or "").strip()
|
out_root = (cfg.get("output_dir") or "").strip()
|
||||||
if not out_root:
|
if not out_root:
|
||||||
@@ -1176,7 +1257,20 @@ class Api:
|
|||||||
target = self._music_output_dir(out_root, "Spotify")
|
target = self._music_output_dir(out_root, "Spotify")
|
||||||
subfolder = target.name
|
subfolder = target.name
|
||||||
|
|
||||||
converted = []
|
# Tentativo di token Spotify per fetch generi (best-effort, opzionale)
|
||||||
|
cid = (cfg.get("client_id") or "").strip()
|
||||||
|
secret = (cfg.get("client_secret") or "").strip()
|
||||||
|
token: Optional[str] = None
|
||||||
|
if cid and secret:
|
||||||
|
try:
|
||||||
|
token = spotify_client.get_access_token(cid, secret)
|
||||||
|
except Exception:
|
||||||
|
token = None
|
||||||
|
|
||||||
|
genre_cache: dict = {} # artist_id -> [genres]
|
||||||
|
|
||||||
|
converted: list = []
|
||||||
|
metadata: list = []
|
||||||
for t in tracks:
|
for t in tracks:
|
||||||
title = (t.get("name") or "").strip()
|
title = (t.get("name") or "").strip()
|
||||||
artists = (t.get("artists") or "").strip()
|
artists = (t.get("artists") or "").strip()
|
||||||
@@ -1184,6 +1278,27 @@ class Api:
|
|||||||
continue
|
continue
|
||||||
converted.append({"name": title, "artist": artists})
|
converted.append({"name": title, "artist": artists})
|
||||||
|
|
||||||
|
# Genere: primo genere dell'artista (se disponibile e token OK)
|
||||||
|
genre = ""
|
||||||
|
artist_id = (t.get("artist_id") or "").strip()
|
||||||
|
if token and artist_id:
|
||||||
|
if artist_id not in genre_cache:
|
||||||
|
genre_cache[artist_id] = spotify_client.get_artist_genres(token, artist_id)
|
||||||
|
genres = genre_cache.get(artist_id) or []
|
||||||
|
if genres:
|
||||||
|
genre = genres[0]
|
||||||
|
|
||||||
|
tn = t.get("track_number") or 0
|
||||||
|
metadata.append({
|
||||||
|
"title": title,
|
||||||
|
"artist": artists,
|
||||||
|
"album": (t.get("album") or "").strip(),
|
||||||
|
"date": (t.get("release_date") or "").strip(),
|
||||||
|
"genre": genre,
|
||||||
|
"tracknumber": str(tn) if tn else "",
|
||||||
|
"cover_url": (t.get("cover_url_large") or t.get("image_url") or "").strip(),
|
||||||
|
})
|
||||||
|
|
||||||
if not converted:
|
if not converted:
|
||||||
return {"ok": False, "error": "Nessun brano valido"}
|
return {"ok": False, "error": "Nessun brano valido"}
|
||||||
|
|
||||||
@@ -1191,6 +1306,7 @@ class Api:
|
|||||||
"tracks": converted,
|
"tracks": converted,
|
||||||
"output_dir": out_root,
|
"output_dir": out_root,
|
||||||
"subfolder": subfolder,
|
"subfolder": subfolder,
|
||||||
|
"metadata": metadata,
|
||||||
})
|
})
|
||||||
|
|
||||||
# ---- YouTube ----
|
# ---- YouTube ----
|
||||||
@@ -1242,7 +1358,14 @@ class Api:
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
def youtube_search_download(self, tracks: list) -> dict:
|
def youtube_search_download(self, tracks: list) -> dict:
|
||||||
"""Scarica direttamente gli URL YouTube selezionati (no re-search)."""
|
"""Scarica direttamente gli URL YouTube selezionati (no re-search).
|
||||||
|
|
||||||
|
Per ogni video prova a dedurre i metadati:
|
||||||
|
1) Se le creds Spotify sono presenti, cerca il titolo su Spotify
|
||||||
|
e usa i tag Spotify se c'è un match (best-effort).
|
||||||
|
2) Altrimenti fallback: parsing "Artista - Titolo" dal titolo del
|
||||||
|
video; uploader come fallback per l'album.
|
||||||
|
"""
|
||||||
cfg = load_config()
|
cfg = load_config()
|
||||||
out_root = (cfg.get("output_dir") or "").strip()
|
out_root = (cfg.get("output_dir") or "").strip()
|
||||||
if not out_root:
|
if not out_root:
|
||||||
@@ -1251,8 +1374,21 @@ class Api:
|
|||||||
target = self._music_output_dir(out_root, "YouTube")
|
target = self._music_output_dir(out_root, "YouTube")
|
||||||
subfolder = target.name
|
subfolder = target.name
|
||||||
|
|
||||||
|
# Token Spotify opzionale per enrichment
|
||||||
|
cid = (cfg.get("client_id") or "").strip()
|
||||||
|
secret = (cfg.get("client_secret") or "").strip()
|
||||||
|
token: Optional[str] = None
|
||||||
|
if cid and secret:
|
||||||
|
try:
|
||||||
|
token = spotify_client.get_access_token(cid, secret)
|
||||||
|
except Exception:
|
||||||
|
token = None
|
||||||
|
|
||||||
|
genre_cache: dict = {} # artist_id -> [genres]
|
||||||
|
|
||||||
urls: list = []
|
urls: list = []
|
||||||
titles: list = []
|
titles: list = []
|
||||||
|
metadata: list = []
|
||||||
for t in tracks:
|
for t in tracks:
|
||||||
url = (t.get("url") or "").strip()
|
url = (t.get("url") or "").strip()
|
||||||
title = (t.get("title") or "").strip()
|
title = (t.get("title") or "").strip()
|
||||||
@@ -1261,6 +1397,51 @@ class Api:
|
|||||||
urls.append(url)
|
urls.append(url)
|
||||||
titles.append(title)
|
titles.append(title)
|
||||||
|
|
||||||
|
md: dict = {}
|
||||||
|
# 1) Enrichment via Spotify search (se token)
|
||||||
|
if token:
|
||||||
|
sp = spotify_client.enrich_from_youtube_title(token, title)
|
||||||
|
if sp:
|
||||||
|
# Genere: opzionale
|
||||||
|
genre = ""
|
||||||
|
artist_id = (sp.get("artist_id") or "").strip()
|
||||||
|
if artist_id:
|
||||||
|
if artist_id not in genre_cache:
|
||||||
|
genre_cache[artist_id] = spotify_client.get_artist_genres(token, artist_id)
|
||||||
|
genres = genre_cache.get(artist_id) or []
|
||||||
|
if genres:
|
||||||
|
genre = genres[0]
|
||||||
|
tn = sp.get("track_number") or 0
|
||||||
|
md = {
|
||||||
|
"title": (sp.get("name") or "").strip(),
|
||||||
|
"artist": (sp.get("artists") or "").strip(),
|
||||||
|
"album": (sp.get("album") or "").strip(),
|
||||||
|
"date": (sp.get("release_date") or "").strip(),
|
||||||
|
"genre": genre,
|
||||||
|
"tracknumber": str(tn) if tn else "",
|
||||||
|
"cover_url": (sp.get("cover_url_large") or sp.get("image_url") or "").strip(),
|
||||||
|
}
|
||||||
|
|
||||||
|
# 2) Fallback: parsing del titolo YouTube
|
||||||
|
if not md.get("title") or not md.get("artist"):
|
||||||
|
parsed = _parse_track_line(title)
|
||||||
|
if parsed:
|
||||||
|
fb_artist = parsed["artist"]
|
||||||
|
fb_title = parsed["name"]
|
||||||
|
else:
|
||||||
|
fb_artist = ""
|
||||||
|
fb_title = title
|
||||||
|
md = {
|
||||||
|
"title": fb_title,
|
||||||
|
"artist": fb_artist,
|
||||||
|
"album": (t.get("channel") or "").strip(),
|
||||||
|
"date": (t.get("release_date") or "").strip(),
|
||||||
|
"genre": "",
|
||||||
|
"tracknumber": "",
|
||||||
|
"cover_url": (t.get("image_url") or "").strip(),
|
||||||
|
}
|
||||||
|
metadata.append(md)
|
||||||
|
|
||||||
if not urls:
|
if not urls:
|
||||||
return {"ok": False, "error": "Nessun URL valido"}
|
return {"ok": False, "error": "Nessun URL valido"}
|
||||||
|
|
||||||
@@ -1269,4 +1450,5 @@ class Api:
|
|||||||
"titles": titles,
|
"titles": titles,
|
||||||
"output_dir": out_root,
|
"output_dir": out_root,
|
||||||
"subfolder": subfolder,
|
"subfolder": subfolder,
|
||||||
|
"metadata": metadata,
|
||||||
})
|
})
|
||||||
+1
-1
@@ -5,7 +5,7 @@ import os
|
|||||||
import sys
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
VERSION = "v1.8.3"
|
VERSION = "v1.8.4"
|
||||||
|
|
||||||
|
|
||||||
APP_NAME = "MusicTools"
|
APP_NAME = "MusicTools"
|
||||||
|
|||||||
+48
-2
@@ -121,6 +121,8 @@ def download_playlist(
|
|||||||
bitrate: str = "320K",
|
bitrate: str = "320K",
|
||||||
cookies_path: Optional[str] = None,
|
cookies_path: Optional[str] = None,
|
||||||
progress_callback: Optional[Callable] = None,
|
progress_callback: Optional[Callable] = None,
|
||||||
|
output_filenames: Optional[list] = None,
|
||||||
|
post_download_callback: Optional[Callable] = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Scarica tutti i brani dalla lista di tracce.
|
"""Scarica tutti i brani dalla lista di tracce.
|
||||||
|
|
||||||
@@ -130,6 +132,12 @@ def download_playlist(
|
|||||||
bitrate: qualita audio (es. "320K")
|
bitrate: qualita audio (es. "320K")
|
||||||
cookies_path: percorso file cookies (opzionale)
|
cookies_path: percorso file cookies (opzionale)
|
||||||
progress_callback: callback(track_index, total, track_name, status, percent)
|
progress_callback: callback(track_index, total, track_name, status, percent)
|
||||||
|
output_filenames: opzionale, parallelo a `tracks`. Se presente e non
|
||||||
|
vuoto per l'indice i, yt-dlp scrive `<output_filenames[i]>.mp3`
|
||||||
|
invece di usare il titolo YouTube. Passa gia' sanitizzato.
|
||||||
|
post_download_callback: opzionale, chiamato come `(i, filepath)`
|
||||||
|
dopo yt-dlp exit 0 con il path effettivo del file .mp3 salvato.
|
||||||
|
Usato per tagging ID3 post-download.
|
||||||
"""
|
"""
|
||||||
reset_stop()
|
reset_stop()
|
||||||
total = len(tracks)
|
total = len(tracks)
|
||||||
@@ -183,6 +191,15 @@ def download_playlist(
|
|||||||
if progress_callback:
|
if progress_callback:
|
||||||
progress_callback(i, total, query, "downloading", 0)
|
progress_callback(i, total, query, "downloading", 0)
|
||||||
|
|
||||||
|
# Nome file custom (per tagging: "Artista - Titolo") oppure titolo YouTube grezzo
|
||||||
|
custom_stem = ""
|
||||||
|
if output_filenames and i < len(output_filenames):
|
||||||
|
custom_stem = (output_filenames[i] or "").strip()
|
||||||
|
if custom_stem:
|
||||||
|
out_template = str(Path(output_dir) / f"{custom_stem}.%(ext)s")
|
||||||
|
else:
|
||||||
|
out_template = str(Path(output_dir) / "%(title)s.%(ext)s")
|
||||||
|
|
||||||
# Scarica con yt-dlp subprocess
|
# Scarica con yt-dlp subprocess
|
||||||
cmd = [
|
cmd = [
|
||||||
ytdlp,
|
ytdlp,
|
||||||
@@ -193,7 +210,7 @@ def download_playlist(
|
|||||||
"--add-metadata",
|
"--add-metadata",
|
||||||
"--no-check-certificates",
|
"--no-check-certificates",
|
||||||
"--newline",
|
"--newline",
|
||||||
"--output", str(Path(output_dir) / "%(title)s.%(ext)s"),
|
"--output", out_template,
|
||||||
video_url,
|
video_url,
|
||||||
]
|
]
|
||||||
ffmpeg_dir = find_ffmpeg_dir()
|
ffmpeg_dir = find_ffmpeg_dir()
|
||||||
@@ -231,6 +248,12 @@ def download_playlist(
|
|||||||
_mark_done(done_file, query)
|
_mark_done(done_file, query)
|
||||||
done_set.add(query)
|
done_set.add(query)
|
||||||
existing_files = _scan_existing_files(out_path)
|
existing_files = _scan_existing_files(out_path)
|
||||||
|
if post_download_callback and custom_stem:
|
||||||
|
mp3_path = Path(output_dir) / f"{custom_stem}.mp3"
|
||||||
|
try:
|
||||||
|
post_download_callback(i, str(mp3_path))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
if progress_callback:
|
if progress_callback:
|
||||||
progress_callback(i, total, query, "done", 100)
|
progress_callback(i, total, query, "done", 100)
|
||||||
else:
|
else:
|
||||||
@@ -406,6 +429,8 @@ def download_urls(
|
|||||||
bitrate: str = "320K",
|
bitrate: str = "320K",
|
||||||
cookies_path: Optional[str] = None,
|
cookies_path: Optional[str] = None,
|
||||||
progress_callback: Optional[Callable] = None,
|
progress_callback: Optional[Callable] = None,
|
||||||
|
output_filenames: Optional[list] = None,
|
||||||
|
post_download_callback: Optional[Callable] = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Scarica direttamente da URL YouTube (bypassa search).
|
"""Scarica direttamente da URL YouTube (bypassa search).
|
||||||
|
|
||||||
@@ -420,6 +445,12 @@ def download_urls(
|
|||||||
bitrate: qualita audio (default 320K)
|
bitrate: qualita audio (default 320K)
|
||||||
cookies_path: file cookies opzionale
|
cookies_path: file cookies opzionale
|
||||||
progress_callback: callback(idx, total, title, status, percent)
|
progress_callback: callback(idx, total, title, status, percent)
|
||||||
|
output_filenames: opzionale, parallelo a `urls`. Se presente e non
|
||||||
|
vuoto per l'indice i, yt-dlp scrive `<output_filenames[i]>.mp3`
|
||||||
|
invece di usare il titolo YouTube. Passa gia' sanitizzato.
|
||||||
|
post_download_callback: opzionale, chiamato come `(i, filepath)`
|
||||||
|
dopo yt-dlp exit 0 con il path effettivo del file .mp3 salvato.
|
||||||
|
Usato per tagging ID3 post-download.
|
||||||
"""
|
"""
|
||||||
global _current_process
|
global _current_process
|
||||||
reset_stop()
|
reset_stop()
|
||||||
@@ -456,6 +487,15 @@ def download_urls(
|
|||||||
if progress_callback:
|
if progress_callback:
|
||||||
progress_callback(i, total, key, "downloading", 0)
|
progress_callback(i, total, key, "downloading", 0)
|
||||||
|
|
||||||
|
# Nome file custom (per tagging: "Artista - Titolo") oppure titolo YouTube grezzo
|
||||||
|
custom_stem = ""
|
||||||
|
if output_filenames and i < len(output_filenames):
|
||||||
|
custom_stem = (output_filenames[i] or "").strip()
|
||||||
|
if custom_stem:
|
||||||
|
out_template = str(Path(output_dir) / f"{custom_stem}.%(ext)s")
|
||||||
|
else:
|
||||||
|
out_template = str(Path(output_dir) / "%(title)s.%(ext)s")
|
||||||
|
|
||||||
cmd = [
|
cmd = [
|
||||||
ytdlp,
|
ytdlp,
|
||||||
"--extract-audio",
|
"--extract-audio",
|
||||||
@@ -465,7 +505,7 @@ def download_urls(
|
|||||||
"--add-metadata",
|
"--add-metadata",
|
||||||
"--no-check-certificates",
|
"--no-check-certificates",
|
||||||
"--newline",
|
"--newline",
|
||||||
"--output", str(Path(output_dir) / "%(title)s.%(ext)s"),
|
"--output", out_template,
|
||||||
url,
|
url,
|
||||||
]
|
]
|
||||||
ffmpeg_dir = find_ffmpeg_dir()
|
ffmpeg_dir = find_ffmpeg_dir()
|
||||||
@@ -502,6 +542,12 @@ def download_urls(
|
|||||||
_mark_done(done_file, key)
|
_mark_done(done_file, key)
|
||||||
done_set.add(key)
|
done_set.add(key)
|
||||||
existing_files = _scan_existing_files(out_path)
|
existing_files = _scan_existing_files(out_path)
|
||||||
|
if post_download_callback and custom_stem:
|
||||||
|
mp3_path = Path(output_dir) / f"{custom_stem}.mp3"
|
||||||
|
try:
|
||||||
|
post_download_callback(i, str(mp3_path))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
if progress_callback:
|
if progress_callback:
|
||||||
progress_callback(i, total, key, "done", 100)
|
progress_callback(i, total, key, "done", 100)
|
||||||
else:
|
else:
|
||||||
|
|||||||
+62
-6
@@ -308,23 +308,74 @@ def search_tracks(token: str, query: str, limit: int = 50) -> list:
|
|||||||
|
|
||||||
def _track_to_dict(t: dict) -> dict:
|
def _track_to_dict(t: dict) -> dict:
|
||||||
"""Mappa il track object Spotify sul nostro schema uniforme."""
|
"""Mappa il track object Spotify sul nostro schema uniforme."""
|
||||||
images = t.get("album", {}).get("images", []) or []
|
album = t.get("album", {}) or {}
|
||||||
|
images = album.get("images", []) or []
|
||||||
# Spotify torna 3 taglie ordinate large->small. Prendo la più piccola (~64px)
|
# Spotify torna 3 taglie ordinate large->small. Prendo la più piccola (~64px)
|
||||||
# se disponibile, altrimenti la prima che c'è.
|
# per la thumbnail UI, la più grande (~640px) per il tagging cover.
|
||||||
image_url = ""
|
image_url = "" # ~64px per UI thumbnail
|
||||||
|
cover_url_large = "" # ~640px per tagging ID3
|
||||||
if images:
|
if images:
|
||||||
image_url = images[-1].get("url", "") or images[0].get("url", "")
|
image_url = images[-1].get("url", "") or images[0].get("url", "")
|
||||||
|
cover_url_large = images[0].get("url", "") or image_url
|
||||||
|
# release_date può essere YYYY, YYYY-MM o YYYY-MM-DD in base a release_date_precision
|
||||||
|
release_date = str(album.get("release_date") or "").strip()
|
||||||
|
# track_number (opzionale: presente su brani da album, assente su top-tracks flat)
|
||||||
|
tracknumber = t.get("track_number")
|
||||||
|
# artist_id (primo artista) per fetchare i generi ID3
|
||||||
|
artists_arr = t.get("artists", []) or []
|
||||||
|
artist_id = artists_arr[0].get("id", "") if artists_arr else ""
|
||||||
return {
|
return {
|
||||||
"id": t.get("id", ""),
|
"id": t.get("id", ""),
|
||||||
"url": t.get("external_urls", {}).get("spotify", ""),
|
"url": t.get("external_urls", {}).get("spotify", ""),
|
||||||
"name": t.get("name", ""),
|
"name": t.get("name", ""),
|
||||||
"artists": ", ".join(a.get("name", "") for a in t.get("artists", [])),
|
"artists": ", ".join(a.get("name", "") for a in artists_arr),
|
||||||
"album": t.get("album", {}).get("name", ""),
|
"album": album.get("name", ""),
|
||||||
"duration_sec": int(t.get("duration_ms", 0)) // 1000,
|
"duration_sec": int(t.get("duration_ms", 0)) // 1000,
|
||||||
"image_url": image_url,
|
"image_url": image_url,
|
||||||
|
"cover_url_large": cover_url_large,
|
||||||
|
"release_date": release_date,
|
||||||
|
"track_number": int(tracknumber) if tracknumber else 0,
|
||||||
|
"artist_id": artist_id,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def get_artist_genres(token: str, artist_id: str) -> list:
|
||||||
|
"""Ritorna la lista di generi dell'artista (può essere vuota).
|
||||||
|
|
||||||
|
Silenzia gli errori HTTP: usato per arricchimento best-effort dei tag
|
||||||
|
ID3, non deve mai bloccare il download.
|
||||||
|
"""
|
||||||
|
if not artist_id:
|
||||||
|
return []
|
||||||
|
try:
|
||||||
|
resp = requests.get(
|
||||||
|
f"https://api.spotify.com/v1/artists/{artist_id}",
|
||||||
|
headers={"Authorization": f"Bearer {token}"},
|
||||||
|
timeout=10,
|
||||||
|
)
|
||||||
|
if resp.status_code != 200:
|
||||||
|
return []
|
||||||
|
return list(resp.json().get("genres") or [])
|
||||||
|
except Exception:
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
def enrich_from_youtube_title(token: str, video_title: str) -> dict:
|
||||||
|
"""Cerca su Spotify il primo match per il titolo di un video YouTube.
|
||||||
|
|
||||||
|
Ritorna un dict con la stessa shape di _track_to_dict, oppure {} se
|
||||||
|
nessun match / errore. Best-effort per il tagging dei download YouTube.
|
||||||
|
"""
|
||||||
|
query = (video_title or "").strip()
|
||||||
|
if not query:
|
||||||
|
return {}
|
||||||
|
try:
|
||||||
|
results = search_tracks(token, query, limit=1)
|
||||||
|
except Exception:
|
||||||
|
return {}
|
||||||
|
return results[0] if results else {}
|
||||||
|
|
||||||
|
|
||||||
def search_artist_discography(token: str, artist_name: str) -> list:
|
def search_artist_discography(token: str, artist_name: str) -> list:
|
||||||
"""Trova l'artista esatto (o il piu' popolare tra i match) e ritorna
|
"""Trova l'artista esatto (o il piu' popolare tra i match) e ritorna
|
||||||
tutti i suoi brani: top tracks + tracce di ogni album/single.
|
tutti i suoi brani: top tracks + tracce di ogni album/single.
|
||||||
@@ -385,6 +436,7 @@ def search_artist_discography(token: str, artist_name: str) -> list:
|
|||||||
alb_id = alb.get("id")
|
alb_id = alb.get("id")
|
||||||
alb_name = alb.get("name", "")
|
alb_name = alb.get("name", "")
|
||||||
alb_images = alb.get("images", []) or []
|
alb_images = alb.get("images", []) or []
|
||||||
|
alb_release_date = alb.get("release_date", "")
|
||||||
if not alb_id:
|
if not alb_id:
|
||||||
continue
|
continue
|
||||||
time.sleep(0.1) # rate limit interno
|
time.sleep(0.1) # rate limit interno
|
||||||
@@ -397,7 +449,11 @@ def search_artist_discography(token: str, artist_name: str) -> list:
|
|||||||
r_at.raise_for_status()
|
r_at.raise_for_status()
|
||||||
for t in r_at.json().get("items", []):
|
for t in r_at.json().get("items", []):
|
||||||
t = dict(t)
|
t = dict(t)
|
||||||
t.setdefault("album", {"name": alb_name, "images": alb_images})
|
t.setdefault("album", {
|
||||||
|
"name": alb_name,
|
||||||
|
"images": alb_images,
|
||||||
|
"release_date": alb_release_date,
|
||||||
|
})
|
||||||
collected.append(_track_to_dict(t))
|
collected.append(_track_to_dict(t))
|
||||||
|
|
||||||
# 5. Dedupe
|
# 5. Dedupe
|
||||||
|
|||||||
+109
@@ -0,0 +1,109 @@
|
|||||||
|
"""Scrittura tag ID3 su file MP3 via mutagen, con cover art da URL."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
import requests
|
||||||
|
from mutagen.easyid3 import EasyID3
|
||||||
|
from mutagen.id3 import ID3, APIC
|
||||||
|
from mutagen.mp3 import MP3
|
||||||
|
|
||||||
|
|
||||||
|
SAFE_FILENAME_RE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||||||
|
|
||||||
|
|
||||||
|
def sanitize_filename_stem(stem: str, max_len: int = 180) -> str:
|
||||||
|
"""Rimuove caratteri non validi per un filename e limita lunghezza."""
|
||||||
|
s = SAFE_FILENAME_RE.sub("_", stem or "").strip()
|
||||||
|
s = re.sub(r"\s+", " ", s)
|
||||||
|
if len(s) > max_len:
|
||||||
|
s = s[:max_len].rstrip()
|
||||||
|
return s or "untitled"
|
||||||
|
|
||||||
|
|
||||||
|
def build_filename_stem(artist: str, title: str) -> str:
|
||||||
|
"""Compone 'Artista - Titolo' sanitizzato."""
|
||||||
|
artist = (artist or "").strip()
|
||||||
|
title = (title or "").strip()
|
||||||
|
if artist and title:
|
||||||
|
return sanitize_filename_stem(f"{artist} - {title}")
|
||||||
|
return sanitize_filename_stem(title or artist or "untitled")
|
||||||
|
|
||||||
|
|
||||||
|
def _download_cover(url: str) -> Optional[bytes]:
|
||||||
|
"""Scarica bytes cover, ritorna None su qualsiasi errore."""
|
||||||
|
try:
|
||||||
|
resp = requests.get(url, timeout=15)
|
||||||
|
if resp.status_code == 200:
|
||||||
|
return resp.content
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _cover_mime(url: str, content: bytes) -> str:
|
||||||
|
"""Deduci MIME dell'immagine da URL o magic bytes."""
|
||||||
|
u = url.lower()
|
||||||
|
if u.endswith(".png"):
|
||||||
|
return "image/png"
|
||||||
|
if u.endswith((".jpg", ".jpeg")):
|
||||||
|
return "image/jpeg"
|
||||||
|
# magic bytes
|
||||||
|
if content[:8] == b"\x89PNG\r\n\x1a\n":
|
||||||
|
return "image/png"
|
||||||
|
if content[:3] == b"\xff\xd8\xff":
|
||||||
|
return "image/jpeg"
|
||||||
|
return "image/jpeg" # default
|
||||||
|
|
||||||
|
|
||||||
|
def write_tags(filepath: str, metadata: dict) -> None:
|
||||||
|
"""Scrive ID3 tag su file MP3 esistente.
|
||||||
|
|
||||||
|
metadata: dict con chiavi opzionali:
|
||||||
|
title, artist, album, date (YYYY o YYYY-MM-DD), genre,
|
||||||
|
tracknumber, cover_url
|
||||||
|
|
||||||
|
Errori non fatali (file mancante, tag non scrivibili): silenziosi
|
||||||
|
per non bloccare il download principale.
|
||||||
|
"""
|
||||||
|
p = Path(filepath)
|
||||||
|
if not p.exists():
|
||||||
|
return
|
||||||
|
|
||||||
|
# 1. Tag testuali via EasyID3
|
||||||
|
try:
|
||||||
|
try:
|
||||||
|
audio = EasyID3(str(p))
|
||||||
|
except Exception:
|
||||||
|
# Se il file non ha header ID3, aggiungine uno vuoto
|
||||||
|
mp3 = MP3(str(p))
|
||||||
|
if mp3.tags is None:
|
||||||
|
mp3.add_tags()
|
||||||
|
mp3.save()
|
||||||
|
audio = EasyID3(str(p))
|
||||||
|
for key in ("title", "artist", "album", "date", "genre", "tracknumber"):
|
||||||
|
val = metadata.get(key)
|
||||||
|
if val:
|
||||||
|
audio[key] = str(val)
|
||||||
|
audio.save(str(p))
|
||||||
|
except Exception:
|
||||||
|
return # ID3 broken, non provo la cover
|
||||||
|
|
||||||
|
# 2. Cover art via ID3 APIC
|
||||||
|
cover_url = (metadata.get("cover_url") or "").strip()
|
||||||
|
if not cover_url:
|
||||||
|
return
|
||||||
|
cover_bytes = _download_cover(cover_url)
|
||||||
|
if not cover_bytes:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
mime = _cover_mime(cover_url, cover_bytes)
|
||||||
|
tags = ID3(str(p))
|
||||||
|
tags.delall("APIC")
|
||||||
|
tags.add(APIC(encoding=3, mime=mime, type=3, desc="Cover", data=cover_bytes))
|
||||||
|
tags.save(str(p))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
@@ -0,0 +1,57 @@
|
|||||||
|
"""Test per core.tagger — filename sanitization + write_tags safety net."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from core import tagger
|
||||||
|
|
||||||
|
|
||||||
|
class TestSanitizeFilenameStem:
|
||||||
|
def test_removes_bad_chars(self):
|
||||||
|
# Tutti i char vietati su Windows/macOS: < > : " / \ | ? *
|
||||||
|
got = tagger.sanitize_filename_stem('bad<name>:"/\\|?*test')
|
||||||
|
# Nessuno dei char proibiti sopravvive
|
||||||
|
for ch in '<>:"/\\|?*':
|
||||||
|
assert ch not in got
|
||||||
|
# E ho ottenuto qualcosa di non vuoto
|
||||||
|
assert got
|
||||||
|
|
||||||
|
def test_collapses_whitespace(self):
|
||||||
|
assert tagger.sanitize_filename_stem("a b c") == "a b c"
|
||||||
|
|
||||||
|
def test_truncates_long_stem(self):
|
||||||
|
s = "x" * 500
|
||||||
|
got = tagger.sanitize_filename_stem(s, max_len=180)
|
||||||
|
assert len(got) <= 180
|
||||||
|
|
||||||
|
def test_empty_returns_untitled(self):
|
||||||
|
assert tagger.sanitize_filename_stem("") == "untitled"
|
||||||
|
assert tagger.sanitize_filename_stem(" ") == "untitled"
|
||||||
|
|
||||||
|
|
||||||
|
class TestBuildFilenameStem:
|
||||||
|
def test_format_artist_dash_title(self):
|
||||||
|
assert tagger.build_filename_stem("Kapuchon", "Hot Sauce") == "Kapuchon - Hot Sauce"
|
||||||
|
|
||||||
|
def test_multiple_artists_ok(self):
|
||||||
|
got = tagger.build_filename_stem("Kapuchon, Miss Monique & GLZ", "Hot Sauce (Extended)")
|
||||||
|
assert got == "Kapuchon, Miss Monique & GLZ - Hot Sauce (Extended)"
|
||||||
|
|
||||||
|
def test_missing_artist_uses_title_only(self):
|
||||||
|
assert tagger.build_filename_stem("", "Only Title") == "Only Title"
|
||||||
|
|
||||||
|
def test_missing_title_uses_artist_only(self):
|
||||||
|
assert tagger.build_filename_stem("Only Artist", "") == "Only Artist"
|
||||||
|
|
||||||
|
def test_sanitizes_bad_chars(self):
|
||||||
|
got = tagger.build_filename_stem("A/B", "C:D")
|
||||||
|
# Nessuno slash o colon residuo
|
||||||
|
assert "/" not in got
|
||||||
|
assert ":" not in got
|
||||||
|
|
||||||
|
|
||||||
|
class TestWriteTagsMissingFile:
|
||||||
|
def test_missing_file_is_noop(self):
|
||||||
|
# Non deve sollevare alcuna eccezione se il file non esiste
|
||||||
|
tagger.write_tags("/nonexistent/path/to/file.mp3", {
|
||||||
|
"title": "x", "artist": "y", "cover_url": "http://example.com/x.jpg",
|
||||||
|
})
|
||||||
Reference in new issue
Block a user