src/router/xtc_recognize/recognize.py

"""识曲主流程：buffer -> PCM -> FP -> 网易云识曲 -> 批量补详情。"""
from __future__ import annotations

import logging
from typing import Any

from .amr_decoder import decode_to_pcm_s16le
from .fingerprint import generate_fp
from .ncm_proxy import audio_match, get_song_detail_batch

logger = logging.getLogger(__name__)


async def recognize_from_audio(
    audio_bytes: bytes,
    *,
    duration: int = 3,
    mime: str = "",
) -> dict:
    """主入口。

    Args:
        audio_bytes: 原始音频字节（amr/wav/mp3/m4a 等 ffmpeg 支持的格式）
        duration: 采样时长（秒）
        mime: MIME 类型（仅用于日志）

    Returns:
        {fp: str, list: [...], raw_match: dict}
    """
    logger.info(
        "[recognize] start, input=%d bytes mime=%s duration=%d",
        len(audio_bytes), mime, duration,
    )

    # 1) 转码
    pcm = await decode_to_pcm_s16le(audio_bytes, duration_sec=duration, mime=mime)

    # 2) 指纹
    fp = await generate_fp(pcm)

    # 3) 识曲
    match_res = await audio_match(fp, duration)
    candidates = ((match_res.get("data") or {}).get("result")) or []
    logger.info("[recognize] match candidates=%d", len(candidates))

    if not candidates:
        return {"fp": fp, "list": [], "raw_match": match_res}

    # 4) 批量补全详情
    ids = [str(c.get("song", {}).get("id")) for c in candidates if c.get("song", {}).get("id")]
    details = await get_song_detail_batch(ids) if ids else []
    detail_map = {str(d["id"]): d for d in details if d}

    # 5) 组装结果，按 score 排序
    out_list = []
    for c in candidates:
        song = c.get("song") or {}
        sid = str(song.get("id")) if song.get("id") else ""
        d = detail_map.get(sid, {})
        out_list.append({
            "id": song.get("id"),
            "name": d.get("name") or song.get("name"),
            "artists": d.get("artists") or "/".join(
                a.get("name", "") for a in song.get("artists", [])
            ),
            "album": d.get("album") or (song.get("album") or {}).get("name"),
            "cover": d.get("cover") or (song.get("album") or {}).get("picUrl"),
            "duration": d.get("duration"),
            "matchScore": c.get("score"),
            "startTime": c.get("startTime"),
        })
    out_list.sort(key=lambda x: (x.get("matchScore") or 0), reverse=True)

    return {"fp": fp, "list": out_list, "raw_match": match_res}