File size: 2,448 Bytes
d731d8e 6d61ecf 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 | """识曲主流程：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}
|