src/router/xtc_recognize/router.py

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