""" Musicanaz Backend API — v9 (HuggingFace Spaces · 16 GB RAM · 2 vCPU) ======================================================================= CHANGES IN v9 ───────────────────────────────────────────────────────────────────── REMOVED • Download feature (yt-dlp / download_bp.py) — fully stripped • Last.fm — removed; it requires a paid plan for geo charts ADDED — FREE TRENDING SOURCE • Spotify Web API (Client Credentials, 100% free, no card needed) – Featured playlists + top tracks by country – Set SPOTIFY_CLIENT_ID + SPOTIFY_CLIENT_SECRET env vars – Token auto-refreshed every hour; falls back gracefully if unset – New endpoint: GET /spotify/trending?country=IN&limit=50 10-HOUR CHARTS & TRENDING CACHE • _cache_charts: dedicated TTLCache (TTL = 36 000 s / 10 h) • On first request for a new country → fetch + cache • All subsequent requests for that country → served from cache instantly • Disk write-through (diskcache) → survives across uvicorn worker restarts within the same OS process • Per-country asyncio.Lock prevents thundering-herd: if 50 users request "IN" charts simultaneously, only ONE fetch fires; the rest wait for the result APPLE MUSIC PROXY • Dedicated `_get_apple_client()` always routes through proxy (when PROXY_USERNAME / PROXY_PASSWORD are set) SESSION / CONCURRENCY SAFETY • Per-country asyncio.Lock (charts + trending separately) • Global Semaphore caps concurrent external fetches to 20 • ThreadPoolExecutor stays at 8 workers (2 vCPU, I/O-bound YTM calls) • All mutable shared state protected by threading.Lock or asyncio.Lock LOGGING • RequestIDMiddleware tags every request with a short UUID prefix • Each external fetch logs source, country, count, latency • HTTPException handler logs path + status + detail • Full tracebacks logged at DEBUG level for unexpected exceptions • Structured format: HH:MM:SS LEVEL [req-id] message """ from __future__ import annotations import asyncio import base64 import logging import os import re import time import threading import traceback import uuid from collections import OrderedDict from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager from typing import Any, Dict, List, Optional import diskcache from curl_cffi import requests as cffi_requests from fastapi import FastAPI, HTTPException, Query, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware from fastapi.responses import JSONResponse from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.errors import RateLimitExceeded from slowapi.util import get_remote_address from starlette.middleware.base import BaseHTTPMiddleware from ytmusicapi import YTMusic # ───────────────────────────────────────────────────────────────────────────── # Logging — request-ID aware # ───────────────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-8s %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger("musicanaz") # ───────────────────────────────────────────────────────────────────────────── # Environment / proxy config # ───────────────────────────────────────────────────────────────────────────── _PROXY_HOST = "p.webshare.io" _PROXY_PORT = "80" _PROXY_USER = os.environ.get("PROXY_USERNAME", "") _PROXY_PASS = os.environ.get("PROXY_PASSWORD", "") _SPOTIFY_CLIENT_ID = os.environ.get("SPOTIFY_CLIENT_ID", "") _SPOTIFY_CLIENT_SECRET = os.environ.get("SPOTIFY_CLIENT_SECRET", "") DOCS_PASSWORD = os.environ.get("DOCS_PASSWORD", "prakamya05") def _build_proxies() -> dict | None: if not _PROXY_USER or not _PROXY_PASS: return None proxy_url = f"http://{_PROXY_USER}:{_PROXY_PASS}@{_PROXY_HOST}:{_PROXY_PORT}" return {"http": proxy_url, "https": proxy_url} _PROXIES = _build_proxies() _PROXY_URL = ( f"http://{_PROXY_USER}:{_PROXY_PASS}@{_PROXY_HOST}:{_PROXY_PORT}" if _PROXY_USER and _PROXY_PASS else None ) # ───────────────────────────────────────────────────────────────────────────── # Docs Auth Middleware # ───────────────────────────────────────────────────────────────────────────── class DocsAuthMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): if request.url.path in ("/docs", "/openapi.json", "/redoc"): auth = request.headers.get("Authorization", "") if auth.startswith("Basic "): try: decoded = base64.b64decode(auth[6:]).decode() _, password = decoded.split(":", 1) if password == DOCS_PASSWORD: return await call_next(request) except Exception: pass return Response( "Unauthorized", status_code=401, headers={"WWW-Authenticate": 'Basic realm="Musicanaz Docs"'}, ) return await call_next(request) # ───────────────────────────────────────────────────────────────────────────── # Request-ID Middleware — adds X-Request-ID header + per-request log line # ───────────────────────────────────────────────────────────────────────────── class RequestIDMiddleware(BaseHTTPMiddleware): """Stamps every request with a short ID; logs method / path / status / ms.""" async def dispatch(self, request: Request, call_next): rid = uuid.uuid4().hex[:8] request.state.request_id = rid t0 = time.monotonic() try: response = await call_next(request) except Exception as exc: elapsed = (time.monotonic() - t0) * 1000 log.error("[%s] %s %s → 500 (%.1f ms) — unhandled: %s", rid, request.method, request.url.path, elapsed, exc) log.debug("[%s] traceback:\n%s", rid, traceback.format_exc()) raise elapsed = (time.monotonic() - t0) * 1000 lvl = logging.WARNING if response.status_code >= 400 else logging.INFO log.log(lvl, "[%s] %s %s → %d (%.1f ms)", rid, request.method, request.url.path, response.status_code, elapsed) response.headers["X-Request-ID"] = rid return response # ───────────────────────────────────────────────────────────────────────────── # Circuit breaker (per external source) # ───────────────────────────────────────────────────────────────────────────── _CB_THRESHOLD = 5 _CB_RESET_SECS = 60 class CircuitBreaker: def __init__(self, name: str): self.name = name self._failures = 0 self._opened_at = 0.0 self._lock = threading.Lock() @property def available(self) -> bool: with self._lock: if self._failures < _CB_THRESHOLD: return True if time.monotonic() - self._opened_at > _CB_RESET_SECS: return True return False def success(self): with self._lock: self._failures = 0 self._opened_at = 0.0 def failure(self): with self._lock: self._failures += 1 if self._failures >= _CB_THRESHOLD: self._opened_at = time.monotonic() log.warning("Circuit breaker OPEN: %s", self.name) _cb_ytm = CircuitBreaker("ytmusicapi") _cb_itunes = CircuitBreaker("itunes") _cb_deezer = CircuitBreaker("deezer") _cb_spotify = CircuitBreaker("spotify") # ───────────────────────────────────────────────────────────────────────────── # Two-level cache (L1 in-memory LRU + L2 disk) # ───────────────────────────────────────────────────────────────────────────── _DISK_CACHE_DIR = os.environ.get("CACHE_DIR", "/tmp/ytm_cache") _DISK_CACHE_SIZE = 600 * 1024 * 1024 # 600 MB (16 GB headroom) _disk_cache = diskcache.Cache( _DISK_CACHE_DIR, size_limit=_DISK_CACHE_SIZE, eviction_policy="least-recently-used", statistics=True, ) _DISK_TTL_SHORT = 300 _DISK_TTL_MEDIUM = 1_800 _DISK_TTL_LONG = 21_600 _DISK_TTL_CHARTS = 36_000 # 10 h — matches in-memory chart TTL class TTLCache: """Thread-safe LRU in-memory cache with per-entry TTL + disk write-through.""" def __init__(self, maxsize: int = 256, ttl: int = 300, disk_ttl: int = 0): self._store: OrderedDict[str, tuple[Any, float]] = OrderedDict() self._maxsize = maxsize self._ttl = ttl self._disk_ttl = disk_ttl self._lock = threading.Lock() def get(self, key: str) -> Any: with self._lock: if key in self._store: value, expires = self._store[key] if time.monotonic() <= expires: self._store.move_to_end(key) return value del self._store[key] if self._disk_ttl: try: value = _disk_cache.get(key) if value is not None: self.set(key, value) return value except Exception: pass return None def set(self, key: str, value: Any, ttl: int | None = None) -> None: mem_ttl = ttl or self._ttl with self._lock: if key in self._store: self._store.move_to_end(key) self._store[key] = (value, time.monotonic() + mem_ttl) while len(self._store) > self._maxsize: self._store.popitem(last=False) if self._disk_ttl: try: _disk_cache.set(key, value, expire=self._disk_ttl) except Exception: pass def delete(self, key: str) -> None: with self._lock: self._store.pop(key, None) try: _disk_cache.delete(key) except Exception: pass def __len__(self) -> int: with self._lock: return len(self._store) # 16 GB optimised — generous sizes _cache_short = TTLCache(maxsize=192, ttl=120, disk_ttl=_DISK_TTL_SHORT) _cache_medium = TTLCache(maxsize=288, ttl=600, disk_ttl=_DISK_TTL_MEDIUM) _cache_long = TTLCache(maxsize=480, ttl=3_600, disk_ttl=_DISK_TTL_LONG) # ↓ Charts & trending: one fetch per country every 10 hours _cache_charts = TTLCache(maxsize=500, ttl=36_000, disk_ttl=_DISK_TTL_CHARTS) # ───────────────────────────────────────────────────────────────────────────── # Shared async HTTP clients # • _get_http_client() — general purpose (uses proxy when configured) # • _get_apple_client() — Apple Music RSS; explicitly routes via proxy # ───────────────────────────────────────────────────────────────────────────── _BROWSER_UA = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ) _http_client: cffi_requests.AsyncSession | None = None _apple_client: cffi_requests.AsyncSession | None = None def _make_client(proxy: str | None = None) -> cffi_requests.AsyncSession: proxies = {"http": proxy, "https": proxy} if proxy else None return cffi_requests.AsyncSession( impersonate="chrome", timeout=20.0, proxies=proxies, ) def _get_http_client() -> cffi_requests.AsyncSession: global _http_client if _http_client is None or getattr(_http_client, "_closed", False): _http_client = _make_client(_PROXY_URL) return _http_client def _get_apple_client() -> cffi_requests.AsyncSession: """Always routes through proxy (when configured) for Apple Music RSS.""" global _apple_client if _apple_client is None or getattr(_apple_client, "_closed", False): _apple_client = _make_client(_PROXY_URL) # proxy if set, plain if not return _apple_client # ───────────────────────────────────────────────────────────────────────────── # Global concurrency gate — caps simultaneous outbound fetches # ───────────────────────────────────────────────────────────────────────────── # Created lazily in async context (needs a running event loop) _fetch_semaphore: asyncio.Semaphore | None = None def _get_semaphore() -> asyncio.Semaphore: global _fetch_semaphore if _fetch_semaphore is None: _fetch_semaphore = asyncio.Semaphore(20) return _fetch_semaphore # ───────────────────────────────────────────────────────────────────────────── # Per-country fetch locks — prevents thundering herd on cache miss # One lock dict for charts, one for trending (they fetch independently) # ───────────────────────────────────────────────────────────────────────────── _chart_locks: Dict[str, asyncio.Lock] = {} _trending_locks: Dict[str, asyncio.Lock] = {} def _chart_lock(country: str) -> asyncio.Lock: key = country.upper() if key not in _chart_locks: _chart_locks[key] = asyncio.Lock() return _chart_locks[key] def _trending_lock(country: str) -> asyncio.Lock: key = country.upper() if key not in _trending_locks: _trending_locks[key] = asyncio.Lock() return _trending_locks[key] # ───────────────────────────────────────────────────────────────────────────── # Rate limiter # ───────────────────────────────────────────────────────────────────────────── limiter = Limiter(key_func=get_remote_address, default_limits=["300/minute"]) # ───────────────────────────────────────────────────────────────────────────── # Lifespan # ───────────────────────────────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): log.info("═══ Musicanaz API v9 starting ═══") log.info("Proxy: %s", "enabled" if _PROXY_URL else "disabled") log.info("Spotify: %s", "configured" if _SPOTIFY_CLIENT_ID else "not configured") asyncio.create_task(_background_warmup()) asyncio.create_task(_auto_refresh_charts()) yield log.info("═══ Musicanaz API v9 shutting down ═══") for client in (_http_client, _apple_client): if client and not client.is_closed: await client.aclose() _disk_cache.close() # ───────────────────────────────────────────────────────────────────────────── # App # ───────────────────────────────────────────────────────────────────────────── _START_TIME = time.monotonic() app = FastAPI( title="Musicanaz API — v9", description=( "ytmusicapi + Apple Music + Deezer + Spotify. " "Per-country 10-hour chart cache. 2-vCPU / 16 GB optimised." ), version="9.0.0", lifespan=lifespan, ) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware(RequestIDMiddleware) app.add_middleware(DocsAuthMiddleware) app.add_middleware(GZipMiddleware, minimum_size=300) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST", "DELETE"], allow_headers=["*"], ) # 8 workers — 2 vCPU, ytmusicapi is I/O-bound _executor = ThreadPoolExecutor(max_workers=8) async def run(fn, *args, **kwargs): loop = asyncio.get_running_loop() def _safe(): try: return fn(*args, **kwargs) except StopIteration as exc: raise RuntimeError(f"ytmusicapi StopIteration: {exc}") from exc return await loop.run_in_executor(_executor, _safe) # ───────────────────────────────────────────────────────────────────────────── # Per-locale YTMusic instance pool (LRU, capped at 20) # ───────────────────────────────────────────────────────────────────────────── _ytm_pool: Dict[str, YTMusic] = {} _ytm_pool_od: OrderedDict = OrderedDict() _ytm_lock = threading.Lock() _YTM_POOL_MAX = 20 def get_ytm(country: str = "ZZ", language: str = "en") -> YTMusic: country = (country or "ZZ").upper().strip() language = (language or "en").lower().strip() key = f"{country}:{language}" with _ytm_lock: if key in _ytm_pool: _ytm_pool_od.move_to_end(key, last=True) return _ytm_pool[key] if len(_ytm_pool) >= _YTM_POOL_MAX: oldest_key, _ = _ytm_pool_od.popitem(last=False) _ytm_pool.pop(oldest_key, None) try: session = cffi_requests.Session(impersonate="chrome", proxies=_PROXIES) location = country if country != "ZZ" else "" instance = YTMusic(language=language, location=location) instance._session = session except Exception: try: session = cffi_requests.Session(impersonate="chrome", proxies=_PROXIES) instance = YTMusic() instance._session = session except Exception as exc: raise RuntimeError(f"YTMusic init failed: {exc}") from exc _ytm_pool[key] = instance _ytm_pool_od[key] = True log.info("YTMusic instance created: locale=%s", key) return instance # ───────────────────────────────────────────────────────────────────────────── # Up-Next store # ───────────────────────────────────────────────────────────────────────────── _upnext_store: OrderedDict[str, Dict] = OrderedDict() _upnext_lock = threading.Lock() _UPNEXT_TTL = 7_200 _UPNEXT_MAX = 100 # ───────────────────────────────────────────────────────────────────────────── # Thumbnail helpers # ───────────────────────────────────────────────────────────────────────────── _YT_QUALITY_RANK: Dict[str, int] = { "maxresdefault": 100, "sddefault": 70, "0": 65, "hqdefault": 50, "mqdefault": 30, "2": 20, "1": 15, "3": 10, "default": 5, } _LH3_SIZE_RE = re.compile(r"=(w\d+(-h\d+)?|h\d+|s\d+)(-[a-zA-Z0-9_\-]*)*$") def upgrade_thumbnail_url(url: str) -> str: if not url: return url try: if "lh3.googleusercontent.com" in url: url = _LH3_SIZE_RE.sub("", url) return url + "=w576-h576-l90-rj" if "i.ytimg.com/vi/" in url: url = re.sub( r"/(maxresdefault|sddefault|hqdefault|mqdefault|default|[0-3])\.jpg", "/maxresdefault.jpg", url, ) return url except Exception: pass return url def _thumb_score(t: Any) -> int: if isinstance(t, str): url, w, h = t, 0, 0 else: url = t.get("url", "") w = int(t.get("width", 0) or 0) h = int(t.get("height", 0) or 0) if w > 0 and h > 0: return w * h m = re.search(r"=w(\d+)", url) if m: side = int(m.group(1)) return side * side try: fname = url.rsplit("/", 1)[-1].split("?")[0].split(".")[0] rank = _YT_QUALITY_RANK.get(fname) if rank: return rank * 10_000 except Exception: pass return 0 def best_thumbnails_list(raw: Any) -> list: if not raw: return [] if isinstance(raw, str): raw = [{"url": raw, "width": 0, "height": 0}] if isinstance(raw, dict): raw = [raw] if not isinstance(raw, list): return [] cleaned = [] for t in raw: if isinstance(t, str): t = {"url": t, "width": 0, "height": 0} url = t.get("url", "") if isinstance(t, dict) else "" if not url: continue upgraded = upgrade_thumbnail_url(url) cleaned.append({ "url": upgraded, "width": int(t.get("width", 0) or 0), "height": int(t.get("height", 0) or 0), }) if not cleaned: return [] cleaned.sort(key=_thumb_score, reverse=True) return cleaned # ───────────────────────────────────────────────────────────────────────────── # Normalizers # ───────────────────────────────────────────────────────────────────────────── def _norm_artists(raw: Any) -> list: if not raw: return [] if isinstance(raw, str): return [{"name": raw, "id": ""}] if isinstance(raw, dict): return [{"name": raw.get("name", "") or raw.get("artist", ""), "id": raw.get("id", "")}] if isinstance(raw, list): out = [] for a in raw: if isinstance(a, dict): out.append({"name": a.get("name", "") or a.get("artist", ""), "id": a.get("id", "") or a.get("browseId", "")}) elif isinstance(a, str): out.append({"name": a, "id": ""}) return out return [] def norm_track(t: dict) -> dict: raw = t.get("thumbnails") or t.get("thumbnail") or [] if isinstance(raw, str): raw = [{"url": raw, "width": 0, "height": 0}] thumbs = best_thumbnails_list(raw) album = t.get("album") album_name = album.get("name", "") if isinstance(album, dict) else (album or "") return { "videoId": t.get("videoId", ""), "title": t.get("title", ""), "artists": _norm_artists(t.get("artists") or t.get("artist")), "album": album_name, "duration": t.get("duration", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", "isExplicit": t.get("isExplicit", False), "year": t.get("year", ""), "source": t.get("source", "ytm"), } def norm_artist_result(a: dict) -> dict: thumbs = best_thumbnails_list(a.get("thumbnails") or []) return { "browseId": a.get("browseId", "") or a.get("channelId", ""), "name": a.get("artist", "") or a.get("name", "") or a.get("title", ""), "subscribers": a.get("subscribers", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", } def norm_album_result(a: dict) -> dict: thumbs = best_thumbnails_list(a.get("thumbnails") or []) return { "browseId": a.get("browseId", ""), "title": a.get("title", ""), "artists": _norm_artists(a.get("artists")), "year": a.get("year", ""), "type": a.get("type", "Album"), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", } def norm_playlist_result(p: dict) -> dict: thumbs = best_thumbnails_list(p.get("thumbnails") or []) return { "browseId": p.get("browseId", "") or p.get("playlistId", ""), "title": p.get("title", ""), "author": p.get("author", ""), "itemCount": p.get("itemCount", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", } def norm_podcast_result(p: dict) -> dict: thumbs = best_thumbnails_list(p.get("thumbnails") or []) browse_id = p.get("browseId") or p.get("podcastId") or p.get("channelId") or "" author = p.get("author", "") or ", ".join( a.get("name", "") for a in _norm_artists(p.get("artists")) ) return { "browseId": browse_id, "title": p.get("title", ""), "author": author, "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", } def norm_search_results(raw: list, filter_type: str | None) -> list: if not isinstance(raw, list): return [] out = [] for item in raw: if not isinstance(item, dict): continue rt = (item.get("resultType") or filter_type or "").lower() if rt in ("song", "songs", "video", "videos"): n = norm_track(item) n["resultType"] = "video" if "video" in rt else "song" out.append(n) elif rt in ("artist", "artists"): n = norm_artist_result(item); n["resultType"] = "artist"; out.append(n) elif rt in ("album", "albums", "single", "singles", "ep"): n = norm_album_result(item); n["resultType"] = "album"; out.append(n) elif rt in ("playlist", "playlists"): n = norm_playlist_result(item); n["resultType"] = "playlist"; out.append(n) elif rt in ("podcast", "podcasts", "episode", "episodes"): n = norm_podcast_result(item); n["resultType"] = "podcast"; out.append(n) else: thumbs = best_thumbnails_list(item.get("thumbnails") or []) item["thumbnails"] = thumbs item["thumbnail"] = thumbs[0]["url"] if thumbs else "" out.append(item) return out def _normalise_home(raw: list) -> list: shelves = [] for shelf in raw: if not isinstance(shelf, dict): continue contents = [] for item in (shelf.get("contents") or []): if not isinstance(item, dict): continue if item.get("videoId"): contents.append(norm_track(item)) else: thumbs = best_thumbnails_list(item.get("thumbnails") or []) item["thumbnails"] = thumbs item["thumbnail"] = thumbs[0]["url"] if thumbs else "" contents.append(item) if contents: shelves.append({"title": shelf.get("title", "For You"), "contents": contents}) return shelves def _extract_chart_section(section: Any) -> list: if not section: return [] items = section if isinstance(section, list) else ( section.get("items") or section.get("results") or [] ) out = [] for item in items: if not isinstance(item, dict): continue thumbs = best_thumbnails_list(item.get("thumbnails") or []) item["thumbnails"] = thumbs item["thumbnail"] = thumbs[0]["url"] if thumbs else "" out.append(item) return out def _normalise_charts(raw: dict, country: str) -> dict: return { "country": country, "songs": [norm_track(t) for t in _extract_chart_section(raw.get("songs"))], "videos": [norm_track(t) for t in _extract_chart_section(raw.get("videos")) if t.get("videoId")], "artists": [norm_artist_result(a) for a in _extract_chart_section(raw.get("artists"))], "trending": [norm_track(t) for t in _extract_chart_section(raw.get("trending")) if t.get("videoId")], } def _flatten_mood_categories(raw: Any) -> list: categories: list = [] if isinstance(raw, dict): for section_title, items in raw.items(): if isinstance(items, list): for item in items: if not isinstance(item, dict): continue thumbs = best_thumbnails_list(item.get("thumbnails") or []) categories.append({ "title": item.get("title", ""), "params": item.get("params", ""), "section": section_title, "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", }) elif isinstance(raw, list): for item in raw: if not isinstance(item, dict): continue thumbs = best_thumbnails_list(item.get("thumbnails") or []) categories.append({ "title": item.get("title", ""), "params": item.get("params", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", }) return categories # ───────────────────────────────────────────────────────────────────────────── # External API helpers — Apple Music # ───────────────────────────────────────────────────────────────────────────── _ITUNES_COUNTRY_FALLBACK = "us" async def _fetch_itunes_trending(country: str, limit: int = 50) -> list: """Apple Music RSS top songs — country-aware, free, uses dedicated proxy client.""" if not _cb_itunes.available: log.debug("Apple Music circuit breaker open — skipping fetch") return [] c = country.lower() if country and country not in ("ZZ", "zz") else _ITUNES_COUNTRY_FALLBACK url = f"https://rss.applemarketingtools.com/api/v2/{c}/music/most-played/{limit}/songs.json" t0 = time.monotonic() try: async with _get_semaphore(): resp = await _get_apple_client().get(url) resp.raise_for_status() results = resp.json().get("feed", {}).get("results") or [] tracks = [] for i, item in enumerate(results): art_url = re.sub(r"\d+x\d+bb", "600x600bb", item.get("artworkUrl100", "")) tracks.append({ "videoId": "", "title": item.get("name", ""), "artists": [{"name": item.get("artistName", ""), "id": ""}], "album": item.get("collectionName", ""), "duration": "", "thumbnails": [{"url": art_url, "width": 600, "height": 600}] if art_url else [], "thumbnail": art_url, "isExplicit": item.get("contentAdvisoryRating", "") == "Explicit", "year": item.get("releaseDate", "")[:4], "rank": i + 1, "source": "apple_music", "genres": [g.get("name", "") for g in item.get("genres", []) if isinstance(g, dict)], "url": item.get("url", ""), }) elapsed = (time.monotonic() - t0) * 1000 _cb_itunes.success() log.info("Apple Music: %d tracks for country=%s (%.0f ms)", len(tracks), c, elapsed) return tracks except Exception as exc: _cb_itunes.failure() log.warning("Apple Music fetch FAILED country=%s: %s", c, exc) log.debug("Apple Music traceback:\n%s", traceback.format_exc()) return [] async def _fetch_itunes_top_albums(country: str, limit: int = 20) -> list: """Apple Music RSS top albums.""" c = country.lower() if country and country not in ("ZZ", "zz") else "us" url = f"https://rss.applemarketingtools.com/api/v2/{c}/music/most-played/{limit}/albums.json" try: async with _get_semaphore(): resp = await _get_apple_client().get(url) resp.raise_for_status() results = resp.json().get("feed", {}).get("results") or [] albums = [] for i, item in enumerate(results): art_url = re.sub(r"\d+x\d+bb", "600x600bb", item.get("artworkUrl100", "")) albums.append({ "browseId": "", "title": item.get("name", ""), "artists": [{"name": item.get("artistName", ""), "id": ""}], "year": item.get("releaseDate", "")[:4], "thumbnail": art_url, "thumbnails":[{"url": art_url, "width": 600, "height": 600}] if art_url else [], "rank": i + 1, "source": "apple_music", "url": item.get("url", ""), }) return albums except Exception as exc: log.warning("Apple Music albums FAILED: %s", exc) return [] # ───────────────────────────────────────────────────────────────────────────── # External API helpers — Deezer # ───────────────────────────────────────────────────────────────────────────── async def _fetch_deezer_charts(limit: int = 50) -> dict: """Deezer global charts — free, no API key needed.""" if not _cb_deezer.available: log.debug("Deezer circuit breaker open — skipping fetch") return {} t0 = time.monotonic() try: async with _get_semaphore(): resp = await _get_http_client().get( "https://api.deezer.com/chart/0/tracks", params={"limit": limit}, ) resp.raise_for_status() data = resp.json() tracks = [] for i, item in enumerate(data.get("data") or []): art = (item.get("album", {}).get("cover_xl") or item.get("album", {}).get("cover_big") or item.get("album", {}).get("cover_medium", "")) tracks.append({ "videoId": "", "title": item.get("title", ""), "artists": [{"name": item.get("artist", {}).get("name", ""), "id": ""}], "album": item.get("album", {}).get("title", ""), "duration": str(item.get("duration", "")), "thumbnails": [{"url": art, "width": 500, "height": 500}] if art else [], "thumbnail": art, "isExplicit": item.get("explicit_lyrics", False), "year": "", "rank": i + 1, "source": "deezer", "deezerId": str(item.get("id", "")), "preview": item.get("preview", ""), }) _cb_deezer.success() albums: list = [] artists: list = [] for t_list, endpoint in [(albums, "albums"), (artists, "artists")]: try: async with _get_semaphore(): r2 = await _get_http_client().get( f"https://api.deezer.com/chart/0/{endpoint}", params={"limit": 20}, ) r2.raise_for_status() for item in (r2.json().get("data") or []): pic = (item.get("cover_xl") or item.get("cover_big") or item.get("picture_xl") or item.get("picture_big") or item.get("picture_medium", "")) t_list.append({ "title": item.get("title") or item.get("name", ""), "artist": item.get("artist", {}).get("name", "") if endpoint == "albums" else "", "thumbnail": pic, "source": "deezer", "deezerId": str(item.get("id", "")), "url": item.get("link", ""), }) except Exception as exc: log.debug("Deezer %s fetch skipped: %s", endpoint, exc) elapsed = (time.monotonic() - t0) * 1000 log.info("Deezer: %d chart tracks (%.0f ms)", len(tracks), elapsed) return {"tracks": tracks, "albums": albums, "artists": artists} except Exception as exc: _cb_deezer.failure() log.warning("Deezer charts FAILED: %s", exc) log.debug("Deezer traceback:\n%s", traceback.format_exc()) return {} async def _fetch_deezer_genres() -> list: """Deezer genre list.""" if not _cb_deezer.available: return [] try: async with _get_semaphore(): resp = await _get_http_client().get("https://api.deezer.com/genre") resp.raise_for_status() genres = [] for g in (resp.json().get("data") or []): pic = g.get("picture_xl") or g.get("picture_big") or g.get("picture_medium", "") genres.append({ "id": g.get("id"), "title": g.get("name", ""), "thumbnail": pic, "thumbnails": [{"url": pic, "width": 500, "height": 500}] if pic else [], "source": "deezer", }) _cb_deezer.success() return genres except Exception as exc: _cb_deezer.failure() log.warning("Deezer genres FAILED: %s", exc) return [] async def _fetch_deezer_editorial_playlists(limit: int = 20) -> list: """Deezer editorial playlists (global).""" if not _cb_deezer.available: return [] try: async with _get_semaphore(): resp = await _get_http_client().get( "https://api.deezer.com/editorial/0/charts", params={"limit": limit}, ) resp.raise_for_status() data = resp.json() playlists = [] for item in (data.get("playlists", {}).get("data") or []): pic = item.get("picture_xl") or item.get("picture_big") or item.get("picture_medium", "") playlists.append({ "browseId": str(item.get("id", "")), "title": item.get("title", ""), "subtitle": item.get("description", ""), "thumbnail": pic, "thumbnails": [{"url": pic, "width": 500, "height": 500}] if pic else [], "source": "deezer", }) _cb_deezer.success() return playlists except Exception as exc: _cb_deezer.failure() log.warning("Deezer editorial FAILED: %s", exc) return [] # ───────────────────────────────────────────────────────────────────────────── # External API helpers — Spotify (Client Credentials, 100% free) # ───────────────────────────────────────────────────────────────────────────── # ── Spotify Android-client token state ─────────────────────────────────────── # We simulate the Spotify Android app. Two separate tokens are maintained: # _spotify_client_token — short-lived token from clienttoken.spotify.com # (proves we are a known Spotify client; no secret) # _spotify_access_token — Bearer token from accounts.spotify.com # (used in Authorization header; needs client_secret) # Both are refreshed automatically; all state protected by a threading.Lock. # ───────────────────────────────────────────────────────────────────────────── import secrets as _secrets _SPOTIFY_DEVICE_ID = _secrets.token_hex(16) # stable per-process, random across restarts # Android Spotify app version strings (update periodically if Spotify changes them) _SPOTIFY_ANDROID_VERSION = "8.6.96.470" _SPOTIFY_ANDROID_OS_VERSION = "30" # Android 11 _spotify_access_token: str = "" _spotify_access_expires: float = 0.0 _spotify_client_token: str = "" _spotify_client_token_expires: float = 0.0 _spotify_token_lock = threading.Lock() async def _refresh_spotify_client_token() -> str: """ Fetch a Spotify client-token via clienttoken.spotify.com. This endpoint accepts just the client_id (no secret) and returns a short-lived client-token that proves the caller is a known Spotify client application. Used identically by the real Spotify Android / iOS apps. """ global _spotify_client_token, _spotify_client_token_expires if not _SPOTIFY_CLIENT_ID: return "" payload = { "client_data": { "client_version": _SPOTIFY_ANDROID_VERSION, "client_id": _SPOTIFY_CLIENT_ID, "js_sdk_data": { "device_brand": "Google", "device_model": "sdk_gphone64_x86_64", "os": "android", "os_version": _SPOTIFY_ANDROID_OS_VERSION, "device_id": _SPOTIFY_DEVICE_ID, "device_type": "smartphone", }, } } try: async with _get_semaphore(): resp = await _get_http_client().post( "https://clienttoken.spotify.com/v1/clienttoken", json=payload, headers={"Accept": "application/json"}, ) resp.raise_for_status() data = resp.json() token = ( data.get("granted_token", {}).get("token") or data.get("client_token") or "" ) ttl = int( data.get("granted_token", {}).get("expires_after_seconds") or data.get("refresh_after_seconds") or 1_800 ) with _spotify_token_lock: _spotify_client_token = token _spotify_client_token_expires = time.monotonic() + ttl - 30 log.info("Spotify: client-token refreshed (ttl=%ds)", ttl) return token except Exception as exc: log.warning("Spotify client-token refresh FAILED: %s", exc) log.debug("Spotify client-token traceback:\n%s", traceback.format_exc()) return "" async def _get_spotify_client_token() -> str: with _spotify_token_lock: if _spotify_client_token and time.monotonic() < _spotify_client_token_expires: return _spotify_client_token return await _refresh_spotify_client_token() async def _refresh_spotify_access_token() -> str: """ Fetch a Spotify Bearer access token via client_credentials grant. Uses the provided SPOTIFY_CLIENT_ID + SPOTIFY_CLIENT_SECRET (Android/iOS app creds). Does NOT require Spotify Premium — gives access to all public catalog endpoints. """ global _spotify_access_token, _spotify_access_expires if not _SPOTIFY_CLIENT_ID or not _SPOTIFY_CLIENT_SECRET: return "" creds = base64.b64encode( f"{_SPOTIFY_CLIENT_ID}:{_SPOTIFY_CLIENT_SECRET}".encode() ).decode() try: async with _get_semaphore(): resp = await _get_http_client().post( "https://accounts.spotify.com/api/token", headers={ "Authorization": f"Basic {creds}", "Content-Type": "application/x-www-form-urlencoded", }, content=b"grant_type=client_credentials", ) resp.raise_for_status() data = resp.json() with _spotify_token_lock: _spotify_access_token = data["access_token"] _spotify_access_expires = time.monotonic() + data.get("expires_in", 3_600) - 30 log.info("Spotify: access token refreshed (expires in %ds)", data.get("expires_in", 3_600)) return _spotify_access_token except Exception as exc: log.warning("Spotify access token refresh FAILED: %s", exc) log.debug("Spotify token traceback:\n%s", traceback.format_exc()) return "" async def _get_spotify_tokens() -> tuple[str, str]: """Return (access_token, client_token). Either may be empty on failure.""" with _spotify_token_lock: access_ok = _spotify_access_token and time.monotonic() < _spotify_access_expires client_ok = _spotify_client_token and time.monotonic() < _spotify_client_token_expires access_task = (asyncio.sleep(0) if access_ok else _refresh_spotify_access_token()) client_task = (asyncio.sleep(0) if client_ok else _refresh_spotify_client_token()) await asyncio.gather(access_task, client_task, return_exceptions=True) with _spotify_token_lock: return _spotify_access_token, _spotify_client_token def _spotify_headers(access_token: str, client_token: str) -> dict: """Build the full set of headers the Spotify Android app sends.""" h = { "Authorization": f"Bearer {access_token}", "Accept": "application/json", "Accept-Language": "en", "app-platform": "Android", "spotify-app-version": _SPOTIFY_ANDROID_VERSION, "User-Agent": ( f"Spotify/{_SPOTIFY_ANDROID_VERSION} Android/{_SPOTIFY_ANDROID_OS_VERSION} " "(Google sdk_gphone64_x86_64)" ), } if client_token: h["client-token"] = client_token return h async def _fetch_spotify_trending(country: str, limit: int = 50) -> list: """ Fetch country trending tracks via the Spotify Android client approach. Disabled per user request. """ return [] c = country.upper() if country and country not in ("ZZ", "zz") else "US" access_token, client_token = await _get_spotify_tokens() if not access_token: log.warning("Spotify: no access token available, skipping trending for %s", c) return [] headers = _spotify_headers(access_token, client_token) t0 = time.monotonic() try: # ── 1. Featured playlists (country-localised) ────────────────────── async with _get_semaphore(): fp_resp = await _get_http_client().get( "https://api.spotify.com/v1/browse/featured-playlists", headers=headers, params={"country": c, "limit": 5}, ) fp_resp.raise_for_status() playlists = fp_resp.json().get("playlists", {}).get("items") or [] log.debug("Spotify: %d featured playlists for %s", len(playlists), c) tracks: list = [] per_playlist = max((limit // max(len(playlists), 1)) + 5, 15) for playlist in playlists[:4]: pid = playlist.get("id", "") if not pid: continue try: async with _get_semaphore(): tr_resp = await _get_http_client().get( f"https://api.spotify.com/v1/playlists/{pid}/tracks", headers=headers, params={ "limit": per_playlist, "market": c, "fields": ( "items(track(" "id,name,artists(name,id),album(name,images)," "duration_ms,explicit,popularity))" ), }, ) tr_resp.raise_for_status() for item in (tr_resp.json().get("items") or []): t = item.get("track") if not t or not isinstance(t, dict) or not t.get("id"): continue album = t.get("album") or {} images = album.get("images") or [] # pick largest image (Spotify orders by descending width) art = images[0]["url"] if images else "" tracks.append({ "videoId": "", "title": t.get("name", ""), "artists": [ {"name": a.get("name", ""), "id": a.get("id", "")} for a in (t.get("artists") or []) ], "album": album.get("name", ""), "duration": str(int(t.get("duration_ms", 0)) // 1_000), "thumbnails": [{"url": art, "width": 640, "height": 640}] if art else [], "thumbnail": art, "isExplicit": t.get("explicit", False), "year": "", "rank": len(tracks) + 1, "source": "spotify", "spotifyId": t.get("id", ""), "popularity": t.get("popularity", 0), }) except Exception as exc: log.debug("Spotify playlist %s track fetch skipped: %s", pid, exc) # ── 2. New releases as supplemental source ───────────────────────── if len(tracks) < limit // 2: try: async with _get_semaphore(): nr_resp = await _get_http_client().get( "https://api.spotify.com/v1/browse/new-releases", headers=headers, params={"country": c, "limit": 10}, ) nr_resp.raise_for_status() for album in (nr_resp.json().get("albums", {}).get("items") or []): images = album.get("images") or [] art = images[0]["url"] if images else "" tracks.append({ "videoId": "", "title": album.get("name", ""), "artists": [ {"name": a.get("name", ""), "id": a.get("id", "")} for a in (album.get("artists") or []) ], "album": album.get("name", ""), "duration": "", "thumbnails": [{"url": art, "width": 640, "height": 640}] if art else [], "thumbnail": art, "isExplicit": False, "year": (album.get("release_date") or "")[:4], "rank": len(tracks) + 1, "source": "spotify", "spotifyId": album.get("id", ""), "popularity": 0, }) except Exception as exc: log.debug("Spotify new-releases skipped: %s", exc) elapsed = (time.monotonic() - t0) * 1000 _cb_spotify.success() log.info("Spotify: %d tracks for country=%s (%.0f ms)", len(tracks), c, elapsed) return tracks[:limit] except Exception as exc: _cb_spotify.failure() log.warning("Spotify trending FAILED country=%s: %s", c, exc) log.debug("Spotify traceback:\n%s", traceback.format_exc()) return [] # ───────────────────────────────────────────────────────────────────────────── # Cache-Control header helper # ───────────────────────────────────────────────────────────────────────────── def _cc(seconds: int) -> dict: return {"Cache-Control": f"public, max-age={seconds}, stale-while-revalidate=60"} # ───────────────────────────────────────────────────────────────────────────── # Track deduplication # ───────────────────────────────────────────────────────────────────────────── def _deduplicate_tracks(tracks: list) -> list: seen, out = set(), [] for t in tracks: title = (t.get("title", "") or "").lower().strip() artist = "" arts = t.get("artists", []) if arts and isinstance(arts[0], dict): artist = (arts[0].get("name", "") or "").lower().strip() key = f"{title}|{artist}" if key and key not in seen: seen.add(key) out.append(t) return out # ───────────────────────────────────────────────────────────────────────────── # Chart merging # ───────────────────────────────────────────────────────────────────────────── def _merge_charts( country: str, ytm: dict, apple: Any, deezer: Any, spotify: Any, ) -> dict: """Merge chart data from all sources. Apple Music is primary when YTM is thin.""" apple_tracks = apple if isinstance(apple, list) else [] deezer_data = deezer if isinstance(deezer, dict) else {} spotify_tracks = spotify if isinstance(spotify, list) else [] deezer_tracks = deezer_data.get("tracks", []) ytm_songs = ytm.get("songs", []) ytm_videos = ytm.get("videos", []) ytm_artists = ytm.get("artists", []) ytm_trending = ytm.get("trending", []) primary_songs = apple_tracks if len(apple_tracks) >= 5 else (ytm_songs or apple_tracks) return { "country": country, "songs": primary_songs, "ytm_songs": ytm_songs, "videos": ytm_videos, "artists": ytm_artists, "trending": ytm_trending, "apple_music_top": apple_tracks, "deezer_top": deezer_tracks, "deezer_albums": deezer_data.get("albums", []), "deezer_artists": deezer_data.get("artists", []), "spotify_top": spotify_tracks, "cached_at": int(time.time()), "sources_used": { "ytm": len(ytm_songs) > 0, "apple": len(apple_tracks) > 0, "deezer": len(deezer_tracks) > 0, "spotify": len(spotify_tracks) > 0, }, } # ───────────────────────────────────────────────────────────────────────────── # Background warm-up & auto-refresh # ───────────────────────────────────────────────────────────────────────────── async def _do_fetch_charts(country: str, language: str) -> dict: """Core fetch function: YTM + Apple + Deezer + Spotify in parallel.""" log.info("Charts fetch: country=%s lang=%s", country, language) ytm_task = (run(get_ytm(country, language).get_charts, country) if _cb_ytm.available else asyncio.sleep(0)) apple_task = _fetch_itunes_trending(country, 50) deezer_task = _fetch_deezer_charts(50) spotify_task = _fetch_spotify_trending(country, 50) ytm_raw, apple_tracks, deezer_data, spotify_tracks = await asyncio.gather( ytm_task, apple_task, deezer_task, spotify_task, return_exceptions=True, ) ytm_result: dict = {} if isinstance(ytm_raw, Exception): _cb_ytm.failure() log.warning("YTMusic charts FAILED country=%s: %s", country, ytm_raw) elif isinstance(ytm_raw, dict) and ytm_raw: ytm_result = _normalise_charts(ytm_raw, country) _cb_ytm.success() result = _merge_charts(country, ytm_result, apple_tracks, deezer_data, spotify_tracks) key = f"charts10h:{country}:{language}" _cache_charts.set(key, result) log.info("Charts cached: country=%s (songs=%d, apple=%d, spotify=%d)", country, len(result.get("ytm_songs", [])), len(result.get("apple_music_top", [])), len(result.get("spotify_top", []))) return result async def _warm_home(country: str, language: str) -> None: try: raw = await run(get_ytm(country, language).get_home, 6) shelves = _normalise_home(raw or []) if shelves: _cache_medium.set(f"home:{country}:{language}:6", shelves) except Exception as exc: log.debug("_warm_home country=%s: %s", country, exc) async def _warm_moods(country: str, language: str) -> None: try: raw = await run(get_ytm(country, language).get_mood_categories) categories = _flatten_mood_categories(raw) if categories: _cache_medium.set(f"mood_categories:{country}:{language}", categories) except Exception as exc: log.debug("_warm_moods country=%s: %s", country, exc) async def _background_warmup(): """Pre-populate caches for common locales shortly after startup.""" await asyncio.sleep(5) log.info("Background warm-up starting …") warm_targets = [("ZZ", "en"), ("IN", "en"), ("US", "en"), ("GB", "en")] for country, lang in warm_targets: try: await _do_fetch_charts(country, lang) except Exception as exc: log.warning("Warm-up charts %s failed: %s", country, exc) try: await _warm_home(country, lang) except Exception: pass await asyncio.sleep(1) # small stagger to avoid burst try: await _warm_moods("ZZ", "en") except Exception: pass log.info("Background warm-up complete.") async def _auto_refresh_charts(): """Re-fetch all known countries' charts once every 10 hours.""" await asyncio.sleep(120) # let warm-up finish first while True: await asyncio.sleep(36_000) # 10 hours log.info("Auto-refresh: refreshing all cached chart countries …") # Refresh every country already known in the charts cache # (we track them via disk cache keys) refreshed = 0 try: for k in list(_disk_cache.iterkeys()): sk = str(k) if not sk.startswith("charts10h:"): continue parts = sk.split(":") if len(parts) == 3: _, c, lang = parts try: await _do_fetch_charts(c, lang) refreshed += 1 except Exception as exc: log.warning("Auto-refresh failed for %s/%s: %s", c, lang, exc) await asyncio.sleep(2) except Exception as exc: log.warning("Auto-refresh loop error: %s", exc) log.info("Auto-refresh complete: %d countries refreshed.", refreshed) # ───────────────────────────────────────────────────────────────────────────── # Routes — meta # ───────────────────────────────────────────────────────────────────────────── @app.get("/", include_in_schema=False) async def root(): return {"name": "Musicanaz API", "version": "9.0.0", "status": "ok", "docs": "/docs"} @app.get("/health", tags=["meta"]) async def health(): hits, misses = _disk_cache.stats() return { "status": "ok", "version": "9.0.0", "uptime_seconds": round(time.monotonic() - _START_TIME), "ytm_instances": len(_ytm_pool), "proxy_enabled": _PROXIES is not None, "spotify_enabled": bool(_SPOTIFY_CLIENT_ID), "circuit_breakers": { "ytm": "open" if not _cb_ytm.available else "closed", "itunes": "open" if not _cb_itunes.available else "closed", "deezer": "open" if not _cb_deezer.available else "closed", "spotify": "open" if not _cb_spotify.available else "closed", }, "chart_countries_cached": sum( 1 for k in (_disk_cache.iterkeys() if True else []) if str(k).startswith("charts10h:") ), "disk_cache": { "hits": hits, "misses": misses, "size_mb": round(_disk_cache.volume() / 1_048_576, 1), }, } @app.get("/cache_stats", tags=["meta"]) async def cache_stats(): hits, misses = _disk_cache.stats() return { "memory": { "short": len(_cache_short), "medium": len(_cache_medium), "long": len(_cache_long), "charts": len(_cache_charts), }, "disk": { "entries": len(_disk_cache), "size_mb": round(_disk_cache.volume() / 1_048_576, 1), "hits": hits, "misses": misses, "hit_rate": f"{hits/(hits+misses)*100:.1f}%" if (hits + misses) > 0 else "n/a", }, "upnext": len(_upnext_store), "ytm_pool": len(_ytm_pool), } @app.delete("/cache", tags=["meta"]) async def clear_cache(key_prefix: str = ""): cleared = 0 if key_prefix: for k in list(_disk_cache.iterkeys()): if str(k).startswith(key_prefix): _disk_cache.delete(k) cleared += 1 else: _disk_cache.clear() cleared = -1 log.info("Cache cleared: prefix=%r entries=%s", key_prefix or "(all)", cleared) return {"cleared": cleared if cleared >= 0 else "all"} # ───────────────────────────────────────────────────────────────────────────── # Search # ───────────────────────────────────────────────────────────────────────────── @app.get("/search", tags=["search"]) @limiter.limit("40/minute") async def search( request: Request, response: Response, query: str, filter: Optional[str] = Query(None), scope: Optional[str] = Query(None), limit: int = Query(20, ge=1, le=50), offset: int = Query(0, ge=0), ignore_spelling: bool = False, country: str = Query("ZZ"), language: str = Query("en"), ): fetch_limit = min(offset + limit, 50) cache_key = f"search:{query}:{filter}:{fetch_limit}:{country}:{language}" cached = _cache_short.get(cache_key) if cached is None: if not _cb_ytm.available: raise HTTPException(503, detail="YTMusic source temporarily unavailable") try: raw = await run( get_ytm(country, language).search, query, filter, scope, fetch_limit, ignore_spelling, ) cached = norm_search_results(raw or [], filter) _cache_short.set(cache_key, cached) _cb_ytm.success() except Exception as e: _cb_ytm.failure() log.error("Search failed q=%r: %s", query, e) raise HTTPException(500, detail=str(e)) response.headers.update(_cc(120)) return cached[offset:offset + limit] @app.get("/search_suggestions", tags=["search"]) @limiter.limit("40/minute") async def search_suggestions( request: Request, response: Response, query: str, detailed: bool = False, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"suggestions:{query}:{detailed}:{country}:{language}" cached = _cache_short.get(cache_key) if cached is not None: response.headers.update(_cc(60)) return cached try: data = await run(get_ytm(country, language).get_search_suggestions, query, detailed) _cache_short.set(cache_key, data or []) response.headers.update(_cc(60)) return data or [] except Exception as e: log.warning("Search suggestions failed q=%r: %s", query, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # Home feed # ───────────────────────────────────────────────────────────────────────────── @app.get("/home", tags=["discovery"]) @limiter.limit("30/minute") async def get_home( request: Request, response: Response, limit: int = Query(6, ge=1, le=15), country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"home:{country}:{language}:{limit}" cached = _cache_medium.get(cache_key) if cached is not None: response.headers.update(_cc(300)) return cached try: raw = await run(get_ytm(country, language).get_home, limit) shelves = _normalise_home(raw or []) except Exception as exc: log.warning("Home YTM failed country=%s: %s", country, exc) shelves = [] # Fallback: synthesise home from Apple Music + Spotify when YTM is empty if not shelves: apple_task = _fetch_itunes_trending(country, 20) spotify_task = _fetch_spotify_trending(country, 20) apple_tracks, spotify_tracks = await asyncio.gather( apple_task, spotify_task, return_exceptions=True ) if isinstance(apple_tracks, list) and apple_tracks: shelves.append({"title": "Top Songs", "contents": apple_tracks[:20]}) if isinstance(spotify_tracks, list) and spotify_tracks: shelves.append({"title": "Trending Worldwide", "contents": spotify_tracks[:20]}) _cache_medium.set(cache_key, shelves) response.headers.update(_cc(300)) return shelves # ───────────────────────────────────────────────────────────────────────────── # Charts — 10-hour per-country cache with stampede protection # ───────────────────────────────────────────────────────────────────────────── @app.get("/charts", tags=["discovery"]) @limiter.limit("30/minute") async def get_charts( request: Request, response: Response, country: str = Query("ZZ"), language: str = Query("en"), sources: str = Query("all", description="all | ytm | apple | deezer | spotify"), ): """ Country-aware charts from all sources. Cached per-country for 10 hours. A new country triggers one fetch; all concurrent callers wait on a per-country lock for the result. """ cache_key = f"charts10h:{country}:{language}" cached = _cache_charts.get(cache_key) if cached is not None: response.headers.update(_cc(36_000)) return cached # Per-country lock: only one fetch in-flight at a time async with _chart_lock(country): # Double-check after acquiring lock cached = _cache_charts.get(cache_key) if cached is not None: response.headers.update(_cc(36_000)) return cached try: result = await _do_fetch_charts(country, language) except Exception as exc: log.error("Charts fetch error country=%s: %s", country, exc) log.debug("Charts traceback:\n%s", traceback.format_exc()) raise HTTPException(500, detail=f"Charts fetch failed: {exc}") response.headers.update(_cc(36_000)) return result # ───────────────────────────────────────────────────────────────────────────── # Trending — 10-hour per-country cache with stampede protection # ───────────────────────────────────────────────────────────────────────────── @app.get("/trending", tags=["discovery"]) @limiter.limit("30/minute") async def get_trending( request: Request, response: Response, country: str = Query("ZZ"), language: str = Query("en"), limit: int = Query(25, ge=1, le=100), sources: str = Query("all", description="all | ytm | apple | deezer | spotify"), ): """ Trending songs merged from Apple Music + Deezer + Spotify (+YTM when warm). Cached per-country for 10 hours. First request per country triggers one fetch. """ cache_key = f"trending10h:{country}:{language}" cached = _cache_charts.get(cache_key) if cached is not None: response.headers.update(_cc(36_000)) # Return slice of limit from cached merged list out = dict(cached) out["merged"] = out.get("merged", [])[:limit] return out async with _trending_lock(country): cached = _cache_charts.get(cache_key) if cached is not None: response.headers.update(_cc(36_000)) out = dict(cached) out["merged"] = out.get("merged", [])[:limit] return out log.info("Trending fetch: country=%s lang=%s", country, language) use_apple = sources in ("all", "apple") use_deezer = sources in ("all", "deezer") use_spotify = sources in ("all", "spotify") use_ytm = sources in ("all", "ytm") apple_task = _fetch_itunes_trending(country, 50) if use_apple else asyncio.sleep(0) deezer_task = _fetch_deezer_charts(50) if use_deezer else asyncio.sleep(0) spotify_task = _fetch_spotify_trending(country, 50) if use_spotify else asyncio.sleep(0) # Use YTM from charts cache if already warm ytm_trending: list = [] if use_ytm and _cb_ytm.available: chart_cached = _cache_charts.get(f"charts10h:{country}:{language}") if chart_cached: ytm_trending = (chart_cached.get("trending") or chart_cached.get("ytm_songs") or []) apple_tracks, deezer_data, spotify_tracks = await asyncio.gather( apple_task, deezer_task, spotify_task, return_exceptions=True, ) apple_tracks = apple_tracks if isinstance(apple_tracks, list) else [] deezer_tracks = (deezer_data.get("tracks", []) if isinstance(deezer_data, dict) else []) spotify_tracks = spotify_tracks if isinstance(spotify_tracks, list) else [] merged = _deduplicate_tracks( apple_tracks[:50] + spotify_tracks[:50] + ytm_trending[:50] ) result = { "country": country, "trending": ytm_trending[:50], "apple_top": apple_tracks[:50], "deezer_top": deezer_tracks[:50], "spotify_top": spotify_tracks[:50], "merged": merged, "cached_at": int(time.time()), "sources_used": { "ytm": len(ytm_trending) > 0, "apple": len(apple_tracks) > 0, "deezer": len(deezer_tracks) > 0, "spotify": len(spotify_tracks) > 0, }, } _cache_charts.set(cache_key, result) log.info("Trending cached: country=%s merged=%d", country, len(merged)) out = dict(result) out["merged"] = out["merged"][:limit] response.headers.update(_cc(36_000)) return out # ───────────────────────────────────────────────────────────────────────────── # Apple Music direct endpoints # ───────────────────────────────────────────────────────────────────────────── @app.get("/apple_music/top_songs", tags=["apple_music"]) @limiter.limit("30/minute") async def apple_music_top_songs( request: Request, response: Response, country: str = Query("us"), limit: int = Query(50, ge=1, le=100), ): cache_key = f"am_songs:{country}:{limit}" cached = _cache_charts.get(cache_key) if cached is not None: response.headers.update(_cc(36_000)) return cached tracks = await _fetch_itunes_trending(country, limit) if not tracks: raise HTTPException(503, detail="Apple Music RSS unavailable") result = {"country": country, "tracks": tracks, "count": len(tracks)} _cache_charts.set(cache_key, result) response.headers.update(_cc(36_000)) return result @app.get("/apple_music/top_albums", tags=["apple_music"]) @limiter.limit("30/minute") async def apple_music_top_albums( request: Request, response: Response, country: str = Query("us"), limit: int = Query(20, ge=1, le=50), ): cache_key = f"am_albums:{country}:{limit}" cached = _cache_medium.get(cache_key) if cached is not None: response.headers.update(_cc(1_800)) return cached albums = await _fetch_itunes_top_albums(country, limit) if not albums: raise HTTPException(503, detail="Apple Music RSS unavailable") result = {"country": country, "albums": albums, "count": len(albums)} _cache_medium.set(cache_key, result) response.headers.update(_cc(1_800)) return result # ───────────────────────────────────────────────────────────────────────────── # Deezer direct endpoints # ───────────────────────────────────────────────────────────────────────────── @app.get("/deezer/charts", tags=["deezer"]) @limiter.limit("30/minute") async def deezer_charts_endpoint( request: Request, response: Response, limit: int = Query(50, ge=1, le=100), ): cache_key = f"deezer_charts:{limit}" cached = _cache_charts.get(cache_key) if cached is not None: response.headers.update(_cc(36_000)) return cached data = await _fetch_deezer_charts(limit) if not data: raise HTTPException(503, detail="Deezer unavailable") _cache_charts.set(cache_key, data) response.headers.update(_cc(36_000)) return data # ───────────────────────────────────────────────────────────────────────────── # Spotify direct endpoint # ───────────────────────────────────────────────────────────────────────────── @app.get("/spotify/trending", tags=["spotify"]) @limiter.limit("30/minute") async def spotify_trending( request: Request, response: Response, country: str = Query("US"), limit: int = Query(50, ge=1, le=100), ): """Spotify featured playlists trending tracks by country (free, no login needed).""" if not _SPOTIFY_CLIENT_ID: raise HTTPException(503, detail="Spotify not configured (set SPOTIFY_CLIENT_ID + SPOTIFY_CLIENT_SECRET)") cache_key = f"spotify_trending:{country}:{limit}" cached = _cache_charts.get(cache_key) if cached is not None: response.headers.update(_cc(36_000)) return cached tracks = await _fetch_spotify_trending(country, limit) # Never raise exception for empty tracks (Spotify is disabled) result = {"country": country, "tracks": tracks, "count": len(tracks)} _cache_charts.set(cache_key, result) response.headers.update(_cc(36_000)) return result # ───────────────────────────────────────────────────────────────────────────── # Song metadata # ───────────────────────────────────────────────────────────────────────────── @app.get("/song/{video_id}", tags=["song"]) @limiter.limit("60/minute") async def get_song( request: Request, response: Response, video_id: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"song:{video_id}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(3_600)) return cached try: data = await run(get_ytm(country, language).get_song, video_id) raw = ( data.get("thumbnail", {}).get("thumbnails") or data.get("thumbnails") or [] ) thumbs = best_thumbnails_list(raw) data["thumbnails"] = thumbs data["thumbnail"] = thumbs[0]["url"] if thumbs else "" _cache_long.set(cache_key, data) response.headers.update(_cc(3_600)) return data except Exception as e: log.error("get_song %s: %s", video_id, e) raise HTTPException(500, detail=str(e)) @app.get("/stream/{video_id}", tags=["song"]) @limiter.limit("60/minute") async def get_stream( request: Request, response: Response, video_id: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"stream:{video_id}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(3_600)) return cached try: song_data = await run(get_ytm(country, language).get_song, video_id) vd = song_data.get("videoDetails", {}) if not vd: raise HTTPException(404, "Video not found") raw = (vd.get("thumbnail", {}).get("thumbnails") or song_data.get("thumbnail", {}).get("thumbnails") or []) thumbs = best_thumbnails_list(raw) result = { "video_id": video_id, "videoId": video_id, "url": f"https://www.youtube.com/watch?v={video_id}", "audio_url": f"https://www.youtube.com/watch?v={video_id}", "stream_url": f"https://www.youtube.com/watch?v={video_id}", "title": vd.get("title", ""), "artist": vd.get("author", ""), "channel_id": vd.get("channelId", ""), "duration_seconds": int(vd.get("lengthSeconds") or 0), "views": int(vd.get("viewCount") or 0), "keywords": vd.get("keywords", [])[:10], "is_live": vd.get("isLiveContent", False), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", } _cache_long.set(cache_key, result) response.headers.update(_cc(3_600)) return result except HTTPException: raise except Exception as e: log.error("get_stream %s: %s", video_id, e) raise HTTPException(500, detail=str(e)) @app.get("/now_playing/{video_id}", tags=["song"]) @limiter.limit("60/minute") async def now_playing( request: Request, response: Response, video_id: str, related_limit: int = Query(10, ge=1, le=30), country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"now_playing:{video_id}:{related_limit}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(1_800)) return cached ytm = get_ytm(country, language) song_data, watch_data = await asyncio.gather( run(ytm.get_song, video_id), run(ytm.get_watch_playlist, video_id, None, related_limit + 1), return_exceptions=True, ) stream: dict = {} if isinstance(song_data, dict): vd = song_data.get("videoDetails", {}) raw = (vd.get("thumbnail", {}).get("thumbnails") or song_data.get("thumbnail", {}).get("thumbnails") or []) thumbs = best_thumbnails_list(raw) stream = { "videoId": video_id, "url": f"https://www.youtube.com/watch?v={video_id}", "audio_url": f"https://www.youtube.com/watch?v={video_id}", "title": vd.get("title", ""), "artist": vd.get("author", ""), "duration_seconds": int(vd.get("lengthSeconds") or 0), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", } related: list = [] if isinstance(watch_data, dict): tracks_raw = watch_data.get("tracks") or [] if tracks_raw and tracks_raw[0].get("videoId") == video_id: tracks_raw = tracks_raw[1:] related = [norm_track(t) for t in tracks_raw[:related_limit] if t.get("videoId")] result = {"videoId": video_id, "stream": stream, "related": related} _cache_long.set(cache_key, result, ttl=1_800) response.headers.update(_cc(1_800)) return result @app.get("/related_songs/{video_id}", tags=["song"]) @limiter.limit("30/minute") async def get_related_songs( request: Request, response: Response, video_id: str, limit: int = Query(15, ge=1, le=50), country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"related:{video_id}:{limit}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(1_800)) return cached try: raw = await run(get_ytm(country, language).get_watch_playlist, video_id, None, limit + 1) tracks_raw = raw.get("tracks") or [] if tracks_raw and tracks_raw[0].get("videoId") == video_id: tracks_raw = tracks_raw[1:] tracks = [norm_track(t) for t in tracks_raw[:limit] if t.get("videoId")] result = {"videoId": video_id, "tracks": tracks, "count": len(tracks)} _cache_long.set(cache_key, result, ttl=1_800) response.headers.update(_cc(1_800)) return result except Exception as e: log.error("related_songs %s: %s", video_id, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # Up-Next # ───────────────────────────────────────────────────────────────────────────── @app.get("/upnext/{video_id}", tags=["song"]) @limiter.limit("30/minute") async def get_upnext( request: Request, response: Response, video_id: str, limit: int = Query(20, ge=5, le=50), force_refresh: bool = Query(False), country: str = Query("ZZ"), language: str = Query("en"), ): store_key = f"{video_id}:{country}:{language}" now = time.time() with _upnext_lock: existing = _upnext_store.get(store_key) if existing and not force_refresh: if now - existing.get("created_at", 0) < _UPNEXT_TTL: response.headers.update(_cc(300)) return existing try: raw = await run(get_ytm(country, language).get_watch_playlist, video_id, None, limit) tracks_raw = raw.get("tracks") or [] if tracks_raw and tracks_raw[0].get("videoId") == video_id: tracks_raw = tracks_raw[1:] tracks = [norm_track(t) for t in tracks_raw if t.get("videoId")] queue = { "origin_video_id": video_id, "tracks": tracks, "count": len(tracks), "created_at": now, "country": country, } with _upnext_lock: _upnext_store[store_key] = queue _upnext_store.move_to_end(store_key) while len(_upnext_store) > _UPNEXT_MAX: _upnext_store.popitem(last=False) response.headers.update(_cc(300)) return queue except Exception as e: log.error("upnext %s: %s", video_id, e) raise HTTPException(500, detail=str(e)) @app.delete("/upnext/{video_id}", tags=["song"]) async def reset_upnext(video_id: str, country: str = Query("ZZ"), language: str = Query("en")): store_key = f"{video_id}:{country}:{language}" with _upnext_lock: _upnext_store.pop(store_key, None) return {"cleared": store_key} # ───────────────────────────────────────────────────────────────────────────── # Artist # ───────────────────────────────────────────────────────────────────────────── @app.get("/artist/{artist_id}", tags=["artist"]) @limiter.limit("30/minute") async def get_artist( request: Request, response: Response, artist_id: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"artist:{artist_id}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(3_600)) return cached try: data = await run(get_ytm(country, language).get_artist, artist_id) if not data: raise HTTPException(404, "Artist not found") data["thumbnails"] = best_thumbnails_list(data.get("thumbnails") or []) data["thumbnail"] = data["thumbnails"][0]["url"] if data["thumbnails"] else "" _cache_long.set(cache_key, data) response.headers.update(_cc(3_600)) return data except HTTPException: raise except Exception as e: log.error("get_artist %s: %s", artist_id, e) raise HTTPException(500, detail=str(e)) @app.get("/artist/{artist_id}/songs", tags=["artist"]) @limiter.limit("20/minute") async def get_artist_songs( request: Request, response: Response, artist_id: str, limit: int = Query(20, ge=1, le=100), country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"artist_songs:{artist_id}:{limit}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(600)) return cached try: ytm = get_ytm(country, language) artist_data = await run(ytm.get_artist, artist_id) if not artist_data: raise HTTPException(404, "Artist not found") artist_name = artist_data.get("name", "") songs_section = artist_data.get("songs", {}) all_tracks: list = [] albums_info: list = [] if isinstance(songs_section, dict) and songs_section.get("browseId"): try: songs_raw = await run(ytm.get_artist_songs, artist_id, songs_section.get("params")) all_tracks.extend(songs_raw or []) except Exception: all_tracks.extend(songs_section.get("results") or []) else: all_tracks.extend(songs_section.get("results") or []) for section_key in ("albums", "singles"): section = artist_data.get(section_key, {}) if not isinstance(section, dict): continue entries = section.get("results") or section.get("items") or [] for entry in entries[:3]: thumbs = best_thumbnails_list(entry.get("thumbnails") or []) albums_info.append({ "browseId": entry.get("browseId", ""), "title": entry.get("title", ""), "year": entry.get("year", ""), "type": entry.get("type", "Album"), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", "trackCount": len(entry.get("tracks") or []), }) for t in entry.get("tracks") or []: if isinstance(t, dict) and t.get("videoId"): if not t.get("album"): t = dict(t) t["album"] = {"name": entry.get("title", "")} all_tracks.append(t) seen, deduped = set(), [] for t in all_tracks: vid = t.get("videoId", "") if vid and vid not in seen: seen.add(vid) deduped.append(norm_track(t)) result = { "artistId": artist_id, "name": artist_name, "songs": deduped[:limit], "total": len(deduped), "albums": albums_info, } _cache_long.set(cache_key, result) response.headers.update(_cc(600)) return result except HTTPException: raise except Exception as e: log.error("artist_songs %s: %s", artist_id, e) raise HTTPException(500, detail=str(e)) @app.get("/artist/{artist_id}/albums", tags=["artist"]) @limiter.limit("20/minute") async def get_artist_albums( request: Request, response: Response, artist_id: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"artist_albums:{artist_id}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(600)) return cached try: ytm = get_ytm(country, language) artist_data = await run(ytm.get_artist, artist_id) if not artist_data: raise HTTPException(404, "Artist not found") artist_name = artist_data.get("name", "") channel_id = artist_data.get("channelId", artist_id) album_entries: list = [] for section_key in ("albums", "singles"): section = artist_data.get(section_key, {}) if not isinstance(section, dict): continue params = section.get("params") if params and channel_id: try: more = await run(ytm.get_artist_albums, channel_id, params) album_entries.extend( more if isinstance(more, list) else (more.get("results") or more.get("items") or []) ) except Exception: album_entries.extend(section.get("results") or section.get("items") or []) else: album_entries.extend(section.get("results") or section.get("items") or []) async def _album_detail(entry: dict) -> dict: bid = entry.get("browseId") thumbs = best_thumbnails_list(entry.get("thumbnails") or []) base = { "browseId": bid or "", "title": entry.get("title", ""), "year": entry.get("year", ""), "type": entry.get("type", "Album"), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", "tracks": [], } if not bid: return base try: d = await run(ytm.get_album, bid) base["tracks"] = [norm_track(t) for t in (d.get("tracks") or []) if t.get("videoId")] except Exception: pass return base albums_out: list = [] for i in range(0, len(album_entries), 5): batch = album_entries[i:i+5] results = await asyncio.gather(*[_album_detail(e) for e in batch]) albums_out.extend(results) result = { "artistId": artist_id, "name": artist_name, "albums": albums_out, "totalAlbums": len(albums_out), "totalTracks": sum(len(a["tracks"]) for a in albums_out), } _cache_long.set(cache_key, result) response.headers.update(_cc(600)) return result except HTTPException: raise except Exception as e: log.error("artist_albums %s: %s", artist_id, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # Album # ───────────────────────────────────────────────────────────────────────────── @app.get("/album/{album_id}", tags=["album"]) @limiter.limit("30/minute") async def get_album( request: Request, response: Response, album_id: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"album:{album_id}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(3_600)) return cached try: data = await run(get_ytm(country, language).get_album, album_id) if not data: raise HTTPException(404, "Album not found") data["thumbnails"] = best_thumbnails_list(data.get("thumbnails") or []) data["thumbnail"] = data["thumbnails"][0]["url"] if data["thumbnails"] else "" data["tracks"] = [norm_track(t) for t in (data.get("tracks") or [])] _cache_long.set(cache_key, data) response.headers.update(_cc(3_600)) return data except HTTPException: raise except Exception as e: log.error("get_album %s: %s", album_id, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # Playlist # ───────────────────────────────────────────────────────────────────────────── @app.get("/playlist/{playlist_id}", tags=["playlist"]) @limiter.limit("30/minute") async def get_playlist( request: Request, response: Response, playlist_id: str, limit: int = Query(100, ge=1, le=500), related: bool = False, suggestions_limit: int = 0, country: str = Query("ZZ"), language: str = Query("en"), ): clean_id = playlist_id[2:] if playlist_id.startswith("VL") else playlist_id cache_key = f"playlist:{clean_id}:{limit}:{country}:{language}" cached = _cache_medium.get(cache_key) if cached is not None: response.headers.update(_cc(300)) return cached try: data = await run( get_ytm(country, language).get_playlist, clean_id, limit, related, suggestions_limit, ) if not data: raise HTTPException(404, "Playlist not found") data["tracks"] = [norm_track(t) for t in (data.get("tracks") or []) if isinstance(t, dict)] data["thumbnails"] = best_thumbnails_list(data.get("thumbnails") or []) data["thumbnail"] = data["thumbnails"][0]["url"] if data["thumbnails"] else "" _cache_medium.set(cache_key, data) response.headers.update(_cc(300)) return data except HTTPException: raise except Exception as e: log.error("get_playlist %s: %s", playlist_id, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # Podcast # ───────────────────────────────────────────────────────────────────────────── @app.get("/podcast/{podcast_id}", tags=["podcast"]) @limiter.limit("20/minute") async def get_podcast( request: Request, response: Response, podcast_id: str, limit: int = Query(50, ge=1, le=200), country: str = Query("ZZ"), language: str = Query("en"), ): clean_id = (podcast_id[2:] if podcast_id.startswith("VL") and not podcast_id.startswith("VLM") else podcast_id) cache_key = f"podcast:{clean_id}:{limit}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(1_800)) return cached ytm = get_ytm(country, language) episodes_raw: list = [] meta: dict = {} for fn, *fn_args in [ (ytm.get_podcast, clean_id), (ytm.get_playlist, clean_id, limit), (ytm.get_podcast, podcast_id), (ytm.get_playlist, podcast_id, limit), ]: if episodes_raw: break try: data = await run(fn, *fn_args) if isinstance(data, dict) and data: if not meta: meta = data episodes_raw = data.get("episodes") or data.get("tracks") or [] except Exception: pass if not meta and not episodes_raw: raise HTTPException(404, "Podcast not found") def _norm_ep(ep: dict) -> dict: raw_t = ep.get("thumbnails") or ep.get("thumbnail") or [] if isinstance(raw_t, str): raw_t = [{"url": raw_t, "width": 0}] thumbs = best_thumbnails_list(raw_t) dur = ep.get("duration") or ep.get("durationSeconds") or "" if isinstance(dur, int) and dur > 0: m, s = divmod(dur, 60) h, m = divmod(m, 60) dur = f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}" artists = ep.get("artists") or ep.get("author") or "" if isinstance(artists, list): artists = ", ".join( a.get("name", "") if isinstance(a, dict) else str(a) for a in artists ) return { "videoId": ep.get("videoId", "") or ep.get("id", ""), "title": ep.get("title", ""), "author": artists, "duration": dur, "date": ep.get("date", "") or ep.get("publishedAt", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", "description":ep.get("description", ""), } thumbs_p = best_thumbnails_list(meta.get("thumbnails") or []) result = { "podcastId": clean_id, "title": meta.get("title", ""), "author": meta.get("author", ""), "description":meta.get("description", ""), "thumbnails": thumbs_p, "thumbnail": thumbs_p[0]["url"] if thumbs_p else "", "episodes": [_norm_ep(ep) for ep in episodes_raw[:limit] if isinstance(ep, dict)], "count": min(len(episodes_raw), limit), } _cache_long.set(cache_key, result, ttl=1_800) response.headers.update(_cc(1_800)) return result # ───────────────────────────────────────────────────────────────────────────── # Watch playlist / lyrics # ───────────────────────────────────────────────────────────────────────────── @app.get("/watch/{video_id}", tags=["song"]) @limiter.limit("30/minute") async def get_watch_playlist( request: Request, response: Response, video_id: str, limit: int = Query(25, ge=1, le=50), country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"watch:{video_id}:{limit}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(1_800)) return cached try: raw = await run(get_ytm(country, language).get_watch_playlist, video_id, None, limit) tracks_raw = raw.get("tracks") or [] tracks = [norm_track(t) for t in tracks_raw if isinstance(t, dict) and t.get("videoId")] result = { "videoId": video_id, "tracks": tracks, "count": len(tracks), "lyrics": raw.get("lyrics") or raw.get("lyricsId") or "", "related": raw.get("related") or "", } _cache_long.set(cache_key, result, ttl=1_800) response.headers.update(_cc(1_800)) return result except Exception as e: log.error("watch %s: %s", video_id, e) raise HTTPException(500, detail=str(e)) @app.get("/lyrics/{browse_id}", tags=["song"]) @limiter.limit("30/minute") async def get_lyrics( request: Request, response: Response, browse_id: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"lyrics:{browse_id}" cached = _cache_long.get(cache_key) if cached is not None: response.headers.update(_cc(86_400)) return cached try: data = await run(get_ytm(country, language).get_lyrics, browse_id) _cache_long.set(cache_key, data or {}) response.headers.update(_cc(86_400)) return data or {} except Exception as e: log.warning("get_lyrics %s: %s", browse_id, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # Genres / featured playlists # ───────────────────────────────────────────────────────────────────────────── @app.get("/genres", tags=["discovery"]) @limiter.limit("30/minute") async def get_genres( request: Request, response: Response, country: str = Query("ZZ"), language: str = Query("en"), limit: int = Query(50, ge=1, le=200), ): cache_key = f"genres:{country}:{language}:{limit}" cached = _cache_medium.get(cache_key) if cached is not None: response.headers.update(_cc(600)) return cached ytm_task = run(get_ytm(country, language).get_charts, country) deezer_task = _fetch_deezer_editorial_playlists(20) ytm_charts_raw, deezer_playlists = await asyncio.gather( ytm_task, deezer_task, return_exceptions=True ) playlists: list = [] if not isinstance(ytm_charts_raw, Exception) and isinstance(ytm_charts_raw, dict): for sk in ("playlists", "genres", "moods", "trending"): section = ytm_charts_raw.get(sk, {}) items = section if isinstance(section, list) else ( section.get("items") or section.get("results") or [] ) for item in items: if not isinstance(item, dict): continue bid = item.get("browseId") or item.get("playlistId") if bid and item.get("title"): thumbs = best_thumbnails_list(item.get("thumbnails") or []) playlists.append({ "browseId": bid, "title": item.get("title", ""), "subtitle": item.get("subtitle", "") or item.get("author", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", "source": "ytm", }) try: cats = await run(get_ytm(country, language).get_mood_categories) first_params = None if isinstance(cats, dict): for items in cats.values(): if isinstance(items, list) and items: p = items[0].get("params") if p: first_params = p break elif isinstance(cats, list) and cats: first_params = cats[0].get("params") if first_params: mood_raw = await run(get_ytm(country, language).get_mood_playlists, first_params) for section in (mood_raw or []): if not isinstance(section, dict): continue for item in (section.get("contents") or section.get("playlists") or []): if not isinstance(item, dict): continue bid = item.get("playlistId") or item.get("browseId") if bid: thumbs = best_thumbnails_list(item.get("thumbnails") or []) playlists.append({ "browseId": bid, "title": item.get("title", ""), "subtitle": item.get("subtitle", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", "source": "ytm", }) except Exception: pass if not isinstance(deezer_playlists, Exception): playlists.extend(deezer_playlists) seen, deduped = set(), [] for p in playlists: bid = p.get("browseId", "") if bid and bid not in seen: seen.add(bid) deduped.append(p) result = deduped[:limit] _cache_medium.set(cache_key, result) response.headers.update(_cc(600)) return result # ───────────────────────────────────────────────────────────────────────────── # Mood categories & playlists # ───────────────────────────────────────────────────────────────────────────── @app.get("/mood_categories", tags=["discovery"]) @limiter.limit("30/minute") async def get_mood_categories( request: Request, response: Response, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"mood_categories:{country}:{language}" cached = _cache_medium.get(cache_key) if cached is not None: response.headers.update(_cc(600)) return cached try: ytm_task = run(get_ytm(country, language).get_mood_categories) deezer_task = _fetch_deezer_genres() ytm_raw, deezer_genres = await asyncio.gather(ytm_task, deezer_task, return_exceptions=True) categories: list = [] if not isinstance(ytm_raw, Exception): categories = _flatten_mood_categories(ytm_raw) existing_titles = {c["title"].lower() for c in categories} if not isinstance(deezer_genres, Exception): for g in deezer_genres: if g.get("title", "").lower() not in existing_titles: g["section"] = "Genres" categories.append(g) existing_titles.add(g["title"].lower()) _cache_medium.set(cache_key, categories) response.headers.update(_cc(600)) return categories except Exception as e: log.error("mood_categories country=%s: %s", country, e) raise HTTPException(500, detail=str(e)) @app.get("/mood_playlists/{params}", tags=["discovery"]) @limiter.limit("30/minute") async def get_mood_playlists( request: Request, response: Response, params: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"mood_playlists:{params}:{country}:{language}" cached = _cache_medium.get(cache_key) if cached is not None: response.headers.update(_cc(600)) return cached try: raw = await run(get_ytm(country, language).get_mood_playlists, params) playlists: list = [] for section in (raw if isinstance(raw, list) else []): if not isinstance(section, dict): continue contents = section.get("contents") or section.get("playlists") or [] if isinstance(contents, list): for item in contents: if not isinstance(item, dict): continue thumbs = best_thumbnails_list(item.get("thumbnails") or []) playlists.append({ "browseId": item.get("playlistId") or item.get("browseId", ""), "title": item.get("title", ""), "subtitle": item.get("subtitle", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", }) else: thumbs = best_thumbnails_list(section.get("thumbnails") or []) playlists.append({ "browseId": section.get("playlistId") or section.get("browseId", ""), "title": section.get("title", ""), "subtitle": section.get("subtitle", ""), "thumbnails": thumbs, "thumbnail": thumbs[0]["url"] if thumbs else "", }) _cache_medium.set(cache_key, playlists) response.headers.update(_cc(600)) return playlists except Exception as e: log.error("mood_playlists params=%s: %s", params, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # Explore # ───────────────────────────────────────────────────────────────────────────── @app.get("/explore", tags=["discovery"]) @limiter.limit("30/minute") async def get_explore( request: Request, response: Response, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"explore:{country}:{language}" cached = _cache_medium.get(cache_key) if cached is not None: response.headers.update(_cc(600)) return cached try: data = await run(get_ytm(country, language).get_explore) _cache_medium.set(cache_key, data) response.headers.update(_cc(600)) return data except Exception as e: log.error("get_explore country=%s: %s", country, e) raise HTTPException(500, detail=str(e)) # ───────────────────────────────────────────────────────────────────────────── # User # ───────────────────────────────────────────────────────────────────────────── @app.get("/user/{channel_id}", tags=["user"]) async def get_user( response: Response, channel_id: str, country: str = Query("ZZ"), language: str = Query("en"), ): cache_key = f"user:{channel_id}:{country}:{language}" cached = _cache_long.get(cache_key) if cached is not None: return cached try: data = await run(get_ytm(country, language).get_user, channel_id) _cache_long.set(cache_key, data) return data except Exception as e: log.error("get_user %s: %s", channel_id, e) raise HTTPException(500, detail=str(e)) @app.get("/user_playlists/{channel_id}", tags=["user"]) async def get_user_playlists( response: Response, channel_id: str, params: Optional[str] = None, country: str = Query("ZZ"), language: str = Query("en"), ): try: data = await run(get_ytm(country, language).get_user_playlists, channel_id, params) return data except Exception as e: log.error("user_playlists %s: %s", channel_id, e) raise HTTPException(500, detail=str(e))