From 5bd3c9b5f068336164686b0bd67d0db974415436 Mon Sep 17 00:00:00 2001 From: Savsis Date: Thu, 4 Jun 2026 05:06:28 +0500 Subject: [PATCH] Create platforms.py --- platforms.py | 523 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 523 insertions(+) create mode 100644 platforms.py diff --git a/platforms.py b/platforms.py new file mode 100644 index 0000000..4368d4e --- /dev/null +++ b/platforms.py @@ -0,0 +1,523 @@ +import sys +import re +import json +import time +import requests +import yt_dlp +from pathlib import Path + +SCRIPT_DIR = Path(__file__).parent + + +def _ffmpeg_path() -> str | None: + exe = "ffmpeg.exe" if sys.platform == "win32" else "ffmpeg" + local = SCRIPT_DIR / "ffmpeg" / exe + if local.exists(): + return str(SCRIPT_DIR / "ffmpeg") + return None + +AUDIO_EXTENSIONS = {"mp3", "flac", "wav", "m4a", "ogg", "opus", "mp4", "webm", "aac", "wma"} +THUMB_EXTENSIONS = {"jpg", "jpeg", "png", "webp"} + +_ANON_TOKEN: str | None = None +_TOKEN_EXPIRY: float = 0.0 + +_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" +) + + +def _sanitize_filename(name: str) -> str: + return re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", name).strip() + + +def detect_platform(url: str) -> str: + u = url.lower() + if "spotify.com" in u: + return "spotify" + if "music.apple.com" in u: + return "apple_music" + return "youtube" + + +def _extract_spotify_id(url: str) -> tuple[str, str]: + m = re.search( + r"spotify\.com/(?:[a-z]{2}/)?(?:intl-[a-z]+/)?(track|album|playlist)/([A-Za-z0-9]+)", + url, + ) + return (m.group(1), m.group(2)) if m else ("", "") + + +# --- anonymous token path --- + +def _get_anon_token() -> str: + global _ANON_TOKEN, _TOKEN_EXPIRY + if _ANON_TOKEN and time.time() < _TOKEN_EXPIRY: + return _ANON_TOKEN + resp = requests.get( + "https://open.spotify.com/get_access_token?reason=transport&productType=web_player", + headers={"User-Agent": _UA, "app-platform": "WebPlayer"}, + timeout=10, + ) + resp.raise_for_status() + data = resp.json() + _ANON_TOKEN = data["accessToken"] + _TOKEN_EXPIRY = data.get("accessTokenExpirationTimestampMs", 0) / 1000 - 30 + return _ANON_TOKEN + + +def _api(path: str) -> dict: + token = _get_anon_token() + resp = requests.get( + f"https://api.spotify.com/v1/{path}", + headers={"Authorization": f"Bearer {token}", "User-Agent": _UA}, + timeout=10, + ) + resp.raise_for_status() + return resp.json() + + +def _parse_api_track(track: dict, album_override: dict | None = None) -> dict: + album = album_override or track.get("album") or {} + images = album.get("images") or [] + cover = images[0]["url"] if images else None + artists = track.get("artists") or [] + artist = ", ".join(a["name"] for a in artists) + title = track.get("name", "") + return { + "title": title, + "artist": artist, + "album": album.get("name"), + "year": (album.get("release_date") or "")[:4], + "track_number": track.get("track_number"), + "cover_url": cover, + "search_query": f"{artist} - {title} audio", + "duration": (track.get("duration_ms") or 0) // 1000, + } + + +def _get_via_api(kind: str, sid: str) -> list: + if kind == "track": + return [_parse_api_track(_api(f"tracks/{sid}"))] + + if kind == "album": + album = _api(f"albums/{sid}") + results = [] + page = album.get("tracks", {}) + while True: + for t in page.get("items", []): + results.append(_parse_api_track(t, album)) + nxt = page.get("next") + if not nxt: + break + token = _get_anon_token() + page = requests.get(nxt, headers={"Authorization": f"Bearer {token}", "User-Agent": _UA}, timeout=10).json() + return results + + if kind == "playlist": + results = [] + page = _api(f"playlists/{sid}/tracks?limit=50") + while True: + for item in page.get("items", []): + t = item.get("track") + if t: + results.append(_parse_api_track(t)) + nxt = page.get("next") + if not nxt: + break + token = _get_anon_token() + page = requests.get(nxt, headers={"Authorization": f"Bearer {token}", "User-Agent": _UA}, timeout=10).json() + return results + + return [] + + +# --- embed page scraping fallback --- + +def _embed_url(kind: str, sid: str) -> str: + return f"https://open.spotify.com/embed/{kind}/{sid}" + + +def _scrape_embed(kind: str, sid: str) -> dict: + resp = requests.get( + _embed_url(kind, sid), + headers={"User-Agent": _UA, "Accept-Language": "en-US,en;q=0.9"}, + timeout=12, + ) + resp.raise_for_status() + m = re.search( + r'', + resp.text, + re.DOTALL, + ) + if not m: + raise RuntimeError("Could not parse Spotify embed page") + return json.loads(m.group(1))["props"]["pageProps"]["state"]["data"]["entity"] + + +def _best_cover(images: list) -> str | None: + if not images: + return None + return sorted(images, key=lambda x: x.get("maxWidth", 0), reverse=True)[0]["url"] + + +def _get_via_embed(kind: str, sid: str) -> list: + entity = _scrape_embed(kind, sid) + + if kind == "track": + artists = entity.get("artists") or [] + artist = ", ".join(a["name"] for a in artists) + title = entity.get("name") or entity.get("title", "") + year = (entity.get("releaseDate") or {}).get("isoString", "")[:4] + cover = _best_cover(entity.get("visualIdentity", {}).get("image", [])) + return [ + { + "title": title, + "artist": artist, + "album": None, + "year": year, + "track_number": None, + "cover_url": cover, + "search_query": f"{artist} - {title} audio", + "duration": (entity.get("duration") or 0) // 1000, + } + ] + + if kind in ("album", "playlist"): + album_name = entity.get("name") or entity.get("title", "") + album_artist = entity.get("subtitle", "") + year = (entity.get("releaseDate") or {}).get("isoString", "")[:4] + cover = _best_cover(entity.get("visualIdentity", {}).get("image", [])) + track_list = entity.get("trackList") or [] + results = [] + for t in track_list: + title = t.get("title") or t.get("name", "") + artist = t.get("subtitle") or album_artist + results.append( + { + "title": title, + "artist": artist, + "album": album_name if kind == "album" else None, + "year": year, + "track_number": None, + "cover_url": cover, + "search_query": f"{artist} - {title} audio", + "duration": (t.get("duration") or 0) // 1000, + } + ) + return results + + return [] + + +def get_spotify_metadata(url: str) -> list: + kind, sid = _extract_spotify_id(url) + if not sid: + raise RuntimeError(f"Cannot parse Spotify URL: {url}") + + try: + return _get_via_api(kind, sid) + except Exception: + pass + + return _get_via_embed(kind, sid) + + +# --- Apple Music --- + +def _itunes_track(row: dict, album_name=None, artist_fallback=None, cover_fallback=None, year_fallback=None) -> dict: + cover = (row.get("artworkUrl100") or cover_fallback or "").replace("100x100bb", "600x600bb") + artist = row.get("artistName") or artist_fallback or "" + title = row.get("trackName") or row.get("collectionName") or "" + return { + "title": title, + "artist": artist, + "album": row.get("collectionName") or album_name, + "year": (row.get("releaseDate") or year_fallback or "")[:4], + "track_number": row.get("trackNumber"), + "cover_url": cover, + "search_query": f"{artist} - {title} audio", + "duration": (row.get("trackTimeMillis") or 0) // 1000, + } + + +def _itunes_lookup(track_id: str) -> list: + try: + resp = requests.get(f"https://itunes.apple.com/lookup?id={track_id}", timeout=10) + resp.raise_for_status() + data = resp.json() + if data.get("resultCount"): + return [_itunes_track(data["results"][0])] + except Exception: + pass + return [] + + +def _itunes_album(album_id: str) -> list: + try: + resp = requests.get(f"https://itunes.apple.com/lookup?id={album_id}&entity=song", timeout=10) + resp.raise_for_status() + data = resp.json() + except Exception: + return [] + + album_name = artist_name = cover_url = year = None + results = [] + for r in data.get("results", []): + if r.get("wrapperType") == "collection": + album_name = r.get("collectionName") + artist_name = r.get("artistName") + cover_url = (r.get("artworkUrl100") or "").replace("100x100bb", "600x600bb") + year = (r.get("releaseDate") or "")[:4] + elif r.get("wrapperType") == "track": + results.append(_itunes_track(r, album_name, artist_name, cover_url, year)) + return results + + +def _itunes_search_from_url(url: str) -> list: + path = re.sub(r"https?://[^/]+", "", url.split("?")[0]) + parts = [p for p in path.split("/") if p and not re.fullmatch(r"\d+", p) and p not in ("us","ru","gb","de","fr","jp","kr","cn","au","ca","it","es","pl","tr","ua","kz","by","am","ge","az","uz")] + query = " ".join(parts[-3:]).replace("-", " ") + if not query.strip(): + return [] + try: + resp = requests.get( + f"https://itunes.apple.com/search?term={requests.utils.quote(query)}&media=music&limit=1", + timeout=10, + ) + resp.raise_for_status() + data = resp.json() + if data.get("resultCount"): + return [_itunes_track(data["results"][0])] + except Exception: + pass + return [] + + +def get_apple_music_metadata(url: str) -> list: + m = re.search(r"[?&]i=(\d+)", url) + if m: + r = _itunes_lookup(m.group(1)) + if r: + return r + + for pattern in (r"/song/[^/?#]+/(\d+)", r"/music-video/[^/?#]+/(\d+)", r"/album/[^/?#]+/(\d+)"): + m = re.search(pattern, url) + if m: + sid = m.group(1) + if "/album/" in pattern: + r = _itunes_album(sid) + else: + r = _itunes_lookup(sid) + if r: + return r + + for m in re.finditer(r"/(\d{6,12})(?:[/?#]|$)", url): + r = _itunes_lookup(m.group(1)) + if r: + return r + r = _itunes_album(m.group(1)) + if r: + return r + + return _itunes_search_from_url(url) + + +# --- yt-dlp options --- + +def _build_opts(fmt: str, quality: str, output_dir: str, meta: dict | None = None, embed_thumb: bool = True) -> dict: + if meta and meta.get("title") and meta.get("artist"): + stem = f"{_sanitize_filename(meta['artist'])} - {_sanitize_filename(meta['title'])}" + else: + stem = "%(title)s" + + base = { + "outtmpl": str(Path(output_dir) / f"{stem}.%(ext)s"), + "quiet": True, + "no_warnings": True, + "ignoreerrors": False, + } + + ffmpeg = _ffmpeg_path() + if ffmpeg: + base["ffmpeg_location"] = ffmpeg + + if fmt == "mp4": + base.update( + { + "format": "bestvideo[ext=mp4]+bestaudio[ext=m4a]/bestvideo+bestaudio/best", + "merge_output_format": "mp4", + "writethumbnail": embed_thumb, + "postprocessors": [ + {"key": "FFmpegVideoConvertor", "preferedformat": "mp4"}, + {"key": "FFmpegMetadata"}, + ] + + ([{"key": "EmbedThumbnail"}] if embed_thumb else []), + } + ) + else: + codec_map = {"mp3": "mp3", "flac": "flac", "wav": "wav", "m4a": "m4a", "alac": "alac", "ogg": "vorbis", "opus": "opus"} + quality_map = {"320": "320", "256": "256", "192": "192", "128": "128", "best": "0", "lossless": "0"} + codec = codec_map.get(fmt, "mp3") + q = quality_map.get(quality, "0") + ytdl_embed = embed_thumb and fmt in ("mp3", "m4a", "opus") + base.update( + { + "format": "bestaudio/best", + "writethumbnail": embed_thumb, + "postprocessors": [ + {"key": "FFmpegExtractAudio", "preferredcodec": codec, "preferredquality": q}, + {"key": "FFmpegMetadata"}, + ] + + ([{"key": "EmbedThumbnail"}] if ytdl_embed else []), + } + ) + + return base + + +# --- download helpers --- + +def _snapshot_dir(directory: str) -> set: + result = set() + p = Path(directory) + if not p.exists(): + return result + for f in p.iterdir(): + if f.is_file() and f.suffix.lstrip(".").lower() in AUDIO_EXTENSIONS: + result.add(str(f)) + return result + + +def _apply_thumb_file(audio_file: str, thumb_path: str): + import io + from PIL import Image + + try: + img = Image.open(thumb_path).convert("RGB") + buf = io.BytesIO() + img.save(buf, format="JPEG", quality=95) + cover_data = buf.getvalue() + except Exception: + return + + ext = Path(audio_file).suffix.lower() + try: + if ext == ".flac": + from mutagen.flac import FLAC, Picture + audio = FLAC(audio_file) + audio.clear_pictures() + pic = Picture() + pic.type = 3 + pic.mime = "image/jpeg" + pic.data = cover_data + audio.add_picture(pic) + audio.save() + elif ext == ".wav": + from mutagen.id3 import ID3, APIC, ID3NoHeaderError + try: + tags = ID3(audio_file) + except ID3NoHeaderError: + tags = ID3() + tags.delall("APIC") + tags["APIC"] = APIC(encoding=3, mime="image/jpeg", type=3, desc="Cover", data=cover_data) + tags.save(audio_file) + elif ext == ".ogg": + import base64 + from mutagen.oggvorbis import OggVorbis + from mutagen.flac import Picture + audio = OggVorbis(audio_file) + pic = Picture() + pic.type = 3 + pic.mime = "image/jpeg" + pic.data = cover_data + audio["metadata_block_picture"] = [base64.b64encode(pic.write()).decode("ascii")] + audio.save() + except Exception: + pass + + +def _cleanup_thumbs(output_dir: str): + p = Path(output_dir) + if not p.exists(): + return + for f in p.iterdir(): + if f.is_file() and f.suffix.lstrip(".").lower() in THUMB_EXTENSIONS: + try: + f.unlink() + except Exception: + pass + + +def _embed_saved_thumb(audio_file: str, output_dir: str): + stem = Path(audio_file).stem + for ext in THUMB_EXTENSIONS: + thumb = Path(output_dir) / f"{stem}.{ext}" + if thumb.exists(): + _apply_thumb_file(audio_file, str(thumb)) + try: + thumb.unlink() + except Exception: + pass + return + + +def _download_url(url: str, opts: dict, output_dir: str, fmt: str = "") -> list: + before = _snapshot_dir(output_dir) + with yt_dlp.YoutubeDL(opts) as ydl: + try: + ydl.download([url]) + except yt_dlp.utils.DownloadError as e: + if "Postprocessing" not in str(e): + raise RuntimeError(str(e)) from e + after = _snapshot_dir(output_dir) + new_files = list(after - before) + + if fmt in ("flac", "wav", "ogg"): + for f in new_files: + _embed_saved_thumb(f, output_dir) + + _cleanup_thumbs(output_dir) + return new_files + + +# --- public API --- + +def download_track(url: str, fmt: str, quality: str, output_dir: str, progress_callback=None) -> list: + Path(output_dir).mkdir(parents=True, exist_ok=True) + platform = detect_platform(url) + + if platform == "spotify": + tracks = get_spotify_metadata(url) + if not tracks: + raise RuntimeError("Could not retrieve Spotify metadata.") + return _download_tracks_list(tracks, fmt, quality, output_dir, progress_callback) + + if platform == "apple_music": + tracks = get_apple_music_metadata(url) + if not tracks: + raise RuntimeError("Could not retrieve Apple Music metadata.") + return _download_tracks_list(tracks, fmt, quality, output_dir, progress_callback) + + opts = _build_opts(fmt, quality, output_dir, embed_thumb=True) + if progress_callback: + opts["progress_hooks"] = [progress_callback] + return _download_url(url, opts, output_dir, fmt) + + +def _download_tracks_list(tracks: list, fmt: str, quality: str, output_dir: str, progress_callback=None) -> list: + from converter import embed_cover_and_metadata + + downloaded = [] + for track in tracks: + opts = _build_opts(fmt, quality, output_dir, meta=track, embed_thumb=False) + if progress_callback: + opts["progress_hooks"] = [progress_callback] + new_files = _download_url(f"ytsearch1:{track['search_query']}", opts, output_dir, fmt) + for f in new_files: + embed_cover_and_metadata(f, track) + downloaded.append(f) + + return downloaded