musivapi / app.py
wilooper's picture
Force YTMusic to use curl_cffi session instead of standard requests
6d4bc9f
Raw History Blame Contribute Delete
119 kB
"""
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))