src/router/xtc_recognize/ncm_proxy.py

"""网易云 API 代理：识曲 / 歌曲 URL / 歌曲详情。

所有调用直连网易云公开接口，不依赖 api-enhanced 子服务。
"""
from __future__ import annotations

import logging
from typing import Any, Optional

import httpx

logger = logging.getLogger(__name__)

_NCM_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Linux; Android 10) AppleWebKit/537.36",
    "Referer": "https://music.163.com",
}
_TIMEOUT = 10.0


async def audio_match(audio_fp: str, duration: int = 3) -> dict:
    """调用网易云听歌识曲接口。

    Args:
        audio_fp: base64 编码的音频指纹
        duration: 录音时长（秒）

    Returns:
        网易云原始响应 dict，结构 {data: {result: [{song, startTime, score}]}}
    """
    url = "https://interface.music.163.com/api/music/audio/match"
    params = {
        "sessionId": "0123456789abcdef",
        "algorithmCode": "shazam_v2",
        "duration": str(duration),
        "rawdata": audio_fp,
        "times": "1",
        "decrypt": "1",
    }
    async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
        resp = await client.get(url, params=params, headers=_NCM_HEADERS)
        resp.raise_for_status()
        return resp.json()


async def get_song_url(song_id: str, *, unblock: bool = False, level: str = "standard") -> dict:
    """获取试听 URL。

    优先调用网易云公开接口（无 cookie，仅试听片段）。
    若需要完整播放，由前端开关 unblock=true 触发，但本代理不实现 unblock
    （需要 api-enhanced 的 unblockmusic-utils，这里仅返回公开接口结果）。

    Returns:
        {id, url, type, freeTrialInfo, fee}
    """
    url = "https://music.163.com/api/song/enhance/player/url"
    params = {"ids": f"[{song_id}]", "br": "128000"}
    async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
        resp = await client.get(url, params=params, headers=_NCM_HEADERS)
        resp.raise_for_status()
        data = resp.json()
    item = (data.get("data") or [{}])[0]
    return {
        "id": item.get("id"),
        "url": item.get("url"),
        "type": item.get("type"),
        "freeTrialInfo": None,
        "fee": item.get("fee"),
        "unblock_supported": False,  # 标记给前端：本后端不支持 unblock
    }


async def get_song_detail(song_id: str) -> Optional[dict]:
    """获取歌曲详情（封面、歌手等）。"""
    url = "https://music.163.com/api/v1/song/detail"
    params = {"ids": f"[{song_id}]"}
    async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
        resp = await client.get(url, params=params, headers=_NCM_HEADERS)
        resp.raise_for_status()
        data = resp.json()
    songs = data.get("songs") or []
    if not songs:
        return None
    s = songs[0]
    return {
        "id": s.get("id"),
        "name": s.get("name"),
        "artists": "/".join(a.get("name", "") for a in s.get("artists", [])),
        "album": (s.get("album") or {}).get("name"),
        "cover": (s.get("album") or {}).get("picUrl"),
        "duration": s.get("duration"),
    }


async def get_song_detail_batch(song_ids: list[str]) -> list[Optional[dict]]:
    """批量获取歌曲详情。公开接口不支持批量，逐个调用。"""
    # 网易云公开接口 ids 参数实际是 JSON 数组字符串，可以传多个
    # 但为稳定性考虑，单个查询更可靠
    url = "https://music.163.com/api/v1/song/detail"
    ids_json = "[" + ",".join(str(i) for i in song_ids) + "]"
    params = {"ids": ids_json}
    async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
        resp = await client.get(url, params=params, headers=_NCM_HEADERS)
        resp.raise_for_status()
        data = resp.json()
    songs = data.get("songs") or []
    by_id = {}
    for s in songs:
        by_id[str(s.get("id"))] = {
            "id": s.get("id"),
            "name": s.get("name"),
            "artists": "/".join(a.get("name", "") for a in s.get("artists", [])),
            "album": (s.get("album") or {}).get("name"),
            "cover": (s.get("album") or {}).get("picUrl"),
            "duration": s.get("duration"),
        }
    return [by_id.get(str(sid)) for sid in song_ids]