diff --git a/app.py b/app.py new file mode 100644 index 0000000..dc57fae --- /dev/null +++ b/app.py @@ -0,0 +1,2369 @@ +import asyncio +import unicodedata +import base64 +import hashlib +import hmac +import ipaddress +import json +import logging +import os +import random +import re +import time +import uuid +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any,Optional +from urllib.parse import parse_qs,quote,urljoin,urlparse +import aiosqlite +import httpx +from fastapi import FastAPI,HTTPException,Query,Request +from fastapi.responses import FileResponse,JSONResponse,StreamingResponse +from fastapi.staticfiles import StaticFiles +from pydantic import BaseModel,ConfigDict,Field,field_validator,model_validator +from yt_dlp import YoutubeDL +from network import get_client_ip + +logging.basicConfig(level=os.getenv("LOG_LEVEL","INFO")) +logger=logging.getLogger("minimal-yt") +APP_ENV=os.getenv("APP_ENV","development").strip().lower() +APP_VERSION=os.getenv("APP_VERSION","dev") +SQLITE_CACHE_PATH=os.getenv("SQLITE_CACHE_PATH","/app/data/cache.sqlite3") +AUDIO_CACHE_TTL=int(os.getenv("AUDIO_CACHE_TTL",str(60*4))) +AUDIO_INFO_CACHE_TTL=AUDIO_CACHE_TTL +AUDIO_NEGATIVE_CACHE_TTL=int(os.getenv("AUDIO_NEGATIVE_CACHE_TTL","90")) +SEARCH_CACHE_TTL=int(os.getenv("SEARCH_CACHE_TTL",str(60*5))) +RECOMMENDATION_CACHE_TTL=int(os.getenv("RECOMMENDATION_CACHE_TTL",str(60*10))) +SOURCE_METADATA_CACHE_TTL=int(os.getenv("SOURCE_METADATA_CACHE_TTL",str(60*20))) +NEGATIVE_CACHE_TTL=int(os.getenv("NEGATIVE_CACHE_TTL","90")) +CACHE_MAX_VALUE_BYTES=int(os.getenv("CACHE_MAX_VALUE_BYTES",str(2*1024*1024))) +ANONYMOUS_COOKIE_MAX_AGE=60*60*24*365*2 +CACHE_CLEANUP_INTERVAL_SECONDS=int(os.getenv("CACHE_CLEANUP_INTERVAL_SECONDS","300")) +CACHE_CLEANUP_BATCH_SIZE=int(os.getenv("CACHE_CLEANUP_BATCH_SIZE","1000")) +YTDLP_CONCURRENCY=int(os.getenv("YTDLP_CONCURRENCY","3")) +SEARCH_YTDLP_CONCURRENCY=int(os.getenv("SEARCH_YTDLP_CONCURRENCY","2")) +AUDIO_YTDLP_CONCURRENCY=int(os.getenv("AUDIO_YTDLP_CONCURRENCY","2")) +RECOMMENDATION_YTDLP_CONCURRENCY=int(os.getenv("RECOMMENDATION_YTDLP_CONCURRENCY","2")) +MAX_SEARCH_FETCH=int(os.getenv("MAX_SEARCH_FETCH","100")) +RATE_LIMIT_WINDOW_SECONDS=int(os.getenv("RATE_LIMIT_WINDOW_SECONDS","60")) +RATE_LIMIT_SEARCH_MAX=int(os.getenv("RATE_LIMIT_SEARCH_MAX","30")) +RATE_LIMIT_AUDIO_MAX=int(os.getenv("RATE_LIMIT_AUDIO_MAX","20")) +USER_AGENTS = [ + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", +] +DEFAULT_USER_AGENT = os.getenv("DEFAULT_USER_AGENT",random.choice(USER_AGENTS)) +MEDIA_TOKEN_TTL_SECONDS=int(os.getenv("MEDIA_TOKEN_TTL_SECONDS","120")) +MEDIA_TOKEN_BIND_ANONYMOUS_ID=os.getenv("MEDIA_TOKEN_BIND_ANONYMOUS_ID","false").strip().lower()=="true" +MEDIA_TOKEN_SECRET=os.getenv("MEDIA_TOKEN_SECRET","") +AUDIO_UPSTREAM_ORIGIN=os.getenv("AUDIO_UPSTREAM_ORIGIN","https://www.youtube.com") +AUDIO_UPSTREAM_REFERER=os.getenv("AUDIO_UPSTREAM_REFERER","https://www.youtube.com/") +ALLOWED_MEDIA_HOST_SUFFIXES=tuple(value.strip().lower() for value in os.getenv("ALLOWED_MEDIA_HOST_SUFFIXES",".googlevideo.com,googlevideo.com").split(",") if value.strip()) +HTTP_CONNECT_TIMEOUT_SECONDS=float(os.getenv("HTTP_CONNECT_TIMEOUT_SECONDS","10")) +HTTP_RESPONSE_HEADER_TIMEOUT_SECONDS=float(os.getenv("HTTP_RESPONSE_HEADER_TIMEOUT_SECONDS","60")) +HTTP_IDLE_READ_TIMEOUT_SECONDS=float(os.getenv("HTTP_IDLE_READ_TIMEOUT_SECONDS","0")) +HTTP_MAX_CONNECTIONS=int(os.getenv("HTTP_MAX_CONNECTIONS","100")) +HTTP_MAX_KEEPALIVE_CONNECTIONS=int(os.getenv("HTTP_MAX_KEEPALIVE_CONNECTIONS","20")) +MAX_ACTIVE_AUDIO_STREAMS=int(os.getenv("MAX_ACTIVE_AUDIO_STREAMS","16")) +MAX_ACTIVE_AUDIO_STREAMS_PER_IP=int(os.getenv("MAX_ACTIVE_AUDIO_STREAMS_PER_IP","4")) +MAX_ACTIVE_AUDIO_STREAMS_PER_ANONYMOUS_ID=int(os.getenv("MAX_ACTIVE_AUDIO_STREAMS_PER_ANONYMOUS_ID","4")) +MAX_REQUEST_BODY_BYTES=int(os.getenv("MAX_REQUEST_BODY_BYTES",str(256*1024))) +CHUNK_SIZE=1024*256 +MAX_MEDIA_REDIRECTS=int(os.getenv("MAX_MEDIA_REDIRECTS","3")) +KEY_LOCKS_MAX=int(os.getenv("KEY_LOCKS_MAX","4096")) +RATE_LIMIT_BUCKETS_MAX=int(os.getenv("RATE_LIMIT_BUCKETS_MAX","20000")) +PROXY_FILE=os.getenv("PROXY_FILE","proxies.txt").strip() +PROXY_USER=os.getenv("PROXY_USER","").strip() +PROXY_PASS=os.getenv("PROXY_PASS","").strip() +PROXY_SCHEMES=("http",) +PROXY_RETRY_ATTEMPTS=max(1,int(os.getenv("PROXY_RETRY_ATTEMPTS","5"))) +PROXY_RELOAD_SECONDS=max(5,int(os.getenv("PROXY_RELOAD_SECONDS","30"))) +PROXY_FAILURE_COOLDOWN_SECONDS=max(10,int(os.getenv("PROXY_FAILURE_COOLDOWN_SECONDS","120"))) +HTTP_TIMEOUT=httpx.Timeout(connect=HTTP_CONNECT_TIMEOUT_SECONDS,read=None if HTTP_IDLE_READ_TIMEOUT_SECONDS==0 else HTTP_IDLE_READ_TIMEOUT_SECONDS,write=10.0,pool=10.0) +ANONYMOUS_COOKIE_NAME="minimal_yt_anonymous_id" +ANONYMOUS_ID_RE=re.compile(r"^[A-Za-z0-9._:-]{8,128}$") +SHARE_TOKEN_RE=re.compile(r"^[a-f0-9]{32}$") +VIDEO_ID_RE=re.compile(r"^[A-Za-z0-9_-]{6,32}$") +RANGE_RE=re.compile(r"^bytes=(\d*)-(\d*)$") +REQUEST_ID_RE=re.compile(r"^[A-Za-z0-9._-]{8,128}$") + +MAX_TRACK_TITLE_LENGTH=500 +MAX_TRACK_CHANNEL_LENGTH=300 +MAX_TRACK_URL_LENGTH=2048 +MAX_TRACK_DURATION_STRING_LENGTH=32 +MAX_TRACK_THUMBNAIL_LENGTH=2048 +sqlite_db:Optional[aiosqlite.Connection]=None +shared_http_client:Optional[httpx.AsyncClient]=None +KEY_LOCKS:dict[str,asyncio.Lock]={} +KEY_LOCKS_GUARD=asyncio.Lock() +CACHE_GUARD=asyncio.Lock() +CACHE_LAST_CLEANUP_AT=0.0 +SEARCH_SEMAPHORE=asyncio.Semaphore(SEARCH_YTDLP_CONCURRENCY) +RECOMMENDATION_SEMAPHORE=asyncio.Semaphore(RECOMMENDATION_YTDLP_CONCURRENCY) +AUDIO_SEMAPHORE=asyncio.Semaphore(AUDIO_YTDLP_CONCURRENCY) +YTDLP_SEMAPHORE=asyncio.Semaphore(YTDLP_CONCURRENCY) +RATE_LIMIT_BUCKETS:dict[str,list[tuple[float,int]]]={} +RATE_LIMIT_GUARD=asyncio.Lock() +ACTIVE_STREAMS_SEMAPHORE=asyncio.Semaphore(MAX_ACTIVE_AUDIO_STREAMS) +ACTIVE_STREAMS_GUARD=asyncio.Lock() +ACTIVE_STREAMS_BY_IP:dict[str,int]={} +ACTIVE_STREAMS_BY_ANON:dict[str,int]={} +PROXY_ENDPOINTS:list[str]=[] +PROXY_INDEX=0 +PROXY_GUARD=asyncio.Lock() +PROXY_FAILURES:dict[str,float]={} +PROXY_LAST_LOAD=0.0 + +def require_media_secret()->str: + secret=MEDIA_TOKEN_SECRET.strip() + if len(secret)>=32:return secret + if APP_ENV=="production":raise RuntimeError("MEDIA_TOKEN_SECRET is missing or too short; required in production") + fallback="dev-insecure-media-token-secret-change-me-please-1234567890" + logger.warning("Using development MEDIA_TOKEN_SECRET fallback") + return fallback + +MEDIA_TOKEN_SECRET_EFFECTIVE=require_media_secret() + +@asynccontextmanager +async def lifespan(_:FastAPI): + global sqlite_db,shared_http_client,PROXY_ENDPOINTS,PROXY_LAST_LOAD + database_directory=os.path.dirname(SQLITE_CACHE_PATH) + if database_directory:os.makedirs(database_directory,exist_ok=True) + sqlite_db=await aiosqlite.connect(SQLITE_CACHE_PATH) + await sqlite_db.execute("PRAGMA foreign_keys=ON") + await sqlite_db.execute("PRAGMA journal_mode=WAL") + await sqlite_db.execute("PRAGMA synchronous=NORMAL") + await sqlite_db.execute("PRAGMA busy_timeout=5000") + await sqlite_db.execute("CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY,value TEXT NOT NULL,expires_at REAL NOT NULL,created_at REAL NOT NULL)") + await sqlite_db.execute("CREATE INDEX IF NOT EXISTS idx_cache_expires_at ON cache (expires_at)") + await sqlite_db.execute("CREATE TABLE IF NOT EXISTS anonymous_users (id TEXT PRIMARY KEY,created_at REAL NOT NULL,last_seen_at REAL NOT NULL)") + await ensure_favorites_schema() + await ensure_playlists_schema() + await ensure_player_state_schema() + await sqlite_db.commit() + await cache_cleanup() + PROXY_ENDPOINTS=load_proxy_endpoints() + PROXY_LAST_LOAD=time.time() + shared_http_client=httpx.AsyncClient(timeout=HTTP_TIMEOUT,follow_redirects=False,limits=httpx.Limits(max_connections=HTTP_MAX_CONNECTIONS,max_keepalive_connections=HTTP_MAX_KEEPALIVE_CONNECTIONS)) + yield + if shared_http_client: + await shared_http_client.aclose() + shared_http_client=None + if sqlite_db: + await sqlite_db.close() + sqlite_db=None + +app=FastAPI(title="Minimal yt",version=APP_VERSION,lifespan=lifespan) +app.mount("/static",StaticFiles(directory="static"),name="static") + +@app.middleware("http") +async def security_headers_middleware(request:Request,call_next): + response=await call_next(request) + response.headers.setdefault("X-Content-Type-Options","nosniff") + response.headers.setdefault("X-Frame-Options","DENY") + response.headers.setdefault("Referrer-Policy","strict-origin-when-cross-origin") + response.headers.setdefault("Permissions-Policy","camera=(), microphone=(), geolocation=()") + response.headers.setdefault("Cross-Origin-Opener-Policy","same-origin") + response.headers.setdefault("Cross-Origin-Resource-Policy","same-origin") + response.headers.setdefault("Content-Security-Policy","default-src 'self'; base-uri 'self'; frame-ancestors 'none'; form-action 'self'; object-src 'none'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' https://i.ytimg.com data:; media-src 'self'; connect-src 'self'") + return response + +@app.middleware("http") +async def request_id_middleware(request:Request,call_next): + candidate=request.headers.get("x-request-id","").strip() + request_id=candidate if REQUEST_ID_RE.fullmatch(candidate) else str(uuid.uuid4()) + request.state.request_id=request_id + try: + response=await call_next(request) + except HTTPException: + raise + except Exception: + logger.exception("Unhandled error",extra={"request_id":request_id,"path":request.url.path,"method":request.method}) + return JSONResponse(status_code=500,content={"detail":"Internal server error"},headers={"X-Request-Id":request_id}) + response.headers["X-Request-Id"]=request_id + return response + +@app.middleware("http") +async def request_body_limit_middleware(request:Request,call_next): + if request.method in {"POST","PUT","PATCH","DELETE"}: + content_length=request.headers.get("content-length","") + if content_length: + try: + if int(content_length)>MAX_REQUEST_BODY_BYTES:return JSONResponse(status_code=413,content={"detail":"Request body too large"}) + except ValueError:return JSONResponse(status_code=400,content={"detail":"Invalid Content-Length"}) + return await call_next(request) + +@app.middleware("http") +async def anonymous_id_cookie_middleware(request:Request,call_next): + anonymous_id,should_set_cookie=get_or_create_anonymous_id(request) + request.state.anonymous_id=anonymous_id + request.state.should_set_anonymous_cookie=should_set_cookie + response=await call_next(request) + if should_set_cookie:response.set_cookie(key=ANONYMOUS_COOKIE_NAME,value=anonymous_id,max_age=ANONYMOUS_COOKIE_MAX_AGE,httponly=True,secure=APP_ENV=="production",samesite="lax",path="/") + return response + +@app.get("/") +def index():return FileResponse("static/index.html") + +@app.get("/share/{share_token}") +def shared_playlist_page(share_token:str): + if not SHARE_TOKEN_RE.fullmatch(share_token):raise HTTPException(status_code=404,detail="Not found") + return FileResponse("static/index.html") + +@app.get("/healthz") +async def healthz(): + if not sqlite_db:raise HTTPException(status_code=503,detail="cache not initialized") + async with sqlite_db.execute("SELECT 1") as cursor:await cursor.fetchone() + return {"ok":True,"cache":"sqlite","version":APP_VERSION} + +@app.get("/readyz") +async def readyz(): + if not sqlite_db or not shared_http_client:raise HTTPException(status_code=503,detail="not ready") + return {"ok":True,"app_env":APP_ENV,"media_token_ttl":MEDIA_TOKEN_TTL_SECONDS,"allowed_media_host_suffixes":list(ALLOWED_MEDIA_HOST_SUFFIXES),"media_token_bind_anonymous_id":MEDIA_TOKEN_BIND_ANONYMOUS_ID,"proxy_count":len(PROXY_ENDPOINTS)} + +def get_anonymous_id(request:Request,required:bool=True)->Optional[str]: + anonymous_id=getattr(request.state,"anonymous_id",None) or request.cookies.get(ANONYMOUS_COOKIE_NAME,"").strip() or request.headers.get("x-anonymous-id","").strip() + if not anonymous_id: + if required:raise HTTPException(status_code=400,detail="Missing anonymous id") + return None + if not ANONYMOUS_ID_RE.fullmatch(anonymous_id):raise HTTPException(status_code=400,detail="Invalid anonymous id") + return anonymous_id + +def hash_anonymous_id(anonymous_id:Optional[str])->str:return hashlib.sha256((anonymous_id or "").encode()).hexdigest() + +def get_or_create_anonymous_id(request: Request) -> tuple[str, bool]: + header_id = request.headers.get("x-anonymous-id", "").strip() + cookie_id = request.cookies.get(ANONYMOUS_COOKIE_NAME, "").strip() + + # Prefer the cookie because normal media element requests cannot include + # X-Anonymous-Id. When only the header exists, persist it as the cookie. + anonymous_id = cookie_id or header_id or f"anon_{uuid.uuid4().hex}" + + if not ANONYMOUS_ID_RE.fullmatch(anonymous_id): + raise HTTPException(status_code=400, detail="Invalid anonymous id") + + # Set/update cookie whenever it differs from the selected identity. + return anonymous_id, cookie_id != anonymous_id +async def touch_anonymous_user(user_id:str): + if not sqlite_db:return + now=time.time() + await sqlite_db.execute("INSERT INTO anonymous_users (id,created_at,last_seen_at) VALUES (?,?,?) ON CONFLICT(id) DO UPDATE SET last_seen_at=excluded.last_seen_at",(user_id,now,now)) + await sqlite_db.commit() + +async def ensure_favorites_schema(): + if not sqlite_db:return + await sqlite_db.execute("CREATE TABLE IF NOT EXISTS favorites (user_id TEXT NOT NULL,id TEXT NOT NULL,data TEXT NOT NULL,created_at REAL NOT NULL,updated_at REAL NOT NULL,PRIMARY KEY (user_id,id))") + await sqlite_db.execute("CREATE INDEX IF NOT EXISTS idx_favorites_user_updated_at ON favorites (user_id,updated_at)") + +async def ensure_playlists_schema(): + if not sqlite_db:return + await sqlite_db.execute("CREATE TABLE IF NOT EXISTS playlists (id TEXT PRIMARY KEY,user_id TEXT NOT NULL,name TEXT NOT NULL,share_token TEXT UNIQUE,shared_at REAL,created_at REAL NOT NULL,updated_at REAL NOT NULL)") + async with sqlite_db.execute("PRAGMA table_info(playlists)") as cursor:columns=await cursor.fetchall() + column_names=[column[1] for column in columns] + if "share_token" not in column_names:await sqlite_db.execute("ALTER TABLE playlists ADD COLUMN share_token TEXT") + if "shared_at" not in column_names:await sqlite_db.execute("ALTER TABLE playlists ADD COLUMN shared_at REAL") + await sqlite_db.execute("CREATE INDEX IF NOT EXISTS idx_playlists_user_updated_at ON playlists (user_id,updated_at)") + await sqlite_db.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_playlists_share_token ON playlists (share_token) WHERE share_token IS NOT NULL") + await sqlite_db.execute("CREATE TABLE IF NOT EXISTS playlist_items (id TEXT PRIMARY KEY,user_id TEXT NOT NULL,playlist_id TEXT NOT NULL,item_id TEXT NOT NULL,data TEXT NOT NULL,position INTEGER NOT NULL,created_at REAL NOT NULL,FOREIGN KEY (playlist_id) REFERENCES playlists(id) ON DELETE CASCADE)") + await sqlite_db.execute("CREATE INDEX IF NOT EXISTS idx_playlist_items_playlist_position ON playlist_items (user_id,playlist_id,position)") + await sqlite_db.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_playlist_items_unique_item ON playlist_items (user_id,playlist_id,item_id)") + +async def ensure_player_state_schema(): + if sqlite_db:await sqlite_db.execute("CREATE TABLE IF NOT EXISTS player_state (user_id TEXT PRIMARY KEY,data TEXT NOT NULL,updated_at REAL NOT NULL)") + +async def rate_limit_request(request:Request,route:str,limit:int,window_seconds:int): + client_ip=get_client_ip(request) + anonymous_id=getattr(request.state,"anonymous_id",None) or request.cookies.get(ANONYMOUS_COOKIE_NAME,"").strip() or request.headers.get("x-anonymous-id","").strip() + if anonymous_id and not ANONYMOUS_ID_RE.fullmatch(str(anonymous_id)):anonymous_id=None + subjects=[f"{route}:ip:{client_ip}"]+([f"{route}:anon:{hash_anonymous_id(anonymous_id)}"] if anonymous_id else []) + now=time.time() + stale_before=now-window_seconds + async with RATE_LIMIT_GUARD: + if len(RATE_LIMIT_BUCKETS)>RATE_LIMIT_BUCKETS_MAX: + for key in [k for k,v in RATE_LIMIT_BUCKETS.items() if not v or v[-1][0]<=stale_before]:RATE_LIMIT_BUCKETS.pop(key,None) + checked=[] + for bucket_key in subjects: + bucket=[(timestamp,count) for timestamp,count in RATE_LIMIT_BUCKETS.get(bucket_key,[]) if timestamp>stale_before] + if sum(count for _,count in bucket)>=limit:raise HTTPException(status_code=429,detail="Too many requests",headers={"Retry-After":str(window_seconds)}) + checked.append((bucket_key,bucket)) + for bucket_key,bucket in checked: + bucket.append((now,1)) + RATE_LIMIT_BUCKETS[bucket_key]=bucket + +def make_cache_key(prefix:str,*parts:Any)->str:return f"ytmusic:{prefix}:{hashlib.sha256(json.dumps(parts,sort_keys=True,ensure_ascii=False).encode()).hexdigest()}" +def make_negative_cache_key(prefix:str,*parts:Any)->str:return make_cache_key(f"negative:{prefix}",*parts) + +async def maybe_cache_cleanup(): + global CACHE_LAST_CLEANUP_AT + if not sqlite_db:return + now=time.time() + if now-CACHE_LAST_CLEANUP_ATOptional[Any]: + if not sqlite_db or not key:return None + await maybe_cache_cleanup() + try: + async with sqlite_db.execute("SELECT value,expires_at FROM cache WHERE key=?",(key,)) as cursor:row=await cursor.fetchone() + except Exception: + logger.exception("Cache read failed for key=%s",key[:96]);return None + if not row:return None + raw_value,expires_at=row + try: + if float(expires_at)<=time.time(): + await cache_delete(key);return None + except (TypeError,ValueError): + await cache_delete(key);return None + if not isinstance(raw_value,str) or len(raw_value.encode("utf-8",errors="ignore"))>CACHE_MAX_VALUE_BYTES: + await cache_delete(key);return None + try:return json.loads(raw_value) + except (TypeError,ValueError,json.JSONDecodeError): + await cache_delete(key);return None + +async def cache_set(key:str,value:Any,ttl:int): + if not sqlite_db or not key:return + try:ttl_seconds=int(ttl) + except (TypeError,ValueError):return + if ttl_seconds<=0: + await cache_delete(key);return + try:raw_value=json.dumps(value,ensure_ascii=False,separators=(",",":"),allow_nan=False) + except (TypeError,ValueError):return + if len(raw_value.encode())>CACHE_MAX_VALUE_BYTES:return + now=time.time() + try: + await sqlite_db.execute("INSERT INTO cache (key,value,expires_at,created_at) VALUES (?,?,?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value,expires_at=excluded.expires_at,created_at=excluded.created_at",(key,raw_value,now+ttl_seconds,now)) + await sqlite_db.commit() + except Exception:logger.exception("Cache write failed for key=%s",key[:96]) + +async def cache_delete(key:str): + if not sqlite_db or not key:return + try: + await sqlite_db.execute("DELETE FROM cache WHERE key=?",(key,)) + await sqlite_db.commit() + except Exception:logger.exception("Cache delete failed for key=%s",key[:96]) + +async def cache_cleanup(): + global CACHE_LAST_CLEANUP_AT + if not sqlite_db:return + now=time.time() + try: + while True: + cursor=await sqlite_db.execute("DELETE FROM cache WHERE key IN (SELECT key FROM cache WHERE expires_at<=? LIMIT ?)",(now,CACHE_CLEANUP_BATCH_SIZE)) + deleted=max(cursor.rowcount,0) + await cursor.close() + await sqlite_db.commit() + if deletedasyncio.Lock: + async with KEY_LOCKS_GUARD: + existing=KEY_LOCKS.get(key) + if existing:return existing + if len(KEY_LOCKS)>=KEY_LOCKS_MAX: + for candidate in [k for k,v in KEY_LOCKS.items() if not v.locked()][:max(1,len(KEY_LOCKS)-max(0,KEY_LOCKS_MAX-1))]:KEY_LOCKS.pop(candidate,None) + lock=asyncio.Lock() + KEY_LOCKS[key]=lock + return lock + +def is_allowed_url(url:str)->bool: + try:parsed=urlparse(url) + except ValueError:return False + if parsed.scheme not in {"http","https"} or parsed.username or parsed.password or parsed.port not in {None,80,443}:return False + host=(parsed.hostname or "").lower() + return host in {"youtube.com","www.youtube.com","music.youtube.com","m.youtube.com","youtu.be"} or host.endswith(".youtube.com") + +def is_valid_video_id(video_id:str)->bool:return bool(VIDEO_ID_RE.fullmatch(video_id or "")) +def youtube_watch_url(video_id:str)->Optional[str]:return f"https://www.youtube.com/watch?v={video_id}" if video_id and is_valid_video_id(video_id) else None + +def youtube_video_id_from_url(url:str)->Optional[str]: + if not url or not isinstance(url,str):return None + try:parsed=urlparse(url) + except ValueError:return None + host=(parsed.hostname or "").lower() + path=parsed.path or "" + if host=="youtu.be":return path.strip("/").split("/")[0] or None + if host in {"youtube.com","www.youtube.com","music.youtube.com","m.youtube.com"} or host.endswith(".youtube.com"): + if path=="/watch":return parse_qs(parsed.query).get("v",[None])[0] + if path.startswith("/shorts/") or path.startswith("/embed/"): + parts=path.strip("/").split("/") + return parts[1] if len(parts)>1 else None + return None + +def canonical_video_id_from_url(url:str)->Optional[str]: + video_id=youtube_video_id_from_url(url) + return video_id if video_id and is_valid_video_id(video_id) else None + +def canonical_youtube_url_from_url(url:str)->Optional[str]: + video_id=canonical_video_id_from_url(url) + return youtube_watch_url(video_id) if video_id else None + +def make_audio_cache_key(url:str)->str:return make_cache_key("audio",canonical_video_id_from_url(url) or url) +def make_source_metadata_cache_key(url:str)->str:return make_cache_key("source_meta",canonical_video_id_from_url(url) or url) +def recommendation_text(value:Any)->str:return " ".join(str(value or "").casefold().split()) + +def normalize_youtube_url(entry:dict)->Optional[str]: + video_id=entry.get("id") + webpage_url=entry.get("webpage_url") + raw_url=entry.get("url") + if webpage_url and isinstance(webpage_url,str) and webpage_url.startswith("http"):return webpage_url + if raw_url and isinstance(raw_url,str) and raw_url.startswith("http"):return raw_url + if video_id:return youtube_watch_url(video_id) + if raw_url and isinstance(raw_url,str) and "/" not in raw_url and len(raw_url)>=8:return youtube_watch_url(raw_url) + return None + +def best_thumbnail(entry:dict)->Optional[str]: + thumbnail=entry.get("thumbnail") + if thumbnail:return "https:"+thumbnail if isinstance(thumbnail,str) and thumbnail.startswith("//") else thumbnail + thumbnails=entry.get("thumbnails") or [] + if thumbnails: + sorted_thumbnails=sorted(thumbnails,key=lambda item:(item.get("width") or 0)*(item.get("height") or 0)) + url=sorted_thumbnails[-1].get("url") + if url:return "https:"+url if url.startswith("//") else url + video_id=entry.get("id") + if video_id:return f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg" + raw_url=entry.get("url") + if raw_url and isinstance(raw_url,str) and "/" not in raw_url and len(raw_url)>=8:return f"https://i.ytimg.com/vi/{raw_url}/hqdefault.jpg" + return None + +def is_video_like_result(entry:dict)->bool: + if not entry or not entry.get("title"):return False + entry_type=str(entry.get("_type") or "").lower() + ie_key=str(entry.get("ie_key") or "").lower() + combined_url=f"{entry.get('url') or ''} {entry.get('webpage_url') or ''}".lower() + if entry_type in {"playlist","channel","multi_video","url_transparent"} or ie_key in {"youtubetab","youtubeplaylist","youtubechannel","youtubeuser","youtubequery"}:return False + if any(path in combined_url for path in ("/playlist?","/channel/","/c/","/user/","/feed/","/shorts/","/results?")):return False + video_id=entry.get("id") or youtube_video_id_from_url(str(entry.get("webpage_url") or "")) + return bool(video_id and is_valid_video_id(str(video_id))) + +def normalize_view_count(entry:dict)->Optional[int]: + for value in (entry.get("view_count"),entry.get("views"),entry.get("n_view_count")): + if value is not None: + try:return int(value) + except (TypeError,ValueError):pass + return None + +def music_relevance_score(entry:dict)->int: + score=0 + title=str(entry.get("title") or "").lower() + channel=str(entry.get("channel") or entry.get("uploader") or entry.get("creator") or entry.get("channel_name") or "").lower() + try:duration_seconds=int(entry.get("duration")) if entry.get("duration") is not None else None + except (TypeError,ValueError):duration_seconds=None + view_count=normalize_view_count(entry) + if duration_seconds is not None: + if 120<=duration_seconds<=600:score+=5 + elif 45<=duration_seconds<120:score+=3 + elif 25<=duration_seconds<45:score+=1 + elif duration_seconds>20*60:score-=4 + for term,points in {"topic":6,"vevo":4,"official":3,"records":2,"music":2}.items(): + if term in channel:score+=points + for term,points in {"official audio":8,"official music video":6,"official video":4,"audio":3,"visualizer":3,"lyrics":2,"lyric video":1}.items(): + if term in title:score+=points + for term,points in {"live":-2,"free for profit":-4,"free for":-3,"type beat":-2,"concert":-4,"cover":-2,"remix":-1,"slowed":-1,"sped up":-1,"nightcore":-2,"1 hour":-6,"loop":-3}.items(): + if term in title:score+=points + if view_count is not None: + if view_count>=10_000_000:score+=4 + elif view_count>=1_000_000:score+=3 + elif view_count>=100_000:score+=2 + elif view_count>=10_000:score+=2 + elif view_count<=1_000:score+=1 + return score + +logging.getLogger("httpx").setLevel(logging.WARNING) +logging.getLogger("httpcore").setLevel(logging.WARNING) + +def is_music_result(entry:dict)->bool: + title=" ".join(str(entry.get("title") or "").lower().split()) + channel=" ".join(str(entry.get("channel") or entry.get("uploader") or entry.get("creator") or entry.get("channel_name") or "").lower().split()) + combined_url=f"{entry.get('url') or ''} {entry.get('webpage_url') or ''}".lower() + if any(term in title for term in ("full album","full ep","podcast","interview","reaction","tutorial","reacting","trailer","documentary","behind the scenes","shorts","review","explained","how to ")):return False + if any(term in channel for term in ("podcast","news","clips","tutorial","reaction","review")):return False + if any(part in combined_url for part in ("/playlist?","/shorts/","/channel/","/feed/")):return False + try:duration_seconds=int(entry.get("duration")) if entry.get("duration") is not None else None + except (TypeError,ValueError):duration_seconds=None + if duration_seconds is not None and duration_seconds<25 and not any(term in title for term in ("official audio","music video","visualizer","song","audio")):return False + return True + +def format_duration(seconds:Any)->str: + try:seconds=int(seconds) + except (TypeError,ValueError):return "-" + if seconds<0:return "-" + hours,remainder=divmod(seconds,3600) + minutes,seconds=divmod(remainder,60) + return f"{hours}:{minutes:02d}:{seconds:02d}" if hours else f"{minutes}:{seconds:02d}" + +def normalize_search_entry(entry:dict,index:int=0)->Optional[dict]: + raw_url=entry.get("url") + webpage_url=entry.get("webpage_url") + video_id=entry.get("id") + if not video_id: + for candidate_url in (webpage_url,raw_url): + if isinstance(candidate_url,str): + video_id=youtube_video_id_from_url(candidate_url) + if video_id:break + if not video_id and isinstance(raw_url,str) and "/" not in raw_url and len(raw_url)>=8:video_id=raw_url + if not video_id or not is_valid_video_id(video_id):return None + url=normalize_youtube_url(entry) or youtube_watch_url(video_id) + if not url:return None + duration=entry.get("duration") + duration_string=entry.get("duration_string") or format_duration(duration) + channel=entry.get("channel") or entry.get("uploader") or entry.get("creator") or entry.get("channel_name") or "-" + title=entry.get("title") or "untitled" + is_live=bool(entry.get("is_live") or entry.get("live_status") in {"is_live","is_upcoming","was_live"}) + return {"index":index,"id":video_id,"title":title,"url":url,"channel":channel,"duration":duration,"duration_string":"live" if is_live else duration_string,"thumbnail":best_thumbnail(entry),"is_live":is_live,"view_count":normalize_view_count(entry),"source":"youtube"} + +RECOMMENDATION_LONG_TITLE_CHARS=int(os.getenv("RECOMMENDATION_LONG_TITLE_CHARS","80")) +RECOMMENDATION_LONG_TITLE_WORDS=int(os.getenv("RECOMMENDATION_LONG_TITLE_WORDS","12")) +RECOMMENDATION_TITLE_CLEAN_RE=re.compile(r"(?i)(?Optional[int]: + try: + seconds=int(float(value)) + return seconds if seconds>=0 else None + except (TypeError,ValueError):return None + +def clean_recommendation_title(value:Any)->str: + title=RECOMMENDATION_TITLE_CLEAN_RE.sub(" ",str(value or "")) + title=re.sub(r"\s+"," ",title).strip(" -_|/\\:;,.") + return title or "untitled" + +def recommendation_title_identifier(value:Any)->dict: + raw=str(value or "") + variant=None + for name,pattern,_penalty in RECOMMENDATION_VARIANT_PATTERNS: + if pattern.search(raw): + variant=name + break + return {"variant":variant,"is_free":bool(RECOMMENDATION_FREE_RE.search(raw)),"is_cover":bool(RECOMMENDATION_COVER_RE.search(raw)),"is_concert":bool(RECOMMENDATION_CONCERT_RE.search(raw))} + +def clean_recommendation_text(value:Any)->str: + text=re.sub(r"[\s\-_–—|/\\:;,.()[\]{}]+"," ",str(value or "")) + text=re.sub(r"\s+"," ",text).strip() + for phrase in RECOMMENDATION_NOISE_PHRASES: + pattern=r"\s+".join(re.escape(part) for part in phrase.split()) + text=re.sub(rf"(?i)(?str:return clean_recommendation_text(value).casefold() +def normalized_contains_meaningful_track(track:str,title:str)->bool: + normalized_track=recommendation_identity(track) + normalized_title=recommendation_identity(title) + return len(normalized_track)>=4 and normalized_track in normalized_title + +def clean_recommendation_list(values:Any)->list[str]: + if not values:return [] + if isinstance(values,str):values=[values] + results=[];seen=set() + for raw_value in values: + cleaned=clean_recommendation_text(raw_value);key=cleaned.casefold() + if not cleaned or key in GENERIC_RECOMMENDATION_TAGS or key in seen:continue + seen.add(key);results.append(cleaned) + return results + +def infer_common_genre(*values:Any)->Optional[str]: + texts=[] + for value in values: + if not value:continue + if isinstance(value,str):texts.append(value) + else: + try:texts.extend(str(item) for item in value if item) + except TypeError:texts.append(str(value)) + for text in texts: + normalized=re.sub(r"\s+"," ",re.sub(r"[^a-z0-9&]+"," ",clean_recommendation_text(text).casefold())).strip() + for genre_variant,canonical in COMMON_GENRE_PATTERNS: + genre_key=re.sub(r"\s+"," ",re.sub(r"[^a-z0-9&]+"," ",genre_variant.casefold())).strip() + if re.search(rf"(?dict: + if not is_allowed_url(url):raise ValueError("Unsupported URL") + options=base_ytdlp_options(proxy_url=proxy_url);options.update({"extract_flat":False}) + with YoutubeDL(options) as ydl:info=ydl.extract_info(url,download=False) + if not info:raise ValueError("Could not extract source metadata") + return info + +async def extract_music_source_metadata(url:str)->dict: + canonical_url=canonical_youtube_url_from_url(url) or url + cache_key=make_cache_key("source_meta",canonical_url) + negative_key=make_negative_cache_key("source_meta",canonical_url) + cached=await cache_get(cache_key) + if cached:return cached + if await cache_get(negative_key):raise ValueError("Recommendations temporarily unavailable") + lock=await get_key_lock(cache_key) + async with lock: + cached=await cache_get(cache_key) + if cached:return cached + if await cache_get(negative_key):raise ValueError("Recommendations temporarily unavailable") + last_error=None + try: + async with RECOMMENDATION_SEMAPHORE: + for attempt in range(PROXY_RETRY_ATTEMPTS+1): + proxy_url=await get_next_proxy_url() if attemptdict: + video_id=raw.get("id");channel=raw.get("channel") or raw.get("uploader");artist=raw.get("artist") or channel or raw.get("uploader") + track=raw.get("track") or clean_recommendation_text(raw.get("title"));album=clean_recommendation_text(raw.get("album")) + tags=clean_recommendation_list(raw.get("tags"));categories=clean_recommendation_list(raw.get("categories")) + genre=clean_recommendation_text(raw.get("genre")) or infer_common_genre(tags,categories) + webpage_url=raw.get("webpage_url") or youtube_watch_url(video_id) or fallback_url + return {"id":video_id,"title":raw.get("title") or track or "untitled","track":track,"artist":clean_recommendation_text(artist),"album":album,"genre":genre,"channel":clean_recommendation_text(channel),"uploader":clean_recommendation_text(raw.get("uploader")),"tags":tags,"categories":categories,"duration":raw.get("duration"),"webpage_url":webpage_url,"url":webpage_url,"thumbnail":best_thumbnail(raw)} + +def generate_music_recommendation_queries(source:dict)->list[dict]: + artist=source.get("artist");track=source.get("track");genre=source.get("genre");album=source.get("album");tags=source.get("tags") or [];categories=source.get("categories") or [];candidates=[] + def add(kind:str,query:str,weight:float): + cleaned=clean_recommendation_text(query) + if cleaned:candidates.append((kind,cleaned,weight)) + if artist and track:add("artist_title",f"{artist} {track}",1.00) + if artist and genre:add("artist_genre",f"{artist} {genre}",.90) + if artist and tags:add("artist_tags"," ".join([artist,*tags[:2]]),.85) + if genre and tags:add("genre_tags"," ".join([genre,*tags[:3]]),.80) + if tags:add("tags_only"," ".join(tags[:4]),.75) + if artist and album:add("artist_album",f"{artist} {album}",.70) + if artist and track: + add("track_remix",f"{artist} {track} remix",.65);add("track_live",f"{artist} {track} live",.60);add("track_acoustic",f"{artist} {track} acoustic",.55) + if track and genre:add("title_genre",f"{track} {genre}",.50) + for category in categories: + if category.casefold()!="music" and artist:add("artist_category",f"{artist} {category}",.45) + if artist:add("artist_music_fallback",f"{artist} music",.40) + if not candidates and track:add("title_fallback",track,.35) + deduplicated={} + for index,(kind,query,weight) in enumerate(candidates): + key=query.casefold();existing=deduplicated.get(key) + if existing is None or weight>existing["weight"]:deduplicated[key]={"kind":kind,"query":query,"weight":weight,"_index":index} + results=sorted(deduplicated.values(),key=lambda item:(-item["weight"],item["_index"],item["query"].casefold())) + for item in results:item.pop("_index",None) + return results + +def ytdlp_recommendation_search_sync(query:str,per_query:int,proxy_url:Optional[str])->dict: + options=base_ytdlp_options(proxy_url=proxy_url);options.update({"extract_flat":True,"default_search":"ytsearch","noplaylist":True,"skip_download":True,"quiet":True}) + with YoutubeDL(options) as ydl:return ydl.extract_info(f"ytsearch{per_query}:{query}",download=False) or {} + +async def ytdlp_recommendation_search(query:str,per_query:int)->dict: + last_error=None + async with RECOMMENDATION_SEMAPHORE: + for attempt in range(PROXY_RETRY_ATTEMPTS+1): + proxy_url=await get_next_proxy_url() if attemptbool: + duration=recommendation_duration_seconds(candidate.get("duration")) + return (candidate.get("title_flags") or {}).get("variant")=="playlist" or duration is not None and duration>2*60*60 + +def normalize_recommendation_candidate(entry:dict)->Optional[dict]: + video_id=entry.get("id") + if not video_id or not is_valid_video_id(video_id):return None + raw_title=str(entry.get("title") or "untitled");title=clean_recommendation_title(raw_title) + return {"id":video_id,"title":title,"raw_title":raw_title,"title_flags":recommendation_title_identifier(raw_title),"channel":entry.get("channel") or entry.get("uploader") or entry.get("creator") or entry.get("channel_name") or "-","url":entry.get("webpage_url") or youtube_watch_url(video_id),"duration":entry.get("duration"),"duration_string":entry.get("duration_string") or format_duration(entry.get("duration")),"thumbnail":best_thumbnail(entry)} + +def recommendation_candidate_score(*,source:dict,candidate:dict,query_weight:float,rank:int)->float: + raw_title=str(candidate.get("raw_title") or candidate.get("title") or "");title=str(candidate.get("title") or "");flags=candidate.get("title_flags") or recommendation_title_identifier(raw_title) + source_artist=recommendation_identity(source.get("artist"));source_channel=recommendation_identity(source.get("channel"));candidate_channel=recommendation_identity(candidate.get("channel")) + score=float(query_weight)+max(0.0,.25-rank*.03) + if candidate_channel and (candidate_channel==source_artist or candidate_channel==source_channel):score+=.20 + elif candidate_channel and ((source_artist and source_artist in candidate_channel) or (source_channel and source_channel in candidate_channel)):score+=.10 + if normalized_contains_meaningful_track(source.get("track") or "",title):score-=.10 + if flags.get("is_free"):score-=.35 + if flags.get("is_cover"):score-=.20 + if flags.get("is_concert"):score-=.20 + for name,_pattern,penalty in RECOMMENDATION_VARIANT_PATTERNS: + if flags.get("variant")==name: + score+=penalty;break + if len(raw_title)>RECOMMENDATION_LONG_TITLE_CHARS and len(raw_title.split())>RECOMMENDATION_LONG_TITLE_WORDS:score-=.20 + duration=recommendation_duration_seconds(candidate.get("duration")) + if duration is not None and duration>30*60:score-=.25 + if any(term in recommendation_text(candidate.get("channel")) for term in ("official","vevo","topic")):score+=.06 + return round(score,6) + +def sort_recommendation_results(results:list[dict])->list[dict]:return sorted(results,key=lambda item:(-item["_score"],item["_best_rank"],recommendation_identity(item.get("title")),item.get("id") or "")) + +async def get_music_recommendations(url:str,total:int=20,per_query:int=8)->dict: + total=max(1,min(int(total or 20),100));per_query=max(1,min(int(per_query or 8),25));canonical_url=canonical_youtube_url_from_url(url) or url + cache_key=make_cache_key("recommendations",canonical_url,total,per_query);negative_key=make_negative_cache_key("recommendations",canonical_url,total,per_query) + cached=await cache_get(cache_key) + if cached:return cached + lock=await get_key_lock(cache_key) + async with lock: + cached=await cache_get(cache_key) + if cached:return cached + try: + source=await extract_music_source_metadata(canonical_url);queries=generate_music_recommendation_queries(source);source_id=source.get("id") + search_results=await asyncio.gather(*(ytdlp_recommendation_search(q["query"],per_query) for q in queries),return_exceptions=True) + candidates_by_id={} + for query_info,search_info in zip(queries,search_results): + if isinstance(search_info,Exception):continue + for rank,entry in enumerate(search_info.get("entries") or []): + candidate=normalize_recommendation_candidate(entry) + if not candidate or is_hard_blocked_recommendation(candidate) or source_id and candidate["id"]==source_id:continue + score=recommendation_candidate_score(source=source,candidate=candidate,query_weight=float(query_info["weight"]),rank=rank) + matched_query={"query":query_info["query"],"kind":query_info["kind"],"weight":float(query_info["weight"]),"rank":rank,"score":round(score,6)} + existing=candidates_by_id.get(candidate["id"]) + if existing: + existing["_score"]+=score*.5;existing["_best_rank"]=min(existing["_best_rank"],rank);existing["matched_queries"].append(matched_query) + else: + candidate["_score"]=score;candidate["_best_rank"]=rank;candidate["matched_queries"]=[matched_query];candidates_by_id[candidate["id"]]=candidate + results=[{"id":item["id"],"title":item["title"],"raw_title":item.get("raw_title"),"title_flags":item.get("title_flags") or {},"channel":item["channel"],"url":item["url"],"duration":item["duration"],"duration_string":item["duration_string"],"thumbnail":item["thumbnail"],"score":round(item["_score"],6),"matched_queries":item["matched_queries"]} for item in sort_recommendation_results(list(candidates_by_id.values()))[:total]] + data={"source":source,"queries":queries,"results":results,"count":len(results)} + await cache_set(cache_key,data,RECOMMENDATION_CACHE_TTL) + return data + except Exception as exc: + logger.warning("Recommendations failed url=%s error=%s",canonical_url,exc) + await cache_set(negative_key,{"failed":True},min(NEGATIVE_CACHE_TTL,30)) + raise ValueError("Recommendations temporarily unavailable") from exc + +class StoredTrack(BaseModel): + model_config=ConfigDict(extra="ignore",str_strip_whitespace=True) + id:str=Field(...,min_length=6,max_length=32) + title:str=Field(default="untitled",min_length=1,max_length=MAX_TRACK_TITLE_LENGTH) + channel:Optional[str]=Field(default=None,max_length=MAX_TRACK_CHANNEL_LENGTH) + duration:Optional[int]=Field(default=None,ge=0,le=86400) + duration_string:Optional[str]=Field(default=None,max_length=MAX_TRACK_DURATION_STRING_LENGTH) + thumbnail:Optional[str]=Field(default=None,max_length=MAX_TRACK_THUMBNAIL_LENGTH) + view_count:Optional[int]=Field(default=None,ge=0) + is_live:bool=False + source:str=Field(default="youtube",max_length=32) + @field_validator("id") + @classmethod + def validate_id(cls,value:str)->str: + value=str(value or "").strip() + if not is_valid_video_id(value):raise ValueError("Invalid video id") + return value + @field_validator("title","channel","duration_string","source",mode="before") + @classmethod + def normalize_text(cls,value:Any)->Optional[str]: + if value is None:return None + normalized=" ".join(str(value).split()) + return normalized or None + @field_validator("thumbnail",mode="before") + @classmethod + def normalize_thumbnail(cls,value:Any)->Optional[str]: + return str(value).strip() if value is not None and str(value).strip() else None + @field_validator("thumbnail") + @classmethod + def validate_thumbnail_url(cls,value:Optional[str])->Optional[str]: + if value is None:return None + try:parsed=urlparse(value) + except ValueError as exc:raise ValueError("Invalid thumbnail URL") from exc + if parsed.scheme!="https" or (parsed.hostname or "").lower() not in {"i.ytimg.com","yt3.ggpht.com"}:raise ValueError("Invalid thumbnail URL") + return value + +class FavoriteToggleBody(BaseModel):item:StoredTrack + +class PlaylistItemReorderBody(BaseModel): + item_ids: list[str] = Field(..., min_length=1, max_length=1000) + + @field_validator("item_ids") + @classmethod + def validate_item_ids(cls, values: list[str]) -> list[str]: + for value in values: + try: + uuid.UUID(value) + except (TypeError, ValueError) as exc: + raise ValueError("Invalid playlist item id") from exc + return values + +class PlaylistCreateBody(BaseModel):name:str=Field(...,min_length=1,max_length=200) +class PlaylistRenameBody(BaseModel):name:str=Field(...,min_length=1,max_length=200) +class PlaylistItemAddBody(BaseModel):item:StoredTrack +class PlaylistReorderBody(BaseModel): + item_ids:list[str]=Field(...,min_length=1,max_length=1000) + @field_validator("item_ids") + @classmethod + def validate_item_ids(cls,values:list[str])->list[str]: + for value in values: + try:uuid.UUID(value) + except (TypeError,ValueError) as exc:raise ValueError("Invalid playlist item id") from exc + return values +class PlaylistItemsDeleteBody(BaseModel): + item_ids:list[str]=Field(...,min_length=1,max_length=1000) + @field_validator("item_ids") + @classmethod + def validate_item_ids(cls,values:list[str])->list[str]: + for value in values: + try:uuid.UUID(value) + except (TypeError,ValueError) as exc:raise ValueError("Invalid playlist item id") from exc + return values +class AudioSessionCreateBody(BaseModel): + video_id:str=Field(...,min_length=6,max_length=32) + @field_validator("video_id") + @classmethod + def validate_video_id(cls,value:str)->str: + if not is_valid_video_id(value):raise ValueError("Invalid video id") + return value +class PlayerStateBody(BaseModel): + current_item:Optional[StoredTrack]=None + current_time:float=Field(default=0,ge=0,le=86400) + queue:list[StoredTrack]=Field(default_factory=list,max_length=500) + queue_index:int=Field(default=-1,ge=-1,le=499) + autoplay:bool=True + shuffle:bool=False + loop_mode:str=Field(default="off",max_length=8) + volume:float=Field(default=1.0,ge=0,le=1) + queue_mode:str=Field(default="queue",max_length=32) + source_url:Optional[str]=Field(default=None,max_length=2048) + @field_validator("loop_mode") + @classmethod + def validate_loop_mode(cls,value:str)->str: + normalized=str(value or "off").strip().lower() + if normalized not in {"off","all","one"}:raise ValueError("Invalid loop mode") + return normalized + @field_validator("queue_mode") + @classmethod + def validate_queue_mode(cls,value:str)->str: + normalized=str(value or "queue").strip().lower() + if normalized not in {"queue","recommendations"}:raise ValueError("Invalid queue mode") + return normalized + @model_validator(mode="after") + def validate_queue_index_bounds(self): + if not self.queue or self.queue_index>=len(self.queue):self.queue_index=-1 + return self + +def load_stored_track_json(raw:str)->Optional[dict]: + try:return StoredTrack.model_validate(json.loads(raw)).model_dump(exclude_none=True) + except Exception: + logger.warning("Skipping invalid persisted track record");return None + +def clean_playlist_name(name:str)->str:return " ".join(str(name or "").strip().split())[:80] + +async def get_favorite_by_id(user_id:str,favorite_id:str): + if not sqlite_db:return None + async with sqlite_db.execute("SELECT id,data,created_at,updated_at FROM favorites WHERE user_id=? AND id=?",(user_id,favorite_id)) as cursor:row=await cursor.fetchone() + if not row:return None + favorite_id,data,created_at,updated_at=row;item=load_stored_track_json(data) + return {"id":favorite_id,"item":item,"created_at":created_at,"updated_at":updated_at} if item is not None else None + +async def list_favorites(user_id:str)->list[dict]: + if not sqlite_db:return [] + async with sqlite_db.execute("SELECT id,data,created_at,updated_at FROM favorites WHERE user_id=? ORDER BY updated_at DESC",(user_id,)) as cursor:rows=await cursor.fetchall() + results=[] + for favorite_id,data,created_at,updated_at in rows: + item=load_stored_track_json(data) + if item is not None:results.append({"id":favorite_id,"item":item,"created_at":created_at,"updated_at":updated_at}) + return results + +async def get_playlist_for_user(user_id:str,playlist_id:str): + if not sqlite_db:return None + async with sqlite_db.execute("SELECT id,user_id,name,share_token,shared_at,created_at,updated_at FROM playlists WHERE user_id=? AND id=?",(user_id,playlist_id)) as cursor:row=await cursor.fetchone() + if not row:return None + playlist_id,playlist_user_id,name,share_token,shared_at,created_at,updated_at=row + return {"id":playlist_id,"user_id":playlist_user_id,"name":name,"share_token":share_token,"shared_at":shared_at,"created_at":created_at,"updated_at":updated_at} + +async def get_playlist_items(user_id:str,playlist_id:str)->list[dict]: + if not sqlite_db:return [] + async with sqlite_db.execute("SELECT id,item_id,data,position,created_at FROM playlist_items WHERE user_id=? AND playlist_id=? ORDER BY position ASC,created_at ASC",(user_id,playlist_id)) as cursor:rows=await cursor.fetchall() + items=[] + for row_id,item_id,data,position,created_at in rows: + item=load_stored_track_json(data) + if item is not None: + item.update({"playlist_item_id":row_id,"playlist_id":playlist_id,"item_id":item_id,"position":position,"playlist_created_at":created_at}) + items.append(item) + return items + +def playlist_total_duration(items:list[dict])->int: + total=0 + for item in items: + try:duration=int(item.get("duration") or 0) + except (TypeError,ValueError):duration=0 + if duration>0:total+=duration + return total + +async def get_player_state(user_id:str)->Optional[dict]: + if not sqlite_db:return None + async with sqlite_db.execute("SELECT data,updated_at FROM player_state WHERE user_id=?",(user_id,)) as cursor:row=await cursor.fetchone() + if not row:return None + try:validated=PlayerStateBody.model_validate(json.loads(row[0])).model_dump(exclude_none=True) + except Exception: + logger.warning("Discarding invalid player state for user=%s",user_id[:16]);return None + validated["updated_at"]=row[1] + return validated + +async def save_player_state(user_id:str,state:dict): + if not sqlite_db:return + try:validated=PlayerStateBody.model_validate(state).model_dump(exclude_none=True) + except Exception as exc:raise HTTPException(status_code=422,detail="Invalid player state") from exc + raw_state=json.dumps(validated,ensure_ascii=False,separators=(",",":"),allow_nan=False) + if len(raw_state.encode())>CACHE_MAX_VALUE_BYTES:raise HTTPException(status_code=413,detail="Player state too large") + now=time.time() + await sqlite_db.execute("INSERT INTO player_state (user_id,data,updated_at) VALUES (?,?,?) ON CONFLICT(user_id) DO UPDATE SET data=excluded.data,updated_at=excluded.updated_at",(user_id,raw_state,now)) + await sqlite_db.commit() + +def _proxy_auth(user:str,password:str)->str:return f"{quote(user,safe='')}:{quote(password,safe='')}@" if user or password else "" +def encode_proxy_auth(user:str,password:str)->str:return _proxy_auth(user,password) +def normalize_proxy_scheme(scheme:Optional[str])->str: + scheme=str(scheme or "http").lower().replace("://","").strip() + return scheme if scheme in PROXY_SCHEMES else "http" + +def _parse_proxy_line(line:str)->list[str]: + line=line.split("#",1)[0].strip() + if not line or line.startswith("//"):return [] + try: + if "://" in line: + p=urlparse(line);scheme=p.scheme.lower();host=p.hostname;port=p.port + if scheme not in PROXY_SCHEMES or not host or not port:return [] + user=p.username if p.username is not None else PROXY_USER;password=p.password if p.password is not None else PROXY_PASS + return [f"{scheme}://{_proxy_auth(user,password)}{host}:{port}"] + if "@" in line: + auth,hp=line.rsplit("@",1);user,password=auth.split(":",1) if ":" in auth else (auth,"");host,port=hp.rsplit(":",1) + if not host or not port.isdigit():return [] + return [f"{scheme}://{_proxy_auth(user,password)}{host}:{int(port)}" for scheme in PROXY_SCHEMES] + parts=line.split(":") + if len(parts)==2 and parts[1].isdigit(): + host,port=parts;return [f"{scheme}://{_proxy_auth(PROXY_USER,PROXY_PASS)}{host}:{int(port)}" for scheme in PROXY_SCHEMES] + if len(parts)==4 and parts[1].isdigit(): + host,port,user,password=parts;return [f"{scheme}://{_proxy_auth(user,password)}{host}:{int(port)}" for scheme in PROXY_SCHEMES] + if len(parts)==4 and parts[3].isdigit(): + user,password,host,port=parts;return [f"{scheme}://{_proxy_auth(user,password)}{host}:{int(port)}" for scheme in PROXY_SCHEMES] + except Exception:return [] + return [] + +def load_proxy_endpoints()->list[str]: + if not PROXY_FILE or not os.path.exists(PROXY_FILE):return [] + out=[];seen=set() + try: + with open(PROXY_FILE,encoding="utf-8",errors="ignore") as f: + for line in f: + for proxy in _parse_proxy_line(line): + if proxy not in seen:seen.add(proxy);out.append(proxy) + except Exception:logger.exception("Proxy load failed") + return out + +async def get_next_proxy_url()->Optional[str]: + global PROXY_INDEX,PROXY_ENDPOINTS,PROXY_LAST_LOAD + now=time.time() + async with PROXY_GUARD: + if not PROXY_ENDPOINTS or now-PROXY_LAST_LOAD>=PROXY_RELOAD_SECONDS: + PROXY_ENDPOINTS=load_proxy_endpoints();PROXY_LAST_LOAD=now + if not PROXY_ENDPOINTS:return None + available=[p for p in PROXY_ENDPOINTS if PROXY_FAILURES.get(p,0)<=now] + if not available: + PROXY_FAILURES.clear();available=PROXY_ENDPOINTS + proxy=available[PROXY_INDEX%len(available)] + PROXY_INDEX+=1 + return proxy + +async def mark_proxy_failure(proxy:Optional[str]): + if proxy: + async with PROXY_GUARD:PROXY_FAILURES[proxy]=time.time()+PROXY_FAILURE_COOLDOWN_SECONDS + +async def mark_proxy_success(proxy:Optional[str]): + if proxy: + async with PROXY_GUARD:PROXY_FAILURES.pop(proxy,None) +def base_ytdlp_options(proxy_url:Optional[str]=None,player_client:Optional[str]=None)->dict: + options={"quiet":True,"no_warnings":True,"skip_download":True,"noplaylist":True,"cachedir":False,"socket_timeout":20,"retries":4,"fragment_retries":4,"extractor_retries":4,"file_access_retries":3,"concurrent_fragment_downloads":1,"ignoreerrors":False,"sleep_interval_requests":1,"http_headers":{"User-Agent":DEFAULT_USER_AGENT,"Accept":"*/*","Accept-Language":"en-US,en;q=0.9","Accept-Encoding":"identity"}} + if player_client:options["extractor_args"]={"youtube":{"player_client":[player_client]}} + cookie_file=os.getenv("YTDLP_COOKIE_FILE","").strip() + if cookie_file: + path=Path(cookie_file).expanduser() + if path.is_file() and os.access(path,os.R_OK):options["cookiefile"]=str(path) + else:logger.warning("Configured YTDLP_COOKIE_FILE is not readable") + browser=os.getenv("YTDLP_BROWSER","").strip() + if browser and "cookiefile" not in options: + parts=browser.split(":",1) + options["cookiesfrombrowser"]=(parts[0],) if len(parts)==1 else (parts[0],parts[1]) + if proxy_url:options["proxy"]=proxy_url + return options +def ytdlp_search_sync(query: str, limit: int, proxy_url: Optional[str]): + options = base_ytdlp_options(proxy_url) + options.update({ + "extract_flat": True, + "default_search": "ytsearch", + "skip_download": True, + "noplaylist": True, + "quiet": True, + "retries": 2, + "fragment_retries": 1, + "extractor_retries": 1, + "socket_timeout": 10, + }) + + search_query = " ".join(str(query or "").split()) + with YoutubeDL(options) as ydl: + return ydl.extract_info(f"ytsearch{limit}:{search_query}", download=False) or {} + +async def ytdlp_search(query: str) -> dict: + normalized_query = " ".join(str(query or "").casefold().split()) + cache_key = make_cache_key("search", normalized_query, MAX_SEARCH_FETCH) + negative_key = make_negative_cache_key("search", normalized_query) + + cached = await cache_get(cache_key) + if cached: + return cached + + cached_negative = await cache_get(negative_key) + if cached_negative: + return cached_negative + + lock = await get_key_lock(cache_key) + async with lock: + cached = await cache_get(cache_key) + if cached: + return cached + + cached_negative = await cache_get(negative_key) + if cached_negative: + return cached_negative + + last_error = None + info = {} + + async with SEARCH_SEMAPHORE: + for attempt in range(PROXY_RETRY_ATTEMPTS + 1): + proxy = await get_next_proxy_url() if attempt < PROXY_RETRY_ATTEMPTS else None + try: + info = await run_ytdlp_in_executor( + ytdlp_search_sync, + query, + MAX_SEARCH_FETCH, + proxy, + ) + if proxy: + await mark_proxy_success(proxy) + break + except Exception as exc: + last_error = exc + if proxy: + await mark_proxy_failure(proxy) + if attempt >= PROXY_RETRY_ATTEMPTS: + raise + await asyncio.sleep(min(0.2 * (attempt + 1), 0.8)) + else: + raise last_error or ValueError("Search temporarily unavailable") + + ranked_by_video_id = {} + for entry in info.get("entries") or []: + if not is_video_like_result(entry) or not is_music_result(entry): + continue + + item = normalize_search_entry(entry, index=0) + if not item or not item.get("url") or not item.get("id"): + continue + + video_id = str(item["id"]) + score = music_relevance_score(entry) + existing = ranked_by_video_id.get(video_id) + + if existing is None or score > existing[0]: + ranked_by_video_id[video_id] = (score, item) + + ranked = sorted( + ranked_by_video_id.values(), + key=lambda pair: ( + -pair[0], + str(pair[1].get("title") or "").casefold(), + str(pair[1].get("id") or ""), + ), + ) + + results = [] + for index, (_score, item) in enumerate(ranked, start=1): + item["index"] = index + results.append(item) + + data = { + "results": results, + "count": len(results), + "query": query, + "limit": MAX_SEARCH_FETCH, + } + + if results: + await cache_set(cache_key, data, SEARCH_CACHE_TTL) + await cache_delete(negative_key) + return data + + negative_data = { + "results": [], + "count": 0, + "query": query, + "limit": MAX_SEARCH_FETCH, + } + await cache_set(negative_key, negative_data, min(NEGATIVE_CACHE_TTL, 30)) + return negative_data + + + +def b64url_encode(data:bytes)->str:return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") +def b64url_decode(data:str)->bytes:return base64.urlsafe_b64decode(data+"="*(-len(data)%4)) + +def sign_media_payload(payload:dict)->str: + payload_part=b64url_encode(json.dumps(payload,separators=(",",":"),sort_keys=True).encode()) + signature=hmac.new(MEDIA_TOKEN_SECRET_EFFECTIVE.encode(),payload_part.encode("ascii"),hashlib.sha256).digest() + return f"{payload_part}.{b64url_encode(signature)}" + +def verify_media_token(token:str,anonymous_id:Optional[str])->dict: + if not token or len(token)>1024:raise HTTPException(status_code=404,detail="Invalid media token") + try:payload_part,signature_part=token.split(".",1);provided_signature=b64url_decode(signature_part) + except Exception as exc:raise HTTPException(status_code=404,detail="Invalid media token") from exc + expected_signature=hmac.new(MEDIA_TOKEN_SECRET_EFFECTIVE.encode(),payload_part.encode("ascii"),hashlib.sha256).digest() + if not hmac.compare_digest(expected_signature,provided_signature):raise HTTPException(status_code=404,detail="Invalid media token") + try:payload=json.loads(b64url_decode(payload_part).decode());expires_at=int(payload["exp"]) + except Exception as exc:raise HTTPException(status_code=404,detail="Invalid media token") from exc + if payload.get("v")!=1 or expires_at<=int(time.time()):raise HTTPException(status_code=404,detail="Invalid media token") + video_id=payload.get("vid") + if not isinstance(video_id,str) or not is_valid_video_id(video_id):raise HTTPException(status_code=404,detail="Invalid media token") + if MEDIA_TOKEN_BIND_ANONYMOUS_ID: + expected_hash=payload.get("ah") + if not isinstance(expected_hash,str) or not hmac.compare_digest(expected_hash,hash_anonymous_id(anonymous_id)):raise HTTPException(status_code=404,detail="Invalid media token") + return payload + +def create_media_token(video_id:str,anonymous_id:Optional[str])->tuple[str,int]: + payload={"v":1,"vid":video_id,"exp":int(time.time())+MEDIA_TOKEN_TTL_SECONDS} + if MEDIA_TOKEN_BIND_ANONYMOUS_ID: + if not anonymous_id:raise HTTPException(status_code=400,detail="Missing anonymous id") + payload["ah"]=hash_anonymous_id(anonymous_id) + return sign_media_payload(payload),MEDIA_TOKEN_TTL_SECONDS + +def is_public_ip_literal(hostname:str)->bool: + try:ip=ipaddress.ip_address(hostname) + except ValueError:return False + return not (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified) + +def validate_media_url(url:str)->str: + try:parsed=urlparse(url) + except ValueError as exc:raise ValueError("Malformed media URL") from exc + if parsed.scheme!="https" or parsed.username or parsed.password or parsed.port not in {None,443}:raise ValueError("Invalid media URL") + hostname=(parsed.hostname or "").strip().lower() + if not hostname or hostname=="localhost" or is_public_ip_literal(hostname):raise ValueError("Invalid media URL host") + try:ip=ipaddress.ip_address(hostname) + except ValueError:ip=None + if ip is not None and (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_unspecified):raise ValueError("Invalid media URL host") + for suffix in ALLOWED_MEDIA_HOST_SUFFIXES: + clean_suffix=suffix.lower() + if clean_suffix.startswith("."): + if hostname.endswith(clean_suffix):return url + elif hostname==clean_suffix or hostname.endswith("."+clean_suffix):return url + raise ValueError("Invalid media URL host") +def filter_upstream_headers(headers: dict) -> dict[str, str]: + """ + Keep only safe request headers needed for Googlevideo media fetches. + + Do not forward Host, Cookie, X-Forwarded-*, Content-Length, or arbitrary + values from upstream metadata. + """ + allowed = { + "accept", + "accept-language", + "origin", + "referer", + "user-agent", + } + + cleaned: dict[str, str] = {} + + for key, value in (headers or {}).items(): + name = str(key).strip() + header_value = str(value).replace("\r", "").replace("\n", "").strip() + + if not name or not header_value: + continue + + if name.lower() in allowed: + cleaned[name] = header_value[:2048] + + return cleaned + +def pick_best_audio_format(info:dict)->dict: + formats = info.get("formats") or [] + candidates = [] + + for fmt in formats: + if not fmt.get("url"): + continue + if fmt.get("has_drm"): + continue + + protocol = str(fmt.get("protocol") or "").lower() + if "m3u8" in protocol or "dash" in protocol: + continue + + acodec = fmt.get("acodec") + vcodec = fmt.get("vcodec") + if acodec and acodec != "none" and (not vcodec or vcodec == "none"): + candidates.append(fmt) + + if not candidates: + for fmt in formats: + if not fmt.get("url"): + continue + if fmt.get("has_drm"): + continue + protocol = str(fmt.get("protocol") or "").lower() + if "m3u8" in protocol or "dash" in protocol: + continue + candidates.append(fmt) + + if not candidates and info.get("url"): + return info + + def score(fmt:dict)->tuple: + ext = str(fmt.get("ext") or "").lower() + abr = float(fmt.get("abr") or fmt.get("tbr") or 0) + filesize = int(fmt.get("filesize") or fmt.get("filesize_approx") or 0) + ext_score = { + "m4a": 70, + "mp4": 60, + "webm": 40, + "opus": 35, + "mp3": 30, + "aac": 25, + }.get(ext, 0) + return (ext_score, abr, filesize) + + candidates.sort(key=score, reverse=True) + return candidates[0] if candidates else info + + + +def is_ytdlp_bot_verification_error(exc:Exception)->bool: + text=str(exc).lower() + return "sign in to confirm" in text or "not a bot" in text or "confirm you’re not a bot" in text or "confirm you're not a bot" in text + +def resolve_audio_sync(url:str,proxy_url:Optional[str],player_client:Optional[str]=None)->dict: + if not is_allowed_url(url):raise ValueError("Unsupported URL") + options=base_ytdlp_options(proxy_url,player_client) + options.update({"format":"bestaudio/best","extract_flat":False,"prefer_free_formats":False,"retries":4,"fragment_retries":2,"extractor_retries":2,"file_access_retries":1,"socket_timeout":12}) + try: + with YoutubeDL(options) as ydl:info=ydl.extract_info(url,download=False) + if not info:raise ValueError("Could not extract media info") + if info.get("_type")=="playlist" and info.get("entries"):info=next((entry for entry in info["entries"] if entry),None) or info + audio_format=pick_best_audio_format(info) + audio_url=audio_format.get("url") or info.get("url") + if not audio_url:raise ValueError("Could not resolve audio URL") + validate_media_url(audio_url) + http_headers=filter_upstream_headers({**(info.get("http_headers") or {}),**(audio_format.get("http_headers") or {})}) + http_headers.setdefault("User-Agent",DEFAULT_USER_AGENT) + http_headers.setdefault("Accept","*/*") + http_headers["Accept-Encoding"]="identity" + return {"title":info.get("title") or "track","video_id":info.get("id") or youtube_video_id_from_url(info.get("webpage_url") or url),"webpage_url":info.get("webpage_url") or url,"audio_url":audio_url,"ext":audio_format.get("ext") or info.get("ext"),"duration":info.get("duration"),"thumbnail":info.get("thumbnail"),"http_headers":http_headers,"proxy_url":proxy_url,"player_client":player_client,"created_at":time.time()} + except Exception as exc: + if is_ytdlp_bot_verification_error(exc):raise ValueError(f"YouTube bot verification failed for player client {player_client or 'default'}") from exc + raise + +#async def resolve_audio(url:str)->dict: +# last_error=None +# async with AUDIO_SEMAPHORE: +# for attempt in range(PROXY_RETRY_ATTEMPTS+1): +# proxy=await get_next_proxy_url() if attempt dict: + """ + Compatibility wrapper. + + All resolution paths use get_cached_audio_info(), which handles cache, + player-client selection, proxy fallback, validation, and negative cache. + """ + return await get_cached_audio_info(url, force_refresh=True) + +async def get_cached_audio_info( + url: str, + force_refresh: bool = False, +) -> dict: + canonical_url = canonical_youtube_url_from_url(url) or url + cache_key = make_cache_key("audio", canonical_url) + negative_key = make_negative_cache_key("audio", canonical_url) + + if not force_refresh: + if await cache_get(negative_key): + raise ValueError("Audio temporarily unavailable") + + cached = await cache_get(cache_key) + if cached and cached.get("audio_url"): + cached.setdefault("proxy_url", None) + cached.setdefault("player_client", None) + return cached + + lock = await get_key_lock(cache_key) + + async with lock: + if not force_refresh: + if await cache_get(negative_key): + raise ValueError("Audio temporarily unavailable") + + cached = await cache_get(cache_key) + if cached and cached.get("audio_url"): + cached.setdefault("proxy_url", None) + cached.setdefault("player_client", None) + return cached + + player_clients = tuple( + client.strip() + for client in os.getenv( + "YTDLP_PLAYER_CLIENTS", + "android,mweb,ios", + ).split(",") + if client.strip() + ) or ("mweb",) + + last_error: Optional[Exception] = None + saw_drm = False + + # Always try direct first. Then try unique configured proxies. + proxy_attempts: list[Optional[str]] = [None] + + for _ in range(PROXY_RETRY_ATTEMPTS): + proxy_url = await get_next_proxy_url() + + if proxy_url and proxy_url not in proxy_attempts: + proxy_attempts.append(proxy_url) + + for proxy_url in proxy_attempts: + for player_client in player_clients: + try: + result = await run_ytdlp_in_executor( + resolve_audio_sync, + canonical_url, + proxy_url, + player_client, + ) + + if not result or not result.get("audio_url"): + raise ValueError("Could not resolve audio URL") + + validate_media_url(result["audio_url"]) + + result["proxy_url"] = proxy_url + result["player_client"] = player_client + result["canonical_url"] = canonical_url + result["created_at"] = time.time() + + await cache_set( + cache_key, + result, + AUDIO_INFO_CACHE_TTL, + ) + await cache_delete(negative_key) + + if proxy_url: + await mark_proxy_success(proxy_url) + + return result + + except Exception as exc: + last_error = exc + kind = classify_ytdlp_error(exc) + + if kind == "drm": + saw_drm = True + + logger.info( + "Audio resolution failed " + "video=%s client=%s proxy=%s kind=%s error=%r", + canonical_video_id_from_url(canonical_url) or "unknown", + player_client, + "configured" if proxy_url else "direct", + kind, + exc, + ) + + # The current proxy failed for every configured player client. + if proxy_url: + await mark_proxy_failure(proxy_url) + + await cache_set( + negative_key, + {"failed": True}, + AUDIO_NEGATIVE_CACHE_TTL, + ) + + if saw_drm: + raise ValueError( + "This video is DRM protected and unavailable for direct audio streaming" + ) + + raise ValueError(str(last_error or "Audio temporarily unavailable")) + +def classify_ytdlp_error(exc: Exception) -> str: + text = str(exc).lower() + + if "drm" in text: + return "drm" + + if ( + "sign in to confirm" in text + or "not a bot" in text + or "bot verification" in text + or "confirm you're not a bot" in text + or "confirm you’re not a bot" in text + ): + return "bot_verification" + + if "timed out" in text or "timeout" in text: + return "timeout" + + if "proxy" in text: + return "proxy" + + if "403" in text or "forbidden" in text: + return "forbidden" + + if "429" in text or "too many requests" in text: + return "rate_limited" + + return "unknown" + +async def clear_audio_cache(url:str): + canonical_url=canonical_youtube_url_from_url(url) or url + await cache_delete(make_cache_key("audio",canonical_url)) + await cache_delete(make_negative_cache_key("audio",canonical_url)) + +async def prewarm_audio_for_video(video_id:str): + youtube_url=youtube_watch_url(video_id) + if not youtube_url:return + try:await get_cached_audio_info(youtube_url) + except Exception:logger.debug("Audio prewarm failed for %s",video_id) + +def validate_range_header(range_header:Optional[str])->Optional[str]: + if not range_header:return None + value=range_header.strip() + if len(value)>128 or "," in value:raise HTTPException(status_code=416,detail="Invalid Range header") + match=RANGE_RE.fullmatch(value) + if not match:raise HTTPException(status_code=416,detail="Invalid Range header") + start,end=match.groups() + if not start and not end or start and not start.isdigit() or end and not end.isdigit() or start and end and int(end)=MAX_ACTIVE_AUDIO_STREAMS_PER_IP:raise HTTPException(503,"Audio capacity unavailable") + ACTIVE_STREAMS_BY_IP[client_ip]=ip_count+1 + if anonymous_id: + anon_count=ACTIVE_STREAMS_BY_ANON.get(anonymous_id,0) + if anon_count>=MAX_ACTIVE_AUDIO_STREAMS_PER_ANONYMOUS_ID: + ACTIVE_STREAMS_BY_IP[client_ip]-=1 + if ACTIVE_STREAMS_BY_IP[client_ip]<=0:ACTIVE_STREAMS_BY_IP.pop(client_ip,None) + raise HTTPException(503,"Audio capacity unavailable") + ACTIVE_STREAMS_BY_ANON[anonymous_id]=anon_count+1 + except Exception: + if acquired:ACTIVE_STREAMS_SEMAPHORE.release() + raise + +async def release_stream_slot(client_ip:str,anonymous_id:Optional[str]): + async with ACTIVE_STREAMS_GUARD: + if client_ip in ACTIVE_STREAMS_BY_IP: + ACTIVE_STREAMS_BY_IP[client_ip]-=1 + if ACTIVE_STREAMS_BY_IP[client_ip]<=0:ACTIVE_STREAMS_BY_IP.pop(client_ip,None) + if anonymous_id and anonymous_id in ACTIVE_STREAMS_BY_ANON: + ACTIVE_STREAMS_BY_ANON[anonymous_id]-=1 + if ACTIVE_STREAMS_BY_ANON[anonymous_id]<=0:ACTIVE_STREAMS_BY_ANON.pop(anonymous_id,None) + ACTIVE_STREAMS_SEMAPHORE.release() + +def sanitize_etag(value:Optional[str])->Optional[str]:return value if value and len(value)<=256 else None +def sanitize_content_type(value:Optional[str])->str: + value=str(value or "audio/mpeg").split("\r",1)[0].split("\n",1)[0].strip() + return value[:128] if value else "audio/mpeg" +def sanitize_last_modified(value:Optional[str])->Optional[str]:return value if value and len(value)<=128 else None + +def content_disposition_filename( + filename: str, + extension: str, + download: bool, +) -> str: + disposition = "attachment" if download else "inline" + + # HTTP header fallback: ASCII only, safe for Latin-1 header encoding. + ascii_name = ( + unicodedata.normalize("NFKD", filename) + .encode("ascii", "ignore") + .decode("ascii") + ) + ascii_name = re.sub(r'[\x00-\x1f\x7f"\\]+', " ", ascii_name) + ascii_name = re.sub(r"\s+", " ", ascii_name).strip(" .") or "track" + + # RFC 5987 lets browsers use the actual UTF-8 title safely. + utf8_name = quote(f"{filename}.{extension}", safe="") + + return ( + f'{disposition}; filename="{ascii_name}.{extension}"; ' + f"filename*=UTF-8''{utf8_name}" + ) +def safe_download_filename(value:str)->str: + value=re.sub(r'[\x00-\x1f<>:"/\\|?*]+'," ",str(value or "track"));value=" ".join(value.split()).strip(" .") + return (value or "track")[:120] +def download_extension_from_audio_info(audio_info:dict,content_type:Optional[str]=None)->str: + ext=str(audio_info.get("ext") or "").strip().lower();content_type=str(content_type or "").split(";",1)[0].strip().lower() + if ext in {"m4a","mp4","mp3","aac","opus","ogg","webm"}:return "m4a" if ext=="mp4" else ext + if content_type in {"audio/mp4","audio/x-m4a","video/mp4"}:return "m4a" + if content_type in {"audio/mpeg","audio/mp3"}:return "mp3" + if content_type in {"audio/aac","audio/aacp"}:return "aac" + if content_type in {"audio/ogg","application/ogg"}:return "ogg" + if content_type in {"audio/webm","video/webm"}:return "webm" + return "m4a" +def download_content_type_for_extension(extension:str,fallback:Optional[str]=None)->str: + extension=str(extension or "").lower() + return {"m4a":"audio/mp4","mp3":"audio/mpeg","aac":"audio/aac","ogg":"audio/ogg","opus":"audio/ogg","webm":"audio/webm"}.get(extension,sanitize_content_type(fallback)) + +async def open_validated_upstream_stream( + *, + audio_url: str, + upstream_headers: dict[str, str], + method: str = "GET", + proxy_url: Optional[str] = None, +) -> tuple[httpx.AsyncClient, httpx.Response, str]: + base_url = validate_media_url(audio_url) + method = method.upper() + + if method not in {"GET", "HEAD"}: + raise ValueError("Unsupported upstream request method") + + if shared_http_client is None: + raise RuntimeError("Shared HTTP client is not initialized") + + redirect_statuses = {301, 302, 303, 307, 308} + retry_statuses = {408, 425, 429, 500, 502, 503, 504} + last_error: Optional[Exception] = None + + # The resolved media URL was obtained using proxy_url, so try that path + # first. Then alternative proxies, then direct as a final fallback. + attempts: list[Optional[str]] = [] + + if proxy_url: + attempts.append(proxy_url) + + for _ in range(PROXY_RETRY_ATTEMPTS): + retry_proxy = await get_next_proxy_url() + + if retry_proxy and retry_proxy not in attempts: + attempts.append(retry_proxy) + + if None not in attempts: + attempts.append(None) + + for attempt_index, proxy in enumerate(attempts, start=1): + client = shared_http_client + owns_client = False + current_url = base_url + + try: + if proxy: + client = httpx.AsyncClient( + timeout=HTTP_TIMEOUT, + follow_redirects=False, + limits=httpx.Limits( + max_connections=1, + max_keepalive_connections=0, + ), + proxy=proxy, + ) + owns_client = True + + for _ in range(MAX_MEDIA_REDIRECTS + 1): + upstream_request = client.build_request( + method, + current_url, + headers=upstream_headers, + ) + + response = await client.send(upstream_request, stream=True) + status = response.status_code + + if status in redirect_statuses: + location = response.headers.get("location") + await response.aclose() + + if not location: + raise HTTPException( + status_code=502, + detail="Audio upstream redirect was invalid", + ) + + current_url = urljoin(current_url, location) + validate_media_url(current_url) + continue + + if status < 400: + if proxy: + await mark_proxy_success(proxy) + + return client, response, current_url + + await response.aclose() + + # Caller must refresh the cached signed Googlevideo URL. + if status in {403, 404, 410}: + raise HTTPException( + status_code=status, + detail="Audio URL rejected; refresh required", + ) + + if status in retry_statuses: + last_error = RuntimeError( + f"Media upstream returned HTTP {status}" + ) + + if proxy: + await mark_proxy_failure(proxy) + + break + + raise HTTPException( + status_code=502, + detail=f"Audio upstream returned HTTP {status}", + ) + + else: + last_error = RuntimeError( + f"Too many upstream redirects for {urlparse(base_url).hostname}" + ) + + except HTTPException: + if owns_client: + await client.aclose() + raise + + except Exception as exc: + last_error = exc + + if proxy: + await mark_proxy_failure(proxy) + + logger.info( + "Media upstream failed " + "attempt=%s proxy=%s host=%s error=%r", + attempt_index, + "configured" if proxy else "direct", + urlparse(current_url).hostname, + exc, + ) + + finally: + # A successful response returns before this closes it. Failed + # proxy-specific clients must be closed here. + if owns_client: + try: + await client.aclose() + except Exception: + logger.debug("Failed closing temporary proxy HTTP client") + + if attempt_index < len(attempts): + await asyncio.sleep(min(0.4 * attempt_index, 2.0)) + + logger.warning( + "Media upstream permanently failed host=%s error=%r", + urlparse(base_url).hostname, + last_error, + ) + + raise HTTPException( + status_code=502, + detail="Audio temporarily unavailable", + ) from last_error + +@app.get("/api/debug/source") +async def debug_source(url:str=Query(...,min_length=1,max_length=2048)): + if APP_ENV=="production":raise HTTPException(404,"Not found") + canonical_url=canonical_youtube_url_from_url(url) or url + try: + raw=await run_ytdlp_in_executor(ytdlp_source_metadata_sync,canonical_url,None) + return {"ok":True,"id":raw.get("id"),"title":raw.get("title"),"channel":raw.get("channel") or raw.get("uploader")} + except Exception as exc: + logger.exception("Debug source extraction failed");raise HTTPException(502,str(exc)) from exc + +@app.get("/api/search") +async def search(q:str=Query(...,min_length=1,max_length=120),limit:int=Query(15,ge=1,le=25),page:int=Query(1,ge=1,le=10),request:Request=None): + try: + await rate_limit_request(request,"search",RATE_LIMIT_SEARCH_MAX,RATE_LIMIT_WINDOW_SECONDS) + start=(page-1)*limit;end=start+limit + if start>=MAX_SEARCH_FETCH:return {"results":[],"count":0,"query":q,"limit":limit,"page":page,"has_more":False,"next_page":None,"loaded":MAX_SEARCH_FETCH} + data=await ytdlp_search(q);all_results=data.get("results",[]);page_results=[{**item,"index":offset} for offset,item in enumerate(all_results[start:end],start=start+1)];has_more=len(all_results)>end and end 0, + ) + + audio_url = validate_media_url( + audio_info.get("audio_url") or "" + ) + + upstream_headers = filter_upstream_headers( + audio_info.get("http_headers") or {} + ) + upstream_headers.setdefault("User-Agent", DEFAULT_USER_AGENT) + upstream_headers.setdefault("Accept", "*/*") + upstream_headers["Accept-Encoding"] = "identity" + upstream_headers.setdefault("Referer", AUDIO_UPSTREAM_REFERER) + + if AUDIO_UPSTREAM_ORIGIN: + upstream_headers.setdefault( + "Origin", + AUDIO_UPSTREAM_ORIGIN, + ) + + # Preserve a browser Range request for seeking/resuming. + if range_header: + upstream_headers["Range"] = range_header + + upstream_client, upstream_response, resolved_url = ( + await open_validated_upstream_stream( + audio_url=audio_url, + upstream_headers=upstream_headers, + method=method, + proxy_url=audio_info.get("proxy_url"), + ) + ) + + status = upstream_response.status_code + + if status == 416: + await upstream_response.aclose() + upstream_response = None + await close_upstream_client() + + raise HTTPException( + status_code=416, + detail="Requested range not satisfiable", + ) + + if status >= 400: + await upstream_response.aclose() + upstream_response = None + await close_upstream_client() + + raise HTTPException( + status_code=502, + detail="Audio temporarily unavailable", + ) + + upstream_content_type = upstream_response.headers.get( + "content-type" + ) + media_type = sanitize_content_type(upstream_content_type) + + # Deliberately omit Content-Length and Content-Range. + # + # Googlevideo can close an upstream connection early. Forwarding + # those headers would promise an exact byte count to Uvicorn; + # when fewer bytes arrive, Uvicorn raises: + # "Response content shorter than Content-Length". + # + # With no fixed downstream length, Uvicorn uses chunked HTTP/1.1 + # to Nginx. Nginx/Caddy pass the stream through normally. + response_headers = { + "Cache-Control": "no-store", + "Accept-Ranges": "bytes", + } + + etag = sanitize_etag( + upstream_response.headers.get("etag") + ) + if etag: + response_headers["ETag"] = etag + + last_modified = sanitize_last_modified( + upstream_response.headers.get("last-modified") + ) + if last_modified: + response_headers["Last-Modified"] = last_modified + + download_requested = ( + bool(download_filename) + or request.query_params.get("download", "").lower() + in {"1", "true", "yes"} + ) + + filename = safe_download_filename( + download_filename + or audio_info.get("title") + or video_id + or "track" + ) + + extension = download_extension_from_audio_info( + audio_info, + upstream_content_type, + ) + + disposition = "attachment" if download_requested else "inline" + + # Header values must be Latin-1 compatible. The ASCII name is a + # fallback; filename*=UTF-8'' keeps Unicode names in browsers. + ascii_filename = ( + filename.encode("ascii", "ignore") + .decode("ascii") + .replace('"', "") + .replace("\\", "") + .strip(" .") + or "track" + ) + + utf8_filename = quote( + f"{filename}.{extension}", + safe="", + ) + + response_headers["Content-Disposition"] = ( + f'{disposition}; filename="{ascii_filename}.{extension}"; ' + f"filename*=UTF-8''{utf8_filename}" + ) + + if download_requested: + media_type = download_content_type_for_extension( + extension, + upstream_content_type, + ) + + if is_head: + await upstream_response.aclose() + upstream_response = None + await close_upstream_client() + await release_once() + + return StreamingResponse( + content=iter(()), + status_code=status, + media_type=media_type, + headers=response_headers, + ) + + stream_response = upstream_response + stream_client = upstream_client + + upstream_response = None + upstream_client = None + response_handed_off = True + + async def stream_generator(): + try: + async for chunk in stream_response.aiter_bytes( + chunk_size=CHUNK_SIZE + ): + if await request.is_disconnected(): + return + + if chunk: + yield chunk + + except httpx.ReadError as exc: + # Upstream disconnect after downstream streaming has + # begun. Do not re-raise: no fixed Content-Length is + # advertised, so ending this chunked response is safe. + logger.warning( + "Audio upstream closed early " + "video=%s ip=%s host=%s error=%r", + video_id, + client_ip, + urlparse(resolved_url).hostname, + exc, + ) + + except Exception as exc: + logger.warning( + "Audio stream interrupted " + "video=%s ip=%s host=%s error=%r", + video_id, + client_ip, + urlparse(resolved_url).hostname, + exc, + ) + + finally: + try: + await stream_response.aclose() + finally: + if stream_client is not shared_http_client: + await stream_client.aclose() + + await release_once() + + return StreamingResponse( + stream_generator(), + status_code=status, + media_type=media_type, + headers=response_headers, + ) + + except HTTPException as exc: + last_error = exc + + if upstream_response is not None: + await upstream_response.aclose() + upstream_response = None + + await close_upstream_client() + + # Signed Googlevideo URLs may be stale or rejected. Clear the + # cached resolved URL and re-run yt-dlp before retrying. + if exc.status_code in {403, 404, 410} and attempt < 2: + logger.info( + "Refreshing rejected audio URL " + "video=%s ip=%s attempt=%s", + video_id, + client_ip, + attempt + 1, + ) + + await clear_audio_cache(youtube_url) + await asyncio.sleep(min(0.25 * (attempt + 1), 1.0)) + continue + + raise + + except Exception as exc: + last_error = exc + + if upstream_response is not None: + await upstream_response.aclose() + upstream_response = None + + await close_upstream_client() + await clear_audio_cache(youtube_url) + + logger.warning( + "Audio stream setup failed " + "video=%s ip=%s attempt=%s error=%r", + video_id, + client_ip, + attempt + 1, + exc, + ) + + if attempt < 2: + await asyncio.sleep(min(0.25 * (attempt + 1), 1.0)) + continue + + logger.error( + "Audio stream setup permanently failed " + "video=%s ip=%s final_error=%r", + video_id, + client_ip, + last_error, + ) + + raise HTTPException( + status_code=502, + detail="Audio temporarily unavailable", + ) + + finally: + if not response_handed_off: + if upstream_response is not None: + await upstream_response.aclose() + + await close_upstream_client() + await release_once() + +#@app.api_route("/media/{media_token}",methods=["GET","HEAD"]) +#async def media_stream(media_token:str,request:Request): +# anonymous_id=get_anonymous_id(request,required=True);await touch_anonymous_user(anonymous_id);payload=verify_media_token(media_token,anonymous_id);video_id=payload["vid"];youtube_url=youtube_watch_url(video_id) +# if not youtube_url:raise HTTPException(status_code=404,detail="Invalid media token") +# return await stream_resolved_audio(request=request,video_id=video_id,youtube_url=youtube_url,anonymous_id=anonymous_id) + +#@app.get("/api/proxy-audio") +#async def proxy_audio(url:str,request:Request): +# anonymous_id=get_anonymous_id(request);await touch_anonymous_user(anonymous_id) +# if not is_allowed_url(url):raise HTTPException(400,"Unsupported URL") +# video_id=youtube_video_id_from_url(url) +# if not video_id or not is_valid_video_id(video_id):raise HTTPException(400,"Invalid YouTube video URL") +# return await stream_resolved_audio(request=request,video_id=video_id,youtube_url=youtube_watch_url(video_id),anonymous_id=anonymous_id) + +#@app.get("/api/download/{video_id}") +#async def download(video_id:str,request:Request,title:str=Query("track",max_length=200)): +# if not is_valid_video_id(video_id):raise HTTPException(400,"Invalid video id") +# anonymous_id=get_anonymous_id(request);await touch_anonymous_user(anonymous_id) +# return await stream_resolved_audio(request=request,video_id=video_id,youtube_url=youtube_watch_url(video_id),anonymous_id=anonymous_id,download_filename=safe_download_filename(title)) + +@app.api_route("/media/{media_token}", methods=["GET", "HEAD"]) +async def media_stream(media_token: str, request: Request): + anonymous_id = get_anonymous_id(request, required=True) + await touch_anonymous_user(anonymous_id) + + payload = verify_media_token(media_token, anonymous_id) + video_id = payload["vid"] + youtube_url = youtube_watch_url(video_id) + + if not youtube_url: + raise HTTPException(status_code=404, detail="Invalid media token") + + return await stream_resolved_audio( + request=request, + video_id=video_id, + youtube_url=youtube_url, + anonymous_id=anonymous_id, + ) + + +@app.get("/api/proxy-audio") +async def proxy_audio(url: str, request: Request): + anonymous_id = get_anonymous_id(request) + await touch_anonymous_user(anonymous_id) + + if not is_allowed_url(url): + raise HTTPException(status_code=400, detail="Unsupported URL") + + video_id = youtube_video_id_from_url(url) + if not video_id or not is_valid_video_id(video_id): + raise HTTPException( + status_code=400, + detail="Invalid YouTube video URL", + ) + + return await stream_resolved_audio( + request=request, + video_id=video_id, + youtube_url=youtube_watch_url(video_id), + anonymous_id=anonymous_id, + ) + + +@app.get("/api/download/{video_id}") +async def download( + video_id: str, + request: Request, + title: str = Query("track", max_length=200), +): + if not is_valid_video_id(video_id): + raise HTTPException(status_code=400, detail="Invalid video id") + + anonymous_id = get_anonymous_id(request) + await touch_anonymous_user(anonymous_id) + + return await stream_resolved_audio( + request=request, + video_id=video_id, + youtube_url=youtube_watch_url(video_id), + anonymous_id=anonymous_id, + download_filename=safe_download_filename(title), + )