2370 lines
117 KiB
Python
2370 lines
117 KiB
Python
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_AT<CACHE_CLEANUP_INTERVAL_SECONDS:return
|
||
async with CACHE_GUARD:
|
||
now=time.time()
|
||
if now-CACHE_LAST_CLEANUP_AT<CACHE_CLEANUP_INTERVAL_SECONDS:return
|
||
await sqlite_db.execute("DELETE FROM cache WHERE key IN (SELECT key FROM cache WHERE expires_at<=? LIMIT ?)",(now,CACHE_CLEANUP_BATCH_SIZE))
|
||
await sqlite_db.commit()
|
||
CACHE_LAST_CLEANUP_AT=now
|
||
|
||
async def cache_get(key:str)->Optional[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 deleted<CACHE_CLEANUP_BATCH_SIZE:break
|
||
CACHE_LAST_CLEANUP_AT=now
|
||
except Exception:logger.exception("Cache cleanup failed")
|
||
|
||
async def get_key_lock(key:str)->asyncio.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)(?<!\w)(official audio|official music video|official video|music video|lyric video|lyrics|lyric)(?!\w)")
|
||
RECOMMENDATION_FREE_RE=re.compile(r"(?i)(?:\[\s*free\s*\]|\*\s*free\s*\*|\(\s*free\s*\))")
|
||
RECOMMENDATION_VARIANT_PATTERNS=(("chopped_and_screwed",re.compile(r"(?i)(?<!\w)chopped\s+and\s+screwed(?!\w)"),-0.10),("sped_up",re.compile(r"(?i)(?<!\w)sped\s+up(?!\w)"),-0.15),("slowed",re.compile(r"(?i)(?<!\w)slowed(?!\w)"),-0.12),("nightcore",re.compile(r"(?i)(?<!\w)nightcore(?!\w)"),-0.20),("screwed",re.compile(r"(?i)(?<!\w)screwed(?!\w)"),-0.12),("mix",re.compile(r"(?i)(?<!\w)mix(?!\w)"),-0.10),("playlist",re.compile(r"(?i)(?<!\w)playlist(?!\w)"),-0.10))
|
||
RECOMMENDATION_COVER_RE=re.compile(r"(?i)(?<!\w)cover(?!\w)")
|
||
RECOMMENDATION_CONCERT_RE=re.compile(r"(?i)(?<!\w)concert(?!\w)")
|
||
RECOMMENDATION_NOISE_PHRASES=("official music video","official video","official audio","music video","lyric video","lyrics","lyric","visualizer","HD","HQ","4K")
|
||
GENERIC_RECOMMENDATION_TAGS={"music","official","video","audio","song","lyrics","lyric"}
|
||
COMMON_GENRE_CANONICAL={"alternative":"alternative","ambient":"ambient","blues":"blues","classical":"classical","country":"country","dance":"dance","disco":"disco","edm":"edm","electronic":"electronic","folk":"folk","funk":"funk","gospel":"gospel","hip hop":"hip hop","hip-hop":"hip hop","house":"house","indie":"indie","j pop":"j-pop","j-pop":"j-pop","jazz":"jazz","k pop":"k-pop","k-pop":"k-pop","latin":"latin","metal":"metal","pop":"pop","punk":"punk","r b":"r&b","r&b":"r&b","rap":"rap","reggae":"reggae","rock":"rock","soul":"soul","techno":"techno","trance":"trance"}
|
||
COMMON_GENRE_PATTERNS=sorted(COMMON_GENRE_CANONICAL.items(),key=lambda item:len(item[0]),reverse=True)
|
||
#RECOMMENDATION_PENALTY_RE=re.compile(r"(?i)(?<!\w)(lyrics|lyric video|sped up|slowed|nightcore|1 hour|loop)(?!\w)")
|
||
RECOMMENDATION_PENALTY_RE = re.compile(
|
||
r"(?i)(?<!\w)(lyrics|lyric video|sped up|slowed|nightcore|1 hour|loop|free\s+for\s+profit|free\s+to\s+use)(?!\w)|\[\s*free[^\]]*\]|\(\s*free[^)]*\)|\*\s*free[^*]*\*"
|
||
)
|
||
def recommendation_duration_seconds(value:Any)->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)(?<!\w){pattern}(?!\w)"," ",text)
|
||
return re.sub(r"\s+"," ",text).strip(" -_|/\\:;,.")
|
||
def recommendation_identity(value:Any)->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"(?<!\w){re.escape(genre_key)}(?!\w)",normalized):return canonical
|
||
return None
|
||
|
||
async def run_ytdlp_in_executor(func,*args):
|
||
async with YTDLP_SEMAPHORE:return await asyncio.get_running_loop().run_in_executor(None,func,*args)
|
||
|
||
def ytdlp_source_metadata_sync(url:str,proxy_url:Optional[str])->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 attempt<PROXY_RETRY_ATTEMPTS else None
|
||
try:
|
||
raw=await run_ytdlp_in_executor(ytdlp_source_metadata_sync,canonical_url,proxy_url)
|
||
data=normalize_music_source_metadata(raw,fallback_url=canonical_url)
|
||
await cache_set(cache_key,data,SOURCE_METADATA_CACHE_TTL);await cache_delete(negative_key)
|
||
if proxy_url:await mark_proxy_success(proxy_url)
|
||
return data
|
||
except Exception as exc:
|
||
last_error=exc
|
||
if proxy_url:await mark_proxy_failure(proxy_url)
|
||
if attempt<PROXY_RETRY_ATTEMPTS:await asyncio.sleep(min(.5*(attempt+1),2))
|
||
except Exception as exc:last_error=exc
|
||
logger.warning("Source metadata failed url=%s error=%s",canonical_url,last_error)
|
||
await cache_set(negative_key,{"failed":True},min(NEGATIVE_CACHE_TTL,30))
|
||
raise ValueError("Recommendations temporarily unavailable") from last_error
|
||
|
||
def normalize_music_source_metadata(raw:dict,fallback_url:str="")->dict:
|
||
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 attempt<PROXY_RETRY_ATTEMPTS else None
|
||
try:
|
||
data=await run_ytdlp_in_executor(ytdlp_recommendation_search_sync,query,per_query,proxy_url)
|
||
if proxy_url:await mark_proxy_success(proxy_url)
|
||
return data or {}
|
||
except Exception as exc:
|
||
last_error=exc
|
||
if proxy_url:await mark_proxy_failure(proxy_url)
|
||
if attempt<PROXY_RETRY_ATTEMPTS:await asyncio.sleep(min(.5*(attempt+1),2))
|
||
logger.warning("Recommendation search failed query=%r error=%s",query,last_error)
|
||
return {}
|
||
|
||
def is_hard_blocked_recommendation(candidate:dict)->bool:
|
||
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<PROXY_RETRY_ATTEMPTS else None
|
||
# try:
|
||
# result=await run_ytdlp_in_executor(resolve_audio_sync,url,proxy)
|
||
# if proxy:await mark_proxy_success(proxy)
|
||
# return result
|
||
# except Exception as exc:
|
||
# last_error=exc
|
||
# if is_ytdlp_bot_verification_error(exc):raise
|
||
# if proxy:await mark_proxy_failure(proxy)
|
||
# if attempt<PROXY_RETRY_ATTEMPTS:await asyncio.sleep(min(.25*(attempt+1),1))
|
||
# raise ValueError(str(last_error or "Audio temporarily unavailable"))
|
||
async def resolve_audio(url: str) -> 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)<int(start):raise HTTPException(status_code=416,detail="Invalid Range header")
|
||
return value
|
||
|
||
async def acquire_stream_slot(client_ip:str,anonymous_id:Optional[str]):
|
||
acquired=False
|
||
try:
|
||
await asyncio.wait_for(ACTIVE_STREAMS_SEMAPHORE.acquire(),timeout=.1);acquired=True
|
||
except asyncio.TimeoutError as exc:raise HTTPException(status_code=503,detail="Audio capacity unavailable") from exc
|
||
try:
|
||
async with ACTIVE_STREAMS_GUARD:
|
||
ip_count=ACTIVE_STREAMS_BY_IP.get(client_ip,0)
|
||
if ip_count>=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<MAX_SEARCH_FETCH
|
||
return {"results":page_results,"count":len(page_results),"query":q,"limit":limit,"page":page,"has_more":has_more,"next_page":page+1 if has_more else None,"loaded":min(end,len(all_results))}
|
||
except HTTPException:raise
|
||
except Exception as exc:
|
||
logger.exception("Search failed for query=%r",q);raise HTTPException(502,"Search temporarily unavailable") from exc
|
||
|
||
@app.get("/api/recommendations")
|
||
async def recommendations(url:str=Query(...,min_length=1,max_length=2048),total:int=Query(20,ge=1,le=100),per_query:int=Query(8,ge=1,le=25),request:Request=None):
|
||
try:
|
||
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")
|
||
await rate_limit_request(request,"search",RATE_LIMIT_SEARCH_MAX,RATE_LIMIT_WINDOW_SECONDS)
|
||
return await get_music_recommendations(youtube_watch_url(video_id),total=total,per_query=per_query)
|
||
except HTTPException:raise
|
||
except ValueError as exc:raise HTTPException(502,str(exc)) from exc
|
||
except Exception as exc:
|
||
logger.exception("Recommendations failed for url=%r",url);raise HTTPException(502,"Recommendations temporarily unavailable") from exc
|
||
|
||
@app.get("/api/favorites")
|
||
async def get_favorites(request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);items=[]
|
||
for favorite in await list_favorites(user_id):
|
||
item=dict(favorite["item"]);item["favorited"]=True;item["id"]=item.get("id") or favorite["id"];items.append(item)
|
||
return {"results":items,"count":len(items)}
|
||
|
||
@app.post("/api/favorites/toggle")
|
||
async def toggle_favorite(body:FavoriteToggleBody,request:Request):
|
||
if not sqlite_db:raise HTTPException(503,"cache not initialized")
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);item=body.item.model_dump(exclude_none=True);favorite_id=item["id"];existing=await get_favorite_by_id(user_id,favorite_id)
|
||
if existing:
|
||
await sqlite_db.execute("DELETE FROM favorites WHERE user_id=? AND id=?",(user_id,favorite_id));await sqlite_db.commit();return {"favorited":False,"id":favorite_id}
|
||
now=time.time();await sqlite_db.execute("INSERT INTO favorites (user_id,id,data,created_at,updated_at) VALUES (?,?,?,?,?)",(user_id,favorite_id,json.dumps(item,ensure_ascii=False),now,now));await sqlite_db.commit()
|
||
return {"favorited":True,"id":favorite_id}
|
||
|
||
@app.get("/api/playlists")
|
||
async def get_playlists(request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id)
|
||
async with sqlite_db.execute("SELECT p.id,p.name,p.share_token,p.shared_at,p.created_at,p.updated_at,COUNT(pi.id) FROM playlists p LEFT JOIN playlist_items pi ON pi.playlist_id=p.id AND pi.user_id=p.user_id WHERE p.user_id=? GROUP BY p.id,p.name,p.share_token,p.shared_at,p.created_at,p.updated_at ORDER BY p.updated_at DESC",(user_id,)) as cursor:rows=await cursor.fetchall()
|
||
results=[{"id":row[0],"name":row[1],"count":row[6],"share_token":row[2],"share_url":f"/share/{row[2]}" if row[2] else None,"shared_at":row[3],"created_at":row[4],"updated_at":row[5]} for row in rows]
|
||
return {"results":results,"count":len(results)}
|
||
|
||
@app.post("/api/playlists")
|
||
async def create_playlist(body:PlaylistCreateBody,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);name=clean_playlist_name(body.name)
|
||
if not name:raise HTTPException(400,"Missing playlist name")
|
||
playlist_id=str(uuid.uuid4());now=time.time();await sqlite_db.execute("INSERT INTO playlists (id,user_id,name,created_at,updated_at) VALUES (?,?,?,?,?)",(playlist_id,user_id,name,now,now));await sqlite_db.commit()
|
||
return {"id":playlist_id,"name":name,"count":0,"created_at":now,"updated_at":now}
|
||
|
||
@app.patch("/api/playlists/{playlist_id}")
|
||
async def rename_playlist(playlist_id:str,body:PlaylistRenameBody,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);name=clean_playlist_name(body.name)
|
||
if not name:raise HTTPException(400,"Missing playlist name")
|
||
now=time.time();cursor=await sqlite_db.execute("UPDATE playlists SET name=?,updated_at=? WHERE user_id=? AND id=?",(name,now,user_id,playlist_id));await sqlite_db.commit()
|
||
if not cursor.rowcount:raise HTTPException(404,"Playlist not found")
|
||
return {"id":playlist_id,"name":name,"updated_at":now}
|
||
|
||
@app.get("/api/playlists/{playlist_id}")
|
||
async def get_playlist(playlist_id:str,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);playlist=await get_playlist_for_user(user_id,playlist_id)
|
||
if not playlist:raise HTTPException(404,"Playlist not found")
|
||
items=await get_playlist_items(user_id,playlist_id);playlist.update({"items":items,"count":len(items),"total_duration":playlist_total_duration(items),"share_url":f"/share/{playlist['share_token']}" if playlist.get("share_token") else None})
|
||
return playlist
|
||
|
||
@app.post("/api/playlists/{playlist_id}/items")
|
||
async def add_playlist_item(playlist_id:str,body:PlaylistItemAddBody,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id)
|
||
if not await get_playlist_for_user(user_id,playlist_id):raise HTTPException(404,"Playlist not found")
|
||
item=body.item.model_dump(exclude_none=True);item_id=item["id"];await sqlite_db.execute("BEGIN IMMEDIATE")
|
||
try:
|
||
async with sqlite_db.execute("SELECT id FROM playlist_items WHERE user_id=? AND playlist_id=? AND item_id=?",(user_id,playlist_id,item_id)) as cursor:
|
||
if await cursor.fetchone():
|
||
await sqlite_db.rollback();return {"added":False,"duplicate":True,"playlist_id":playlist_id,"item_id":item_id}
|
||
async with sqlite_db.execute("SELECT COALESCE(MAX(position),0) FROM playlist_items WHERE user_id=? AND playlist_id=?",(user_id,playlist_id)) as cursor:position=(await cursor.fetchone())[0]+1
|
||
playlist_item_id=str(uuid.uuid4());now=time.time()
|
||
await sqlite_db.execute("INSERT INTO playlist_items (id,user_id,playlist_id,item_id,data,position,created_at) VALUES (?,?,?,?,?,?,?)",(playlist_item_id,user_id,playlist_id,item_id,json.dumps(item,ensure_ascii=False),position,now))
|
||
await sqlite_db.execute("UPDATE playlists SET updated_at=? WHERE user_id=? AND id=?",(now,user_id,playlist_id));await sqlite_db.commit()
|
||
except Exception:
|
||
await sqlite_db.rollback();raise
|
||
return {"added":True,"duplicate":False,"playlist_id":playlist_id,"playlist_item_id":playlist_item_id,"item_id":item_id,"position":position}
|
||
|
||
@app.delete("/api/playlists/{playlist_id}/items/{playlist_item_id}")
|
||
async def delete_playlist_item(playlist_id:str,playlist_item_id:str,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);await sqlite_db.execute("BEGIN IMMEDIATE")
|
||
try:
|
||
cursor=await sqlite_db.execute("DELETE FROM playlist_items WHERE user_id=? AND playlist_id=? AND id=?",(user_id,playlist_id,playlist_item_id));deleted=cursor.rowcount
|
||
if deleted:await sqlite_db.execute("UPDATE playlists SET updated_at=? WHERE user_id=? AND id=?",(time.time(),user_id,playlist_id))
|
||
await sqlite_db.commit()
|
||
except Exception:
|
||
await sqlite_db.rollback();raise
|
||
if not deleted:raise HTTPException(404,"Playlist item not found")
|
||
return {"deleted":True,"playlist_id":playlist_id,"playlist_item_id":playlist_item_id}
|
||
|
||
|
||
#@app.patch("/api/playlists/{playlist_id}/items/reorder")
|
||
#async def reorder_playlist_items(playlist_id:str,body:PlaylistReorderBody,request:Request):
|
||
# user_id=get_anonymous_id(request);await touch_anonymous_user(user_id)
|
||
# if not await get_playlist_for_user(user_id,playlist_id):raise HTTPException(404,"Playlist not found")
|
||
# item_ids=body.item_ids
|
||
# if len(item_ids)!=len(set(item_ids)):raise HTTPException(400,"Duplicate item ids")
|
||
# await sqlite_db.execute("BEGIN IMMEDIATE")
|
||
# try:
|
||
# async with sqlite_db.execute("SELECT id FROM playlist_items WHERE user_id=? AND playlist_id=?",(user_id,playlist_id)) as cursor:existing_ids={row[0] for row in await cursor.fetchall()}
|
||
# if set(item_ids)!=existing_ids:
|
||
# await sqlite_db.rollback();raise HTTPException(400,"Item ids do not match playlist")
|
||
# now=time.time()
|
||
# for position,item_id in enumerate(item_ids,start=1):await sqlite_db.execute("UPDATE playlist_items SET position=? WHERE user_id=? AND playlist_id=? AND id=?",(position,user_id,playlist_id,item_id))
|
||
# await sqlite_db.execute("UPDATE playlists SET updated_at=? WHERE user_id=? AND id=?",(now,user_id,playlist_id));await sqlite_db.commit()
|
||
# except HTTPException:raise
|
||
# except Exception:
|
||
# await sqlite_db.rollback();raise
|
||
# return {"reordered":True,"playlist_id":playlist_id,"count":len(item_ids)}
|
||
@app.patch("/api/playlists/{playlist_id}/items/reorder")
|
||
async def reorder_playlist_items(
|
||
playlist_id: str,
|
||
body: PlaylistItemReorderBody,
|
||
request: Request,
|
||
):
|
||
user_id = get_anonymous_id(request)
|
||
await touch_anonymous_user(user_id)
|
||
|
||
if not sqlite_db:
|
||
raise HTTPException(503, "cache not initialized")
|
||
|
||
if not await get_playlist_for_user(user_id, playlist_id):
|
||
raise HTTPException(404, "Playlist not found")
|
||
|
||
item_ids = body.item_ids
|
||
|
||
if len(item_ids) != len(set(item_ids)):
|
||
raise HTTPException(400, "Duplicate item ids")
|
||
|
||
await sqlite_db.execute("BEGIN IMMEDIATE")
|
||
try:
|
||
async with sqlite_db.execute(
|
||
"""
|
||
SELECT id
|
||
FROM playlist_items
|
||
WHERE user_id = ? AND playlist_id = ?
|
||
ORDER BY position ASC, created_at ASC
|
||
""",
|
||
(user_id, playlist_id),
|
||
) as cur:
|
||
existing_ids = [row[0] for row in await cur.fetchall()]
|
||
|
||
if set(item_ids) != set(existing_ids):
|
||
await sqlite_db.rollback()
|
||
raise HTTPException(400, "Item ids do not match playlist")
|
||
|
||
now = time.time()
|
||
|
||
for position, item_id in enumerate(item_ids, start=1):
|
||
await sqlite_db.execute(
|
||
"""
|
||
UPDATE playlist_items
|
||
SET position = ?
|
||
WHERE user_id = ? AND playlist_id = ? AND id = ?
|
||
""",
|
||
(position, user_id, playlist_id, item_id),
|
||
)
|
||
|
||
await sqlite_db.execute(
|
||
"""
|
||
UPDATE playlists
|
||
SET updated_at = ?
|
||
WHERE user_id = ? AND id = ?
|
||
""",
|
||
(now, user_id, playlist_id),
|
||
)
|
||
await sqlite_db.commit()
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception:
|
||
await sqlite_db.rollback()
|
||
raise
|
||
|
||
return {
|
||
"reordered": True,
|
||
"playlist_id": playlist_id,
|
||
"count": len(item_ids),
|
||
}
|
||
|
||
@app.post("/api/playlists/{playlist_id}/items/bulk-delete")
|
||
async def bulk_delete_playlist_items(playlist_id:str,body:PlaylistItemsDeleteBody,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id)
|
||
if not await get_playlist_for_user(user_id,playlist_id):raise HTTPException(404,"Playlist not found")
|
||
item_ids=list(dict.fromkeys(body.item_ids));placeholders=",".join("?" for _ in item_ids);await sqlite_db.execute("BEGIN IMMEDIATE")
|
||
try:
|
||
cursor=await sqlite_db.execute(f"DELETE FROM playlist_items WHERE user_id=? AND playlist_id=? AND id IN ({placeholders})",[user_id,playlist_id,*item_ids]);deleted=cursor.rowcount
|
||
if deleted:await sqlite_db.execute("UPDATE playlists SET updated_at=? WHERE user_id=? AND id=?",(time.time(),user_id,playlist_id))
|
||
await sqlite_db.commit()
|
||
except Exception:
|
||
await sqlite_db.rollback();raise
|
||
return {"deleted":deleted,"playlist_id":playlist_id,"item_ids":item_ids}
|
||
|
||
@app.post("/api/playlists/{playlist_id}/share")
|
||
async def share_playlist(playlist_id:str,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);playlist=await get_playlist_for_user(user_id,playlist_id)
|
||
if not playlist:raise HTTPException(404,"Playlist not found")
|
||
share_token=playlist.get("share_token") or uuid.uuid4().hex;now=time.time()
|
||
await sqlite_db.execute("UPDATE playlists SET share_token=?,shared_at=COALESCE(shared_at,?),updated_at=? WHERE user_id=? AND id=?",(share_token,now,now,user_id,playlist_id));await sqlite_db.commit()
|
||
return {"shared":True,"share_token":share_token,"share_url":f"/share/{share_token}"}
|
||
|
||
@app.delete("/api/playlists/{playlist_id}/share")
|
||
async def unshare_playlist(playlist_id:str,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);cursor=await sqlite_db.execute("UPDATE playlists SET share_token=NULL,shared_at=NULL,updated_at=? WHERE user_id=? AND id=?",(time.time(),user_id,playlist_id));await sqlite_db.commit()
|
||
if not cursor.rowcount:raise HTTPException(404,"Playlist not found")
|
||
return {"shared":False,"playlist_id":playlist_id}
|
||
|
||
@app.get("/api/shared-playlists/{share_token}")
|
||
async def get_shared_playlist(share_token:str):
|
||
if not SHARE_TOKEN_RE.fullmatch(share_token):raise HTTPException(400,"Invalid share token")
|
||
async with sqlite_db.execute("SELECT id,user_id,name,created_at,updated_at,shared_at FROM playlists WHERE share_token=?",(share_token,)) as cursor:row=await cursor.fetchone()
|
||
if not row:raise HTTPException(404,"Shared playlist not found")
|
||
playlist_id,user_id,name,created_at,updated_at,shared_at=row;items=await get_playlist_items(user_id,playlist_id)
|
||
return {"id":playlist_id,"name":name,"shared":True,"shared_at":shared_at,"created_at":created_at,"updated_at":updated_at,"count":len(items),"total_duration":playlist_total_duration(items),"items":items}
|
||
|
||
@app.delete("/api/playlists/{playlist_id}")
|
||
async def delete_playlist(playlist_id:str,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);cursor=await sqlite_db.execute("DELETE FROM playlists WHERE user_id=? AND id=?",(user_id,playlist_id));await sqlite_db.commit()
|
||
if not cursor.rowcount:raise HTTPException(404,"Playlist not found")
|
||
return {"deleted":True,"playlist_id":playlist_id}
|
||
|
||
@app.get("/api/player-state")
|
||
async def get_saved_player_state(request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id)
|
||
return await get_player_state(user_id) or {"current_item":None,"current_time":0,"queue":[],"queue_index":-1,"autoplay":True,"shuffle":False,"loop_mode":"off","volume":1.0,"queue_mode":"queue","source_url":None}
|
||
|
||
@app.put("/api/player-state")
|
||
async def put_player_state(body:PlayerStateBody,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);await save_player_state(user_id,body.model_dump(exclude_none=True))
|
||
return {"saved":True}
|
||
|
||
@app.post("/api/audio-sessions")
|
||
async def create_audio_session(body:AudioSessionCreateBody,request:Request):
|
||
user_id=get_anonymous_id(request);await touch_anonymous_user(user_id);await rate_limit_request(request,"audio",RATE_LIMIT_AUDIO_MAX,RATE_LIMIT_WINDOW_SECONDS)
|
||
token,expires_in=create_media_token(body.video_id,user_id);asyncio.create_task(prewarm_audio_for_video(body.video_id))
|
||
return {"stream_url":f"/media/{token}","expires_in":expires_in}
|
||
|
||
async def stream_resolved_audio(
|
||
*,
|
||
request: Request,
|
||
video_id: str,
|
||
youtube_url: str,
|
||
anonymous_id: Optional[str],
|
||
download_filename: Optional[str] = None,
|
||
):
|
||
client_ip = get_client_ip(request)
|
||
method = request.method.upper()
|
||
is_head = method == "HEAD"
|
||
range_header = validate_range_header(request.headers.get("range"))
|
||
|
||
await rate_limit_request(
|
||
request,
|
||
"audio",
|
||
RATE_LIMIT_AUDIO_MAX,
|
||
RATE_LIMIT_WINDOW_SECONDS,
|
||
)
|
||
await acquire_stream_slot(client_ip, anonymous_id)
|
||
|
||
upstream_response: Optional[httpx.Response] = None
|
||
upstream_client: Optional[httpx.AsyncClient] = None
|
||
response_handed_off = False
|
||
slot_released = False
|
||
last_error: Optional[Exception] = None
|
||
|
||
async def close_upstream_client():
|
||
nonlocal upstream_client
|
||
|
||
if (
|
||
upstream_client is not None
|
||
and upstream_client is not shared_http_client
|
||
):
|
||
try:
|
||
await upstream_client.aclose()
|
||
finally:
|
||
upstream_client = None
|
||
|
||
async def release_once():
|
||
nonlocal slot_released
|
||
|
||
if not slot_released:
|
||
await release_stream_slot(client_ip, anonymous_id)
|
||
slot_released = True
|
||
|
||
try:
|
||
# Retries are safe only before headers/body are returned to the client.
|
||
for attempt in range(3):
|
||
try:
|
||
audio_info = await get_cached_audio_info(
|
||
youtube_url,
|
||
force_refresh=attempt > 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),
|
||
)
|