35 lines
1.2 KiB
Python
35 lines
1.2 KiB
Python
"""媒体代理：m3u8 播放列表转发（源站 playlist 内 TS 链接带时效签名，每次回源拉取新鲜版本）"""
|
|
import re
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Query, Response
|
|
|
|
from app.core.errors import NOT_FOUND, PARAM_ERROR
|
|
from app.core.response import BizException
|
|
from app.services.collection_service import HEADERS, UPSTREAM
|
|
|
|
router = APIRouter(prefix="/m3u8", tags=["媒体"])
|
|
|
|
# 只允许相对资源路径，防 SSRF
|
|
_PATH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9/._-]*$")
|
|
|
|
|
|
@router.get("/proxy")
|
|
async def m3u8_proxy(path: str = Query(min_length=1, max_length=300)):
|
|
if ".." in path or not _PATH_RE.match(path):
|
|
raise BizException(PARAM_ERROR, "非法路径")
|
|
|
|
url = f"{UPSTREAM}/api/v1/m3u8/proxy"
|
|
async with httpx.AsyncClient(follow_redirects=True) as client:
|
|
resp = await client.get(url, params={"path": path}, headers=HEADERS, timeout=20)
|
|
|
|
text = resp.text
|
|
if resp.status_code != 200 or not text.lstrip().startswith("#EXTM3U"):
|
|
raise BizException(NOT_FOUND, "播放列表获取失败")
|
|
|
|
return Response(
|
|
content=text,
|
|
media_type="application/vnd.apple.mpegurl",
|
|
headers={"Cache-Control": "no-cache", "Access-Control-Allow-Origin": "*"},
|
|
)
|