File size: 4,206 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 | """网易云 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]
|