Archived
snowy-yang
d2e6856899
为什么：歌单解析的唯一上游通道，94 行单函数把「歌单头解析 / trackIds 分批拉明细 / 失败批次记账 / incomplete 标记 / 错误映射」五件事揉在 一个 try 里；曲目字段映射与搜索歌曲路逐字段重复。 拆出（模块级，与 search_music 拆解同风格）： - _song_base_fields：歌曲条目公共字段（搜索/歌单共用同一容错口径）， _map_song_item 改为其加 type 键的包装 - _playlist_track_ids：trackIds → 字符串 ID 列表 - _collect_playlist_tracks：批次循环整体（失败批次计数、脏 ID 跳批、 明细缺 id 跳过、incomplete 判定输入），嵌套 3 是批次语义固有， 集中在单一职责函数内 保留语义：脏 ID（'①'.isdigit() 为真但 int() 抛 ValueError）跳过整批 且不计 failed_batches；批次阶段网络错误仍映射「获取歌单详情请求失败」。 先写测试后动代码：新增 tests/test_playlist_detail.py 11 例，在未改动 实现上跑绿（修正 1 例测试自身的替身消费顺序），覆盖此前零测试路径—— playlist:null 防枚举、业务 code 错误、非 dict 响应、批次失败记账、 脏 ID 跳批、100/批分批、错误映射。 门禁：ruff 全过 · pyright 0/0 · pytest 359 passed（348→359）。
937 lines
36 KiB
Python
937 lines
36 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""网易云 API 主类。
|
||
|
||
（自 music_api.py 拆出，行为与原实现一致。）
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
from collections.abc import Callable
|
||
from random import randrange
|
||
from typing import Any
|
||
|
||
from core import aiohttp_compat as requests # 异步 HTTP 垫片（requests 同形 API）
|
||
|
||
from .client import (
|
||
HTTPClient,
|
||
_as_dict,
|
||
_ensure_dict,
|
||
)
|
||
from .constants import (
|
||
APIConstants,
|
||
)
|
||
from .crypto import (
|
||
CryptoUtils,
|
||
WeapiCrypto,
|
||
)
|
||
from .errors import (
|
||
APIException,
|
||
)
|
||
|
||
|
||
def _song_base_fields(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""歌曲条目公共字段（搜索四路与歌单详情共用同一容错口径）。"""
|
||
return {
|
||
'id': item['id'],
|
||
'name': item.get('name') or '未知歌曲',
|
||
'artists': '/'.join(filter(None, (a.get('name') for a in item.get('ar') or []))) or '未知歌手',
|
||
'album': (item.get('al') or {}).get('name') or '未知专辑',
|
||
'picUrl': (item.get('al') or {}).get('picUrl') or ''
|
||
}
|
||
|
||
|
||
def _map_song_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""歌曲条目映射。"""
|
||
return {
|
||
'type': 'song',
|
||
**_song_base_fields(item)
|
||
}
|
||
|
||
|
||
def _playlist_track_ids(playlist: dict[str, Any]) -> list[str]:
|
||
"""trackIds → 字符串 ID 列表（缺 id 的条目跳过）。"""
|
||
return [str(t['id']) for t in (playlist.get('trackIds') or []) if t.get('id')]
|
||
|
||
|
||
async def _collect_playlist_tracks(track_ids: list[str], headers: dict[str, str],
|
||
cookies: dict[str, str]) -> tuple[list[dict[str, Any]], int]:
|
||
"""分批拉取歌单全部曲目明细，返回 (tracks, failed_batches)。
|
||
|
||
明细批次的业务 code 也要校验，失败批次记入 failed_batches，不再静默缺歌。
|
||
脏 ID 跳过整批（'①'.isdigit() 为真但 int() 抛 ValueError——原实现此处
|
||
未捕获会整歌单 500，现为跳批）；明细缺 id 的歌跳过不拖整批。
|
||
"""
|
||
tracks: list[dict[str, Any]] = []
|
||
failed_batches = 0
|
||
for i in range(0, len(track_ids), 100):
|
||
batch_ids = track_ids[i:i + 100]
|
||
try:
|
||
song_data = {'c': json.dumps([{'id': int(sid), 'v': 0} for sid in batch_ids
|
||
if str(sid).isdigit()])}
|
||
except (TypeError, ValueError):
|
||
continue
|
||
|
||
song_resp = await requests.post(APIConstants.SONG_DETAIL_V3, data=song_data,
|
||
headers=headers, cookies=cookies, timeout=30)
|
||
song_resp.raise_for_status()
|
||
|
||
song_result = song_resp.json()
|
||
if not isinstance(song_result, dict):
|
||
# 上游 null/数组体按失败批次计，不让 AttributeError 失真
|
||
failed_batches += 1
|
||
continue
|
||
if song_result.get('code') != 200:
|
||
# 本批次业务失败：计为失败批次，交付方可见残缺状态
|
||
failed_batches += 1
|
||
continue
|
||
for song in (song_result.get('songs') or []):
|
||
if not song.get('id'):
|
||
continue
|
||
tracks.append(_song_base_fields(song))
|
||
return tracks, failed_batches
|
||
|
||
|
||
def _map_album_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""专辑条目映射（artists 为空**不**兜底成"未知歌手"——与歌曲路不同口径）。"""
|
||
return {
|
||
'type': 'album',
|
||
'id': item['id'],
|
||
'name': item.get('name') or '未知名称',
|
||
'artists': '/'.join(filter(None, (a.get('name') for a in item.get('artists') or []))),
|
||
'alias': '',
|
||
'album': '',
|
||
'picUrl': item.get('picUrl')
|
||
}
|
||
|
||
|
||
def _map_artist_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""歌手条目映射（alias 缺失/为空 → ''）。"""
|
||
return {
|
||
'type': 'artist',
|
||
'id': item['id'],
|
||
'name': item.get('name') or '未知名称',
|
||
'artists': '',
|
||
'alias': '/'.join(item.get('alias', [])) if item.get('alias') else '',
|
||
'album': '',
|
||
'picUrl': item.get('picUrl')
|
||
}
|
||
|
||
|
||
def _map_playlist_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""歌单条目映射。"""
|
||
return {
|
||
'type': 'playlist',
|
||
'id': item['id'],
|
||
'name': item.get('name') or '未知名称',
|
||
'artists': '',
|
||
'creator': (item.get('creator') or {}).get('nickname', ''),
|
||
'trackCount': item.get('trackCount', 0),
|
||
'album': '',
|
||
'picUrl': item.get('coverImgUrl')
|
||
}
|
||
|
||
|
||
# search_type → (上游结果键, 条目映射)；未知类型不在表内 → 空结果
|
||
_SEARCH_RESULT_KEYS: dict[int, str] = {
|
||
1: 'songs', 10: 'albums', 100: 'artists', 1000: 'playlists',
|
||
}
|
||
_SEARCH_ITEM_MAPPERS: dict[int, Callable[[dict[str, Any]], dict[str, Any]]] = {
|
||
1: _map_song_item, 10: _map_album_item, 100: _map_artist_item, 1000: _map_playlist_item,
|
||
}
|
||
|
||
|
||
def _collect_search_items(result_data: dict[str, Any], search_type: int) -> list[dict[str, Any]]:
|
||
"""按类型取上游结果键并逐条映射；残缺条目（无 id，下架/屏蔽）跳过，不拖死整页。"""
|
||
key = _SEARCH_RESULT_KEYS.get(search_type)
|
||
mapper = _SEARCH_ITEM_MAPPERS.get(search_type)
|
||
if key is None or mapper is None:
|
||
return []
|
||
items: list[dict[str, Any]] = []
|
||
for item in result_data.get(key, []):
|
||
if not item.get('id'):
|
||
continue
|
||
items.append(mapper(item))
|
||
return items
|
||
|
||
|
||
class NeteaseAPI:
|
||
"""网易云音乐API主类"""
|
||
|
||
def __init__(self):
|
||
self.http_client = HTTPClient()
|
||
self.crypto_utils = CryptoUtils()
|
||
|
||
async def get_song_url(self, song_id: int, quality: str, cookies: dict[str, str]) -> dict[str, Any]:
|
||
"""获取歌曲播放URL
|
||
|
||
Args:
|
||
song_id: 歌曲ID
|
||
quality: 音质等级 (standard, exhigh, lossless, hires, sky, jyeffect, jymaster)
|
||
cookies: 用户cookies
|
||
|
||
Returns:
|
||
包含歌曲URL信息的字典
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
config = APIConstants.DEFAULT_CONFIG.copy()
|
||
config["requestId"] = str(randrange(20000000, 30000000))
|
||
|
||
payload = {
|
||
'ids': [song_id],
|
||
'level': quality,
|
||
'encodeType': 'mp4' if quality == 'dolby' else 'flac',
|
||
'header': json.dumps(config),
|
||
}
|
||
|
||
if quality == 'sky':
|
||
payload['immerseType'] = 'c51'
|
||
|
||
params = self.crypto_utils.encrypt_params(APIConstants.SONG_URL_V1, payload)
|
||
response_text = await self.http_client.post_request(APIConstants.SONG_URL_V1, params, cookies)
|
||
|
||
result = json.loads(response_text)
|
||
# 合法 JSON 但非对象（null/数组）时给明确错误，不让 AttributeError 失真
|
||
if not isinstance(result, dict):
|
||
raise APIException("获取歌曲URL失败: 响应格式异常")
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取歌曲URL失败: {result.get('message', '未知错误')}")
|
||
|
||
return result
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析响应数据失败: {e}") from e
|
||
|
||
async def get_song_detail(self, song_id: int) -> dict[str, Any]:
|
||
"""获取歌曲详细信息
|
||
|
||
Args:
|
||
song_id: 歌曲ID
|
||
|
||
Returns:
|
||
包含歌曲详细信息的字典
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
data = {'c': json.dumps([{"id": song_id, "v": 0}])}
|
||
# 补齐 UA/Referer——本方法是全站唯一裸 UA 请求，风控/460 单点偏高
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
response = await requests.post(APIConstants.SONG_DETAIL_V3, data=data,
|
||
headers=headers, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
# 合法 JSON 但非对象（null/数组）时给明确错误，不让 AttributeError 失真
|
||
if not isinstance(result, dict):
|
||
raise APIException("获取歌曲详情失败: 响应格式异常")
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取歌曲详情失败: {result.get('message', '未知错误')}")
|
||
|
||
return result
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取歌曲详情请求失败: {e}") from e
|
||
except json.JSONDecodeError as e:
|
||
raise APIException(f"解析歌曲详情响应失败: {e}") from e
|
||
|
||
async def get_lyric(self, song_id: int, cookies: dict[str, str]) -> dict[str, Any]:
|
||
"""获取歌词信息
|
||
|
||
Args:
|
||
song_id: 歌曲ID
|
||
cookies: 用户cookies
|
||
|
||
Returns:
|
||
包含歌词信息的字典
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
data = {
|
||
'id': song_id,
|
||
'cp': 'false',
|
||
'tv': '0',
|
||
'lv': '0',
|
||
'rv': '0',
|
||
'kv': '0',
|
||
'yv': '1',
|
||
'ytv': '0',
|
||
'yrv': '0'
|
||
}
|
||
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.post(APIConstants.LYRIC_API, data=data,
|
||
headers=headers, cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'get_lyric 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取歌词失败: {result.get('message', '未知错误')}")
|
||
|
||
return result
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取歌词请求失败: {e}") from e
|
||
except json.JSONDecodeError as e:
|
||
raise APIException(f"解析歌词响应失败: {e}") from e
|
||
|
||
async def search_music(self, keywords: str, cookies: dict[str, str], limit: int = 10,
|
||
search_type: int = 1, offset: int = 0) -> list[dict[str, Any]]:
|
||
"""搜索音乐
|
||
|
||
Args:
|
||
keywords: 搜索关键词
|
||
cookies: 用户cookies
|
||
limit: 返回数量限制
|
||
search_type: 搜索类型 (1-歌曲, 10-专辑, 100-歌手, 1000-歌单)
|
||
offset: 分页偏移（透传上游，翻页生效）
|
||
|
||
Returns:
|
||
搜索结果列表
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
data = {'s': keywords, 'type': search_type, 'limit': limit,
|
||
'offset': max(0, int(offset or 0))}
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.post(APIConstants.SEARCH_API, data=data,
|
||
headers=headers, cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
# 非 dict 响应（null/数组）给明确错误
|
||
if not isinstance(result, dict):
|
||
raise APIException("搜索失败: 响应格式异常")
|
||
if result.get('code') != 200:
|
||
raise APIException(f"搜索失败: {result.get('message', '未知错误')}")
|
||
|
||
result_data = result.get('result') or {}
|
||
return _collect_search_items(result_data, search_type)
|
||
except requests.RequestException as e:
|
||
raise APIException(f"搜索请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析搜索响应失败: {e}") from e
|
||
|
||
async def get_playlist_detail(self, playlist_id: int, cookies: dict[str, str]) -> dict[str, Any]:
|
||
"""获取歌单详情
|
||
|
||
Args:
|
||
playlist_id: 歌单ID
|
||
cookies: 用户cookies
|
||
|
||
Returns:
|
||
歌单详情信息（曲目数与 trackIds 不一致时附 incomplete/missing/failed_batches）
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
data = {'id': playlist_id}
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.post(APIConstants.PLAYLIST_DETAIL_API, data=data,
|
||
headers=headers, cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
# 非 dict 响应（null/数组）给明确错误
|
||
if not isinstance(result, dict):
|
||
raise APIException("获取歌单详情失败: 响应格式异常")
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取歌单详情失败: {result.get('message', '未知错误')}")
|
||
|
||
# 上游对不存在/私密/无权限歌单返回 code:200 + playlist:null ——
|
||
# 显式转为 APIException（四条路由统一转"歌单不存在"语义，不再 500）
|
||
playlist = result.get('playlist') or {}
|
||
if not playlist:
|
||
raise APIException("歌单不存在或无权访问")
|
||
info = {
|
||
'id': playlist.get('id'),
|
||
'name': playlist.get('name'),
|
||
'coverImgUrl': playlist.get('coverImgUrl'),
|
||
'creator': (playlist.get('creator') or {}).get('nickname', ''),
|
||
'trackCount': playlist.get('trackCount'),
|
||
'description': playlist.get('description', ''),
|
||
'tracks': []
|
||
}
|
||
|
||
track_ids = _playlist_track_ids(playlist)
|
||
tracks, failed_batches = await _collect_playlist_tracks(track_ids, headers, cookies)
|
||
info['tracks'] = tracks
|
||
# trackIds 数与实际拿到的不一致时显式标记"不完整"，不冒充完整结果
|
||
if len(tracks) != len(track_ids):
|
||
info['incomplete'] = True
|
||
info['missing'] = len(track_ids) - len(tracks)
|
||
info['failed_batches'] = failed_batches
|
||
|
||
return info
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取歌单详情请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析歌单详情响应失败: {e}") from e
|
||
|
||
async def get_album_detail(self, album_id: int, cookies: dict[str, str]) -> dict[str, Any]:
|
||
"""获取专辑详情
|
||
|
||
Args:
|
||
album_id: 专辑ID
|
||
cookies: 用户cookies
|
||
|
||
Returns:
|
||
专辑详情信息
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
url = f'{APIConstants.ALBUM_DETAIL_API}{album_id}'
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.get(url, headers=headers, cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'get_album_detail 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取专辑详情失败: {result.get('message', '未知错误')}")
|
||
|
||
album = result.get('album') or {}
|
||
info = {
|
||
'id': album.get('id'),
|
||
'name': album.get('name'),
|
||
'coverImgUrl': self.get_pic_url(album.get('pic')),
|
||
'artist': (album.get('artist') or {}).get('name', ''),
|
||
'publishTime': album.get('publishTime'),
|
||
'description': album.get('description', ''),
|
||
'songs': []
|
||
}
|
||
|
||
for song in result.get('songs', []):
|
||
if not song.get('id'): # 残缺条目跳过
|
||
continue
|
||
info['songs'].append({
|
||
'id': song['id'],
|
||
'name': song.get('name') or '未知歌曲',
|
||
'artists': '/'.join(filter(None, (a.get('name') for a in song.get('ar') or []))) or '未知歌手',
|
||
'album': (song.get('al') or {}).get('name') or '未知专辑',
|
||
'picUrl': self.get_pic_url((song.get('al') or {}).get('pic'))
|
||
})
|
||
|
||
return info
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取专辑详情请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析专辑详情响应失败: {e}") from e
|
||
|
||
async def get_toplist(self) -> list[dict[str, Any]]:
|
||
"""获取网易云官方榜单列表
|
||
|
||
Returns:
|
||
榜单列表信息
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.get(APIConstants.TOPLIST_API, headers=headers, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'get_toplist 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取榜单失败: {result.get('message', '未知错误')}")
|
||
|
||
toplists = []
|
||
for item in (result.get('list') or []):
|
||
toplists.append({
|
||
'id': item.get('id'),
|
||
'name': item.get('name'),
|
||
'coverImgUrl': item.get('coverImgUrl'),
|
||
'description': item.get('description', ''),
|
||
'updateFrequency': item.get('updateFrequency', ''),
|
||
'playCount': item.get('playCount', 0),
|
||
'trackCount': item.get('trackCount', 0),
|
||
'topTitle': item.get('ToplistType', '')
|
||
})
|
||
return toplists
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取榜单请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析榜单响应失败: {e}") from e
|
||
|
||
async def get_artist_hot_songs(self, artist_id: int) -> dict[str, Any]:
|
||
"""获取歌手信息和热门歌曲
|
||
|
||
Args:
|
||
artist_id: 歌手ID
|
||
|
||
Returns:
|
||
歌手信息和热门歌曲列表
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
url = f'{APIConstants.ARTIST_API}{artist_id}'
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.get(url, headers=headers, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'get_artist_hot_songs 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取歌手信息失败: {result.get('message', '未知错误')}")
|
||
|
||
artist = result.get('artist') or {}
|
||
songs = []
|
||
for song in (result.get('hotSongs') or []):
|
||
songs.append({
|
||
'id': song.get('id'),
|
||
'name': song.get('name'),
|
||
'artists': '/'.join(a.get('name', '') for a in (song.get('ar') or [])),
|
||
'album': ((song.get('al') or {}).get('name', '')),
|
||
'picUrl': (song.get('al') or {}).get('picUrl', '')
|
||
})
|
||
|
||
return {
|
||
'artist': {
|
||
'id': artist.get('id'),
|
||
'name': artist.get('name'),
|
||
'picUrl': artist.get('picUrl'),
|
||
'alias': '/'.join(artist.get('alias', [])) if artist.get('alias') else '',
|
||
'briefDesc': artist.get('briefDesc', ''),
|
||
'musicSize': artist.get('musicSize', len(songs))
|
||
},
|
||
'songs': songs
|
||
}
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取歌手信息请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析歌手信息响应失败: {e}") from e
|
||
|
||
async def get_song_comments(self, song_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]:
|
||
"""获取歌曲评论
|
||
|
||
Args:
|
||
song_id: 歌曲ID
|
||
limit: 每页数量
|
||
offset: 偏移量
|
||
|
||
Returns:
|
||
评论数据（热门评论 + 全部评论）
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
"""
|
||
try:
|
||
url = f'{APIConstants.COMMENT_API}{song_id}?limit={limit}&offset={offset}'
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.get(url, headers=headers, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'get_song_comments 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取评论失败: {result.get('message', '未知错误')}")
|
||
|
||
def _comment_dict(c):
|
||
c = c or {}
|
||
user = c.get('user') or {}
|
||
be_replied = c.get('beReplied') or []
|
||
replied = (be_replied[0] or {}) if be_replied else {}
|
||
replied_user = replied.get('user') or {}
|
||
vip_rights = _as_dict(user.get('vipRights'))
|
||
associator = _as_dict(vip_rights.get('associator'))
|
||
location_info = _as_dict(user.get('locationInfo'))
|
||
return {
|
||
'id': c.get('commentId'),
|
||
'content': c.get('content', ''),
|
||
'time': c.get('time', 0),
|
||
'likedCount': c.get('likedCount', 0),
|
||
'user': {
|
||
'nickname': user.get('nickname', ''),
|
||
'avatarUrl': user.get('avatarUrl', ''),
|
||
'vipType': user.get('vipType', 0),
|
||
'vipLevel': associator.get('level', 0),
|
||
'location': location_info.get('location', '')
|
||
},
|
||
'beReplied': [{
|
||
'nickname': replied_user.get('nickname', '') or '',
|
||
'content': replied.get('content') or ''
|
||
}]
|
||
}
|
||
|
||
return {
|
||
'total': result.get('total', 0),
|
||
'hasMore': result.get('hasMore', False),
|
||
'hotComments': [_comment_dict(c) for c in (result.get('hotComments') or [])],
|
||
'comments': [_comment_dict(c) for c in (result.get('comments') or [])],
|
||
}
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取评论请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析评论响应失败: {e}") from e
|
||
|
||
async def get_account_info(self, cookies: dict[str, str]) -> dict[str, Any] | None:
|
||
"""获取当前账号信息（用于个人登录状态展示）
|
||
|
||
Args:
|
||
cookies: 用户cookies
|
||
|
||
Returns:
|
||
账号信息字典，获取失败返回 None
|
||
"""
|
||
try:
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
response = await requests.get(APIConstants.USER_ACCOUNT_API, headers=headers,
|
||
cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'get_account_info 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
return None
|
||
|
||
profile = result.get('profile') or {}
|
||
# 无 userId 说明 Cookie 未关联有效登录态（可能已过期）
|
||
if not profile.get('userId'):
|
||
return None
|
||
|
||
return {
|
||
'userId': profile.get('userId'),
|
||
'nickname': profile.get('nickname', ''),
|
||
'avatarUrl': profile.get('avatarUrl', ''),
|
||
'signature': profile.get('signature', ''),
|
||
'vipType': profile.get('vipType', 0),
|
||
'vipLevel': profile.get('vipLevel', 0)
|
||
}
|
||
except Exception:
|
||
return None
|
||
|
||
async def get_user_playlists(self, user_id: int, cookies: dict[str, str]) -> list[dict[str, Any]]:
|
||
"""获取用户收藏/创建的歌单列表
|
||
|
||
Args:
|
||
user_id: 用户ID
|
||
cookies: 用户cookies
|
||
|
||
Returns:
|
||
歌单列表（含「我喜欢的音乐」等个人歌单）
|
||
|
||
Raises:
|
||
APIException: API调用失败时抛出
|
||
|
||
超过 100 个歌单时按上游 more 标志续页（上限 20 页 = 2000 个），
|
||
不再只取第一页就静默截断。
|
||
"""
|
||
try:
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
|
||
playlists = []
|
||
offset = 0
|
||
page_size = 100
|
||
for _ in range(20): # 有界翻页，防上游 more 异常导致无限循环
|
||
params = {'uid': user_id, 'limit': page_size, 'offset': offset}
|
||
response = await requests.get(APIConstants.USER_PLAYLIST_API, params=params,
|
||
headers=headers, cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'get_user_playlists 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"获取用户歌单失败: {result.get('message', '未知错误')}")
|
||
|
||
page_items = result.get('playlist') or []
|
||
seen_ids = {p.get('id') for p in playlists}
|
||
for item in page_items:
|
||
if item.get('id') in seen_ids:
|
||
continue # 上游列表变动导致重复时跳过
|
||
creator = item.get('creator') or {}
|
||
playlists.append({
|
||
'id': item.get('id'),
|
||
'name': item.get('name', ''),
|
||
'coverImgUrl': item.get('coverImgUrl'),
|
||
'trackCount': item.get('trackCount', 0),
|
||
'playCount': item.get('playCount', 0),
|
||
'creator': creator.get('nickname', ''),
|
||
'is_own': creator.get('userId') == user_id,
|
||
'specialType': item.get('specialType', 0),
|
||
'subscribed': item.get('subscribed', False),
|
||
'description': item.get('description', '')
|
||
})
|
||
more = bool(result.get('more'))
|
||
if not more or len(page_items) < page_size:
|
||
break
|
||
offset += page_size
|
||
return playlists
|
||
except requests.RequestException as e:
|
||
raise APIException(f"获取用户歌单请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析用户歌单响应失败: {e}") from e
|
||
|
||
async def detect_vip_by_probe(self, cookies: dict[str, str],
|
||
probe_song_id: int = 347230) -> dict[str, Any]:
|
||
"""通过歌曲URL接口探测账号可用的最高音质，推断会员类型
|
||
|
||
直接询问网易云接口「这个账号能把歌解析到什么音质」，
|
||
比 account/get 更可靠（旧 Cookie 可能查不到 profile）。
|
||
|
||
Args:
|
||
cookies: 用户cookies
|
||
probe_song_id: 用于探测的歌曲ID（默认《海阔天空》，支持全音质）
|
||
|
||
Returns:
|
||
{'vip_type': int, 'highest_level': str, 'levels_ok': List[str]}
|
||
"""
|
||
levels_ok = await self.probe_levels(cookies, probe_song_id)
|
||
highest = levels_ok[0] if levels_ok else 'standard'
|
||
|
||
if highest in ('jymaster', 'jyeffect', 'sky'):
|
||
vip_type = 110 # 黑胶SVIP（可解析沉浸/母带级）
|
||
elif highest in ('hires', 'lossless'):
|
||
vip_type = 11 # 黑胶VIP
|
||
else:
|
||
vip_type = 0 # 普通用户
|
||
return {'vip_type': vip_type, 'highest_level': highest, 'levels_ok': levels_ok}
|
||
|
||
async def probe_levels(self, cookies: dict[str, str],
|
||
probe_song_id: int = 347230) -> list[str]:
|
||
"""实测账号对每个音质级别能否真正解析
|
||
|
||
只有接口返回的 level 与请求一致且带 URL，才算该音质可解析。
|
||
返回按音质从高到低排序的可解析列表。
|
||
|
||
Args:
|
||
cookies: 用户cookies
|
||
probe_song_id: 探测歌曲ID
|
||
|
||
Returns:
|
||
可解析音质列表（从高到低）
|
||
"""
|
||
ladder = ['jymaster', 'jyeffect', 'sky', 'hires', 'lossless', 'exhigh', 'standard', 'dolby']
|
||
levels_ok = []
|
||
# 探测总预算 60s——上游黑洞时 8 级串行最坏 ~16 分钟线程滞留，
|
||
# 超预算即返回已测得的结果（调用方 5 分钟缓存内复用）
|
||
_deadline = time.monotonic() + 60
|
||
for level in ladder:
|
||
if time.monotonic() > _deadline:
|
||
break
|
||
try:
|
||
result = await self.get_song_url(probe_song_id, level, cookies)
|
||
if result and result.get('data'):
|
||
d = result['data'][0] or {}
|
||
# 请求与返回一致且带直链，才说明该音质真正可用
|
||
if d.get('url') and d.get('level') == level:
|
||
levels_ok.append(level)
|
||
except Exception:
|
||
continue
|
||
return levels_ok
|
||
|
||
async def subscribe_playlist(self, playlist_id, cookies: dict[str, str],
|
||
action: int = 1) -> dict[str, Any]:
|
||
"""收藏/取消收藏歌单（官方接口）
|
||
|
||
Args:
|
||
playlist_id: 歌单ID
|
||
cookies: 个人账号cookies
|
||
action: 1=收藏, 2=取消收藏
|
||
|
||
Returns:
|
||
接口响应
|
||
|
||
Raises:
|
||
APIException: 操作失败时抛出
|
||
"""
|
||
try:
|
||
data = {'id': playlist_id, 't': action}
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
response = await requests.post(APIConstants.PLAYLIST_SUBSCRIBE_API, data=data,
|
||
headers=headers, cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
_ensure_dict(result, 'subscribe_playlist 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"收藏歌单失败: {result.get('message', '未知错误')}")
|
||
return result
|
||
except requests.RequestException as e:
|
||
raise APIException(f"收藏歌单请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析收藏歌单响应失败: {e}") from e
|
||
|
||
async def subscribe_album(self, album_id, cookies: dict[str, str],
|
||
action: int = 1) -> dict[str, Any]:
|
||
"""收藏/取消收藏专辑（官方接口）
|
||
|
||
Args:
|
||
album_id: 专辑ID
|
||
cookies: 个人账号cookies
|
||
action: 1=收藏, 2=取消收藏
|
||
|
||
Returns:
|
||
接口响应
|
||
|
||
Raises:
|
||
APIException: 操作失败时抛出
|
||
"""
|
||
try:
|
||
data = {'id': album_id, 't': action}
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER
|
||
}
|
||
response = await requests.post(APIConstants.ALBUM_SUBSCRIBE_API, data=data,
|
||
headers=headers, cookies=cookies, timeout=30)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
_ensure_dict(result, 'subscribe_album 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"收藏专辑失败: {result.get('message', '未知错误')}")
|
||
return result
|
||
except requests.RequestException as e:
|
||
raise APIException(f"收藏专辑请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析收藏专辑响应失败: {e}") from e
|
||
|
||
async def post_comment(self, thread_id: str, content: str, action: int = 1,
|
||
comment_id: int | None = None,
|
||
cookies: dict[str, str] | None = None) -> dict[str, Any]:
|
||
"""发送/回复歌曲评论（官方 weapi 接口）
|
||
|
||
Args:
|
||
thread_id: 资源线程ID，如 R_SO_4_{songId}
|
||
content: 评论内容
|
||
action: 1=发送评论, 2=回复评论
|
||
comment_id: 回复目标评论ID（action=2 时必填）
|
||
cookies: 个人账号cookies
|
||
|
||
Returns:
|
||
接口响应
|
||
|
||
Raises:
|
||
APIException: 操作失败时抛出
|
||
"""
|
||
try:
|
||
payload: dict[str, Any] = {
|
||
'threadId': thread_id,
|
||
'content': content,
|
||
't': action,
|
||
'type': 6,
|
||
}
|
||
if comment_id:
|
||
payload['commentId'] = comment_id
|
||
# 官方 /weapi/comment 要求携带 csrf_token（取自 Cookie
|
||
# 的 __csrf）——原实现缺失，与加密偏差叠加导致评论疑似从未成功
|
||
csrf = (cookies or {}).get('__csrf') or ''
|
||
if csrf:
|
||
payload['csrf_token'] = csrf
|
||
|
||
body = WeapiCrypto.encrypt(payload)
|
||
headers = {
|
||
'User-Agent': APIConstants.USER_AGENT,
|
||
'Referer': APIConstants.REFERER,
|
||
}
|
||
response = await requests.post(APIConstants.COMMENT_POST_API, data=body,
|
||
headers=headers, cookies=cookies or {}, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
text = response.text.strip()
|
||
if not text:
|
||
# weapi 被网络层吞掉（数据中心 IP 风控），无法确认是否成功
|
||
raise APIException("网易云未返回操作确认（可能被网络风控拦截），请稍后查看评论区确认")
|
||
|
||
result = response.json()
|
||
_ensure_dict(result, 'post_comment 接口响应异常（非标准 JSON 对象）')
|
||
if result.get('code') != 200:
|
||
raise APIException(f"评论失败: {result.get('message', '未知错误')}")
|
||
return result
|
||
except requests.RequestException as e:
|
||
raise APIException(f"评论请求失败: {e}") from e
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
raise APIException(f"解析评论响应失败: {e}") from e
|
||
|
||
def netease_encrypt_id(self, id_str: str) -> str:
|
||
"""网易云加密图片ID算法
|
||
|
||
Args:
|
||
id_str: 图片ID字符串
|
||
|
||
Returns:
|
||
加密后的字符串
|
||
"""
|
||
import base64
|
||
import hashlib
|
||
|
||
magic = list('3go8&$8*3*3h0k(2)2')
|
||
song_id = list(id_str)
|
||
|
||
for i in range(len(song_id)):
|
||
song_id[i] = chr(ord(song_id[i]) ^ ord(magic[i % len(magic)]))
|
||
|
||
m = ''.join(song_id)
|
||
md5_bytes = hashlib.md5(m.encode('utf-8')).digest()
|
||
result = base64.b64encode(md5_bytes).decode('utf-8')
|
||
result = result.replace('/', '_').replace('+', '-')
|
||
|
||
return result
|
||
|
||
def get_pic_url(self, pic_id: int | None, size: int = 300) -> str:
|
||
"""获取网易云加密歌曲/专辑封面直链
|
||
|
||
Args:
|
||
pic_id: 封面ID
|
||
size: 图片尺寸
|
||
|
||
Returns:
|
||
图片URL
|
||
"""
|
||
if pic_id is None:
|
||
return ''
|
||
|
||
enc_id = self.netease_encrypt_id(str(pic_id))
|
||
return f'https://p3.music.126.net/{enc_id}/{pic_id}.jpg?param={size}y{size}'
|