File size: 5,261 Bytes
d731d8e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | """XTC 听歌识曲 FastAPI 路由。
无侵入集成到 gcli2api：所有路径前缀 /xtc，不与现有路由冲突。
"""
from __future__ import annotations
import base64
import logging
from typing import Any, Optional
from fastapi import APIRouter, Query, Request
from fastapi.responses import JSONResponse
from .ncm_proxy import get_song_detail, get_song_url
from .recognize import recognize_from_audio
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/xtc", tags=["XTC Recognize"])
# 可选鉴权：环境变量 XTC_API_KEY 设置后，请求头 X-API-Key 需匹配
import os
_API_KEY = os.environ.get("XTC_API_KEY", "")
def _check_key(request: Request) -> Optional[JSONResponse]:
"""校验 X-API-Key。未配置则放行。"""
if not _API_KEY:
return None
provided = request.headers.get("X-API-Key", "")
if provided == _API_KEY:
return None
return JSONResponse(
status_code=401,
content={"code": 401, "msg": "unauthorized"},
)
def _ok(data: Any) -> JSONResponse:
return JSONResponse(
status_code=200,
content={"code": 200, "data": data},
headers={
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET,POST,OPTIONS",
"Access-Control-Allow-Headers": "Content-Type, X-API-Key",
},
)
def _err(msg: str, status: int = 500) -> JSONResponse:
return JSONResponse(
status_code=status,
content={"code": status, "msg": msg},
headers={
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET,POST,OPTIONS",
"Access-Control-Allow-Headers": "Content-Type, X-API-Key",
},
)
@router.options("/{path:path}")
async def cors_preflight(path: str) -> JSONResponse:
return JSONResponse(
status_code=204,
content=None,
headers={
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Methods": "GET,POST,OPTIONS",
"Access-Control-Allow-Headers": "Content-Type, X-API-Key",
"Access-Control-Max-Age": "86400",
},
)
@router.get("/health")
async def health() -> JSONResponse:
"""健康检查。"""
import shutil
node_ok = shutil.which("node") is not None
ffmpeg_ok = shutil.which("ffmpeg") is not None
return _ok({
"service": "xtc-recognize",
"node": node_ok,
"ffmpeg": ffmpeg_ok,
"auth": bool(_API_KEY),
})
@router.post("/recognize")
async def recognize(request: Request) -> JSONResponse:
"""识曲接口。
Body: JSON
{
"audio": "<base64 编码的音频字节>",
"mime": "audio/amr", # 可选
"name": "record.amr", # 可选
"duration": 3 # 可选，默认 3
}
Returns:
{code:200, data:{fp, list:[{id,name,artists,album,cover,duration,matchScore,startTime}], raw_match}}
"""
auth_err = _check_key(request)
if auth_err:
return auth_err
try:
body = await request.json()
if not isinstance(body, dict):
return _err("invalid body", 400)
audio_b64 = body.get("audio")
if not audio_b64:
return _err("audio required", 400)
duration = int(body.get("duration") or 3)
if duration < 1 or duration > 10:
duration = 3
mime = str(body.get("mime") or "")
name = str(body.get("name") or "")
audio_bytes = base64.b64decode(audio_b64)
if len(audio_bytes) > 2 * 1024 * 1024:
return _err("audio too large (max 2MB)", 400)
result = await recognize_from_audio(
audio_bytes,
duration=duration,
mime=mime,
)
# raw_match 体积大且含敏感字段，对外只返回精简后的 list
return _ok({
"fp": result["fp"],
"list": result["list"],
"count": len(result["list"]),
})
except Exception as e:
logger.exception("[recognize] error")
return _err(str(e) or "recognize failed")
@router.get("/song/url")
async def song_url(
request: Request,
id: str = Query(...),
unblock: str = Query("false"),
level: str = Query("standard"),
) -> JSONResponse:
"""获取试听 URL。
公开接口仅返回 30s 试听片段（VIP 歌曲）。unblock 暂不支持。
"""
auth_err = _check_key(request)
if auth_err:
return auth_err
try:
data = await get_song_url(id, unblock=(unblock == "true"), level=level)
return _ok(data)
except Exception as e:
logger.exception("[song/url] error")
return _err(str(e) or "get url failed")
@router.get("/song/detail")
async def song_detail(
request: Request,
id: str = Query(...),
) -> JSONResponse:
"""获取歌曲详情。"""
auth_err = _check_key(request)
if auth_err:
return auth_err
try:
data = await get_song_detail(id)
if not data:
return _err("song not found", 404)
return _ok(data)
except Exception as e:
logger.exception("[song/detail] error")
return _err(str(e) or "get detail failed")
|