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), )