hhhge/NeteaseUrl
Archived
1
0 Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

refactor: 拆分 4 个巨石模块为包，清理过时注释并删除作废文档

Browse Source 
为什么：单文件 1100~1900 行的巨石模块让「改一处要通读全文」；同时注释里残留
大量已删台账的编号，以及「Flask/FastAPI 双栈迁移期」这类已失实的架构描述，
会误导后来者按不复存在的架构去改代码。

拆分（对外导入路径不变，行为零变更）：
- music_downloader.py  -> music_downloader/（models/downloader/cover/_http/_log/_concurrency）
- auth_system.py       -> auth_system/（db/security/users/credits/ledger/mail/cards/backup...）
- music_api.py         -> music_api/（api/client/crypto/qr/cookies/compat/constants/errors）
- services/runtime.py  -> services/（api_service/_shared/_taskcore/downloads/batch/podcast/
  streaming/tasks）；services/runtime.py 保留为兼容门面

兼容性做法：各包用 __init__.py 以 __all__ 原样再导出全部符号，既有调用点
（auth_system.xxx / services.xxx / from music_api import Y）零改动；路径常量改为经包属性
动态读取（auth_system.DB_PATH），否则测试的 monkeypatch 会静默失效并写到生产库。

修复 9 处代码损伤（上一轮注释清理把 `()` 当「空括号残留」删掉所致；用「与拆分前逐字
等价」校验逐一定位，静态检查只会报出其中 4 处）：
- core/config.py  ).strip() or '*'
- core/state.py   threading.Lock() x2
- qr_login.py     datetime.now() / .strip()
- music_downloader/downloader.py  .lower()
- services/podcast.py  set() / time.time() x2

清理与删除：
- 删除 部署文档.md / 功能介绍.md / README_私有包警告.txt
- 清理指向已删台账与文档的过时编号注释；把「迁移期 / 两栈 / 双框架 / 已迁移」等
  失实描述改写为现状描述（保留 `<-> Flask /path` 映射表等仍有价值的契约来源说明）
- _mkpkg.py：公开包剔除清单去掉已不存在的文档名；私有包警告改为自包含
- pyproject.toml：补 pythonpath = ["."]，使裸跑 pytest 不再 import core 失败

验证：ruff 全通过；pyright 0 errors / 0 warnings（102 文件）；pytest 144 passed；
真实 uvicorn 冒烟 6 条路由符合预期且生产 data/ 指纹未变；忽略注释的逐字等价性校验通过。
This commit is contained in:
snowy-yang committed 2026-09-26 12:37:48 +08:00
1 parent 838d635bc5
commit 6288a98ecc
89 files changed
+5806 -5112
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

music_api/__init__.py
+99
View File
@@ -0,0 +1,99 @@
# -*- coding: utf-8 -*-
"""网易云音乐API模块
提供网易云音乐相关API接口的封装，包括：
- 音乐URL获取
- 歌曲详情获取
- 歌词获取
- 搜索功能
- 歌单和专辑详情
- 二维码登录
"""
# 兼容门面：实现已按职责拆分到同包各子模块，此处原样再导出。
# 用 `__all__` 声明再导出（ruff F401 与 pyright 均据此认定「有意再导出」），
# 使既有调用点（music_api.xxx / from music_api import Y）零改动。
from .api import (
NeteaseAPI,
)
from .client import (
_MAX_UPSTREAM_JSON_BYTES,
HTTPClient,
_as_dict,
_ensure_dict,
)
from .compat import (
account_info,
album_detail,
album_subscribe,
artist_songs,
detect_vip_by_probe,
get_pic_url,
lyric_v1,
name_v1,
playlist_detail,
playlist_subscribe,
post_comment,
probe_levels,
qr_login,
search_music,
song_comments,
toplist_list,
url_v1,
user_playlists,
)
from .constants import (
APIConstants,
QualityLevel,
)
from .cookies import (
DEFAULT_COOKIE_FILE,
load_cookies,
parse_cookie_string,
)
from .crypto import (
CryptoUtils,
WeapiCrypto,
)
from .errors import (
APIException,
)
from .qr import (
QRLoginManager,
)
__all__ = [
'DEFAULT_COOKIE_FILE',
'_MAX_UPSTREAM_JSON_BYTES',
'APIConstants',
'APIException',
'CryptoUtils',
'HTTPClient',
'NeteaseAPI',
'QRLoginManager',
'QualityLevel',
'WeapiCrypto',
'_as_dict',
'_ensure_dict',
'account_info',
'album_detail',
'album_subscribe',
'artist_songs',
'detect_vip_by_probe',
'get_pic_url',
'load_cookies',
'lyric_v1',
'name_v1',
'parse_cookie_string',
'playlist_detail',
'playlist_subscribe',
'post_comment',
'probe_levels',
'qr_login',
'search_music',
'song_comments',
'toplist_list',
'url_v1',
'user_playlists',
]
music_api/__main__.py
+18
View File
@@ -0,0 +1,18 @@
# -*- coding: utf-8 -*-
"""命令行入口（python -m music_api）。"""
if __name__ == "__main__":
# 测试代码
print("网易云音乐API模块")
print("支持的功能:")
print("- 歌曲URL获取")
print("- 歌曲详情获取")
print("- 歌词获取")
print("- 音乐搜索")
print("- 歌单详情")
print("- 专辑详情")
print("- 二维码登录")
music_api/api.py
+903
View File
@@ -0,0 +1,903 @@
# -*- coding: utf-8 -*-
"""网易云 API 主类。
（自 music_api.py 拆出，行为与原实现一致。）
"""
import json
import time
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,
)
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 {}
items: list[dict[str, Any]] = []
if search_type == 1:
# 歌曲（id 改 .get——上游单条残缺（下架/屏蔽）不再拖死整页搜索）
for item in result_data.get('songs', []):
if not item.get('id'):
continue
items.append({
'type': 'song',
'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 ''
})
elif search_type == 10:
# 专辑（同上）
for item in result_data.get('albums', []):
if not item.get('id'):
continue
items.append({
'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')
})
elif search_type == 100:
# 歌手（同上）
for item in result_data.get('artists', []):
if not item.get('id'):
continue
items.append({
'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')
})
elif search_type == 1000:
# 歌单（同上）
for item in result_data.get('playlists', []):
if not item.get('id'):
continue
items.append({
'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')
})
return items
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:
歌单详情信息
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': []
}
# 获取所有trackIds并分批获取详细信息（明细批次的业务 code 也要校验，
# 失败批次记入 failed_batches，不再静默缺歌）
track_ids = [str(t['id']) for t in (playlist.get('trackIds') or []) if t.get('id')]
failed_batches = 0
for i in range(0, len(track_ids), 100):
batch_ids = track_ids[i:i+100]
# trackIds 混入非数字脏 ID 时跳过该条（原实现 int 抛 ValueError
# 不在捕获列表，整歌单 500）；明细缺 id 的歌跳过不拖整批
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
info['tracks'].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': (song.get('al') or {}).get('picUrl') or ''
})
# trackIds 数与实际拿到的不一致时显式标记"不完整"，不冒充完整结果
if len(info['tracks']) != len(track_ids):
info['incomplete'] = True
info['missing'] = len(track_ids) - len(info['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}'
music_api/client.py
+112
View File
@@ -0,0 +1,112 @@
# -*- coding: utf-8 -*-
"""HTTP 客户端与响应体收窄工具。
（自 music_api.py 拆出，行为与原实现一致。）
"""
import asyncio
from typing import Any
from core import aiohttp_compat as requests # 异步 HTTP 垫片（requests 同形 API）
from .constants import (
APIConstants,
)
from .errors import (
APIException,
)
def _ensure_dict(result: Any, msg: str) -> dict[str, Any]:
"""上游 200 但返回 null/数组等非 dict 体（风控/抖动）时统一前置校验"""
if isinstance(result, dict):
return result
raise APIException(msg)
def _as_dict(v: Any) -> dict[str, Any]:
"""上游嵌套字段类型不稳定（dict / None / 列表都真实出现过）：非 dict 一律回落空 dict。
收窄成显式函数，避免满屏 `x if isinstance(x, dict) else {}` 三元式——
那种写法读起来累，且把字段来自 `user.get(...)`（类型 Any|None）时的收窄责任
散落在各处，容易被后来的改动改漏。
"""
return v if isinstance(v, dict) else {}
class HTTPClient:
"""HTTP客户端类"""
@staticmethod
async def post_request(url: str, params: str, cookies: dict[str, str]) -> str:
"""发送POST请求并返回文本响应。
网络异常/5xx 做一次 1 秒退避重试（全站 API 调用此前零重试，
上游抖动即整首失败）；4xx（429/460 风控）不重试——立即重试会加剧风控。"""
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER,
}
request_cookies = APIConstants.DEFAULT_COOKIES.copy()
request_cookies.update(cookies)
last_err = None
for attempt in range(2):
try:
response = await requests.post(url, headers=headers, cookies=request_cookies,
data={"params": params}, timeout=30)
if response.status_code >= 500 and attempt == 0:
await asyncio.sleep(1.0)
continue
response.raise_for_status()
if len(response.content) > _MAX_UPSTREAM_JSON_BYTES:
raise APIException("上游响应体异常（超过大小上限）")
return response.text
except requests.HTTPError as e:
# 4xx（429/460 风控）不重试——原实现 HTTPError 落入
# 下方 RequestException 分支被重试一次，与 意图相反
raise APIException(f"HTTP请求失败: {e}") from e
except requests.RequestException as e:
last_err = e
if attempt == 0:
await asyncio.sleep(1.0)
continue
raise APIException(f"HTTP请求失败: {last_err}")
@staticmethod
async def post_request_full(url: str, params: str, cookies: dict[str, str]) -> requests.Response:
"""发送POST请求并返回完整响应对象（同 post_request 的有界重试）"""
headers = {
'User-Agent': APIConstants.USER_AGENT,
'Referer': APIConstants.REFERER,
}
request_cookies = APIConstants.DEFAULT_COOKIES.copy()
request_cookies.update(cookies)
last_err = None
for attempt in range(2):
try:
response = await requests.post(url, headers=headers, cookies=request_cookies,
data={"params": params}, timeout=30)
if response.status_code >= 500 and attempt == 0:
await asyncio.sleep(1.0)
continue
response.raise_for_status()
if len(response.content) > _MAX_UPSTREAM_JSON_BYTES:
raise APIException("上游响应体异常（超过大小上限）")
return response
except requests.HTTPError as e:
# 4xx 不重试（同上）
raise APIException(f"HTTP请求失败: {e}") from e
except requests.RequestException as e:
last_err = e
if attempt == 0:
await asyncio.sleep(1.0)
continue
raise APIException(f"HTTP请求失败: {last_err}")
# 上游 JSON 响应体上限——requests 已全量入内存，此处拦截的是
# 异常/被劫持上游的超大 body 进入后续 .json() 解析与传播（音频/封面已有字节预算）
_MAX_UPSTREAM_JSON_BYTES = 8 * 1024 * 1024
music_api/compat.py
+128
View File
@@ -0,0 +1,128 @@
# -*- coding: utf-8 -*-
"""同步兼容函数接口（内部经 run_sync 桥接异步实现）。
（自 music_api.py 拆出，行为与原实现一致。）
"""
from typing import Any
from core.async_bridge import run_sync # 模块级同步兼容接口用的同步/异步桥
from .api import (
NeteaseAPI,
)
from .qr import (
QRLoginManager,
)
# 向后兼容的函数接口
def url_v1(song_id: int, level: str, cookies: dict[str, str]) -> dict[str, Any]:
"""获取歌曲URL（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_song_url(song_id, level, cookies))
def name_v1(song_id: int) -> dict[str, Any]:
"""获取歌曲详情（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_song_detail(song_id))
def lyric_v1(song_id: int, cookies: dict[str, str]) -> dict[str, Any]:
"""获取歌词（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_lyric(song_id, cookies))
def search_music(keywords: str, cookies: dict[str, str], limit: int = 10,
search_type: int = 1, offset: int = 0) -> list[dict[str, Any]]:
"""搜索音乐（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.search_music(keywords, cookies, limit, search_type, offset=offset))
def toplist_list() -> list[dict[str, Any]]:
"""获取官方榜单列表（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_toplist())
def artist_songs(artist_id: int) -> dict[str, Any]:
"""获取歌手信息和热门歌曲（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_artist_hot_songs(artist_id))
def song_comments(song_id: int, limit: int = 20, offset: int = 0) -> dict[str, Any]:
"""获取歌曲评论（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_song_comments(song_id, limit, offset))
def account_info(cookies: dict[str, str]) -> dict[str, Any] | None:
"""获取当前账号信息（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_account_info(cookies))
def user_playlists(user_id: int, cookies: dict[str, str]) -> list[dict[str, Any]]:
"""获取用户收藏/创建的歌单列表（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_user_playlists(user_id, cookies))
def detect_vip_by_probe(cookies: dict[str, str], probe_song_id: int = 347230) -> dict[str, Any]:
"""通过歌曲URL接口探测账号会员类型（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.detect_vip_by_probe(cookies, probe_song_id))
def playlist_subscribe(playlist_id: int, cookies: dict[str, str], action: int = 1) -> dict[str, Any]:
"""收藏/取消收藏歌单（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.subscribe_playlist(playlist_id, cookies, action))
def album_subscribe(album_id: int, cookies: dict[str, str], action: int = 1) -> dict[str, Any]:
"""收藏/取消收藏专辑（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.subscribe_album(album_id, cookies, action))
def post_comment(thread_id: str, content: str, action: int = 1,
comment_id: int | None = None,
cookies: dict[str, str] | None = None) -> dict[str, Any]:
"""发送/回复歌曲评论（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.post_comment(thread_id, content, action, comment_id, cookies))
def probe_levels(cookies: dict[str, str], probe_song_id: int = 347230) -> list[str]:
"""实测账号可解析的音质列表（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.probe_levels(cookies, probe_song_id))
def playlist_detail(playlist_id: int, cookies: dict[str, str]) -> dict[str, Any]:
"""获取歌单详情（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_playlist_detail(playlist_id, cookies))
def album_detail(album_id: int, cookies: dict[str, str]) -> dict[str, Any]:
"""获取专辑详情（向后兼容）"""
api = NeteaseAPI()
return run_sync(api.get_album_detail(album_id, cookies))
def get_pic_url(pic_id: int | None, size: int = 300) -> str:
"""获取图片URL（向后兼容）"""
api = NeteaseAPI()
return api.get_pic_url(pic_id, size)
def qr_login() -> str | None:
"""二维码登录（向后兼容）"""
manager = QRLoginManager()
return run_sync(manager.qr_login())
music_api/constants.py
+60
View File
@@ -0,0 +1,60 @@
# -*- coding: utf-8 -*-
"""音质枚举与接口常量。
（自 music_api.py 拆出，行为与原实现一致。）
"""
from enum import Enum
class QualityLevel(Enum):
"""音质等级枚举"""
STANDARD = "standard" # 标准音质
EXHIGH = "exhigh" # 极高音质
LOSSLESS = "lossless" # 无损音质
HIRES = "hires" # Hi-Res音质
SKY = "sky" # 沉浸环绕声
JYEFFECT = "jyeffect" # 高清环绕声
JYMASTER = "jymaster" # 超清母带
DOLBY = "dolby" # 杜比全景声
# 常量定义
class APIConstants:
"""API相关常量"""
AES_KEY = b"e82ckenh8dichen8"
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36 Chrome/91.0.4472.164 NeteaseMusicDesktop/2.10.2.200154'
REFERER = 'https://music.163.com/'
# API URLs
SONG_URL_V1 = "https://interface3.music.163.com/eapi/song/enhance/player/url/v1"
SONG_DETAIL_V3 = "https://interface3.music.163.com/api/v3/song/detail"
LYRIC_API = "https://interface3.music.163.com/api/song/lyric"
SEARCH_API = 'https://music.163.com/api/cloudsearch/pc'
PLAYLIST_DETAIL_API = 'https://music.163.com/api/v6/playlist/detail'
ALBUM_DETAIL_API = 'https://music.163.com/api/v1/album/'
TOPLIST_API = 'https://music.163.com/api/toplist'
ARTIST_API = 'https://music.163.com/api/v1/artist/'
COMMENT_API = 'https://music.163.com/api/v1/resource/comments/R_SO_4_'
USER_ACCOUNT_API = 'https://music.163.com/api/nuser/account/get'
USER_PLAYLIST_API = 'https://music.163.com/api/user/playlist'
PLAYLIST_SUBSCRIBE_API = 'https://music.163.com/api/playlist/subscribe'
ALBUM_SUBSCRIBE_API = 'https://music.163.com/api/album/sub'
COMMENT_POST_API = 'https://music.163.com/weapi/comment'
QR_UNIKEY_API = 'https://interface3.music.163.com/eapi/login/qrcode/unikey'
QR_LOGIN_API = 'https://interface3.music.163.com/eapi/login/qrcode/client/login'
# 默认配置
DEFAULT_CONFIG = {
"os": "pc",
"appver": "",
"osver": "",
"deviceId": "pyncm!"
}
DEFAULT_COOKIES = {
"os": "pc",
"appver": "",
"osver": "",
"deviceId": "pyncm!"
}
music_api/cookies.py
+60
View File
@@ -0,0 +1,60 @@
# -*- coding: utf-8 -*-
"""Cookie 读取与解析。
（自 music_api.py 拆出，行为与原实现一致。）
"""
import os
# ============ Cookie 工具 ============
DEFAULT_COOKIE_FILE = "cookie.txt"
def load_cookies(path: str = DEFAULT_COOKIE_FILE) -> dict[str, str]:
"""从 cookie.txt 直接读取并解析为 dict
支持标准 ``NMTID=xxx; _ntes_nnid=xxx; ...`` 格式。
文件不存在或解析失败返回空 dict（不抛异常，调用方决定如何处理）。
utf-8-sig 剥记事本 BOM（BOM 会让首键变成 \\ufeffMUSIC_U 静默
丢 Cookie）；解析改逐行再按 ``;`` 切（原实现 ';|\\n' 二选一，混排时丢键值对）。
"""
if not os.path.isfile(path):
return {}
try:
with open(path, encoding="utf-8-sig") as f:
content = f.read()
except OSError:
return {}
return parse_cookie_string(content)
def parse_cookie_string(cookie_string: str) -> dict[str, str]:
"""把 ``k1=v1; k2=v2`` 格式字符串解析为 dict
支持 ``;`` 或 ``\\n`` 作为分隔符。
"""
cookies: dict[str, str] = {}
if not cookie_string:
return cookies
cookie_string = cookie_string.strip()
if not cookie_string:
return cookies
# 先逐行再按 ; 切——原实现 ";|\\n" 二选一，混排时行内 \\n
# 会混进 value、后半段键值对丢失
pairs = []
for line in cookie_string.splitlines():
for part in line.split(";"):
part = part.strip()
if part:
pairs.append(part)
for pair in pairs:
pair = pair.strip()
if not pair or "=" not in pair:
continue
key, value = pair.split("=", 1)
key = key.strip()
value = value.strip()
if key and value:
cookies[key] = value
return cookies
music_api/crypto.py
+105
View File
@@ -0,0 +1,105 @@
# -*- coding: utf-8 -*-
"""weapi 加解密（AES + RSA）。
（自 music_api.py 拆出，行为与原实现一致。）
"""
import json
import urllib.parse
from hashlib import md5
from typing import Any
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from .constants import (
APIConstants,
)
class CryptoUtils:
"""加密工具类"""
@staticmethod
def hex_digest(data: bytes) -> str:
"""将字节数据转换为十六进制字符串"""
return "".join([hex(d)[2:].zfill(2) for d in data])
@staticmethod
def hash_digest(text: str) -> bytes:
"""计算MD5哈希值"""
return md5(text.encode("utf-8")).digest()
@staticmethod
def hash_hex_digest(text: str) -> str:
"""计算MD5哈希值并转换为十六进制字符串"""
return CryptoUtils.hex_digest(CryptoUtils.hash_digest(text))
@staticmethod
def encrypt_params(url: str, payload: dict[str, Any]) -> str:
"""加密请求参数"""
url_path = urllib.parse.urlparse(url).path.replace("/eapi/", "/api/")
digest = CryptoUtils.hash_hex_digest(f"nobody{url_path}use{json.dumps(payload)}md5forencrypt")
params = f"{url_path}-36cd479b6b5-{json.dumps(payload)}-36cd479b6b5-{digest}"
# AES加密
padder = padding.PKCS7(algorithms.AES(APIConstants.AES_KEY).block_size).padder()
padded_data = padder.update(params.encode()) + padder.finalize()
cipher = Cipher(algorithms.AES(APIConstants.AES_KEY), modes.ECB())
encryptor = cipher.encryptor()
enc = encryptor.update(padded_data) + encryptor.finalize()
return CryptoUtils.hex_digest(enc)
class WeapiCrypto:
"""网易云 weapi 接口加密（AES-CBC 双重加密 + RSA 公钥加密）
用于发送/回复评论等前端写操作接口（/weapi/comment 等）。
"""
AES_KEY = b"0CoJUm6Qyw8W8jud"
AES_IV = b"0102030405060708"
RSA_MODULUS = "00e0b509f6259df8642dbc35662901477df22677ec152b5ff68ace615bb7b725152b3ab17a876aea8a5aa76d2e417629ec4ee341f56135fccf695280104e0312ecbda92557c93870114af6c9d05c4f7f0c3685b7a46bee255932575cce10b424d813cfe4875d3e82047b97ddef52741d546b8e289dc6935b3ece0462db0a22b8e7"
RSA_EXPONENT = "010001"
# 官方随机 secKey 字符集（官方 JS 与主流逆向实现一致）
SEC_KEY_CHARSET = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
@staticmethod
def _aes_encrypt(text: bytes, key: bytes) -> bytes:
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.padding import PKCS7
padder = PKCS7(128).padder()
padded = padder.update(text) + padder.finalize()
cipher = Cipher(algorithms.AES(key), modes.CBC(WeapiCrypto.AES_IV))
encryptor = cipher.encryptor()
return encryptor.update(padded) + encryptor.finalize()
@staticmethod
def encrypt(payload: dict[str, Any]) -> dict[str, str]:
"""将请求体加密为 weapi 的 params/encSecKey。
按官方协议重写——secKey 取官方字符集随机 16 位，
encSecKey = secKey 反转后做 textbook RSA（无填充）的 256 位 hex。
原实现用 os.urandom + PKCS1v15 且未反转，服务端按 textbook RSA 解出的
AES key 与 params 第二层不一致，评论等 weapi 写接口疑似从未成功过。
官方协议第二层加密的输入是第一层密文的
**base64 字符串**（两次 AES 各自输出 base64）——93 条版本漏了中间那次
base64，服务端解 params 第一层得到二进制、b64 解码必失败。"""
import base64
import secrets as _secrets
text = json.dumps(payload).encode("utf-8")
secret_key = ''.join(_secrets.choice(WeapiCrypto.SEC_KEY_CHARSET) for _ in range(16))
first = base64.b64encode(
WeapiCrypto._aes_encrypt(text, WeapiCrypto.AES_KEY)
).decode("utf-8")
params = base64.b64encode(
WeapiCrypto._aes_encrypt(first.encode("utf-8"), secret_key.encode('utf-8'))
).decode("utf-8")
n = int(WeapiCrypto.RSA_MODULUS, 16)
e = int(WeapiCrypto.RSA_EXPONENT, 16)
reversed_hex = secret_key[::-1].encode('utf-8').hex()
enc_sec_key = format(pow(int(reversed_hex, 16), e, n), '0>256x')
return {"params": params, "encSecKey": enc_sec_key}
music_api/errors.py
+11
View File
@@ -0,0 +1,11 @@
# -*- coding: utf-8 -*-
"""API 异常。
（自 music_api.py 拆出，行为与原实现一致。）
"""
class APIException(Exception):
"""API异常类"""
pass
music_api/qr.py
+181
View File
@@ -0,0 +1,181 @@
# -*- coding: utf-8 -*-
"""扫码登录管理器。
（自 music_api.py 拆出，行为与原实现一致。）
"""
import asyncio
import json
from random import randrange
from .client import (
HTTPClient,
_ensure_dict,
)
from .constants import (
APIConstants,
)
from .crypto import (
CryptoUtils,
)
from .errors import (
APIException,
)
class QRLoginManager:
"""二维码登录管理器"""
def __init__(self):
self.http_client = HTTPClient()
self.crypto_utils = CryptoUtils()
async def generate_qr_key(self) -> str | None:
"""生成二维码的key
Returns:
成功返回unikey，失败返回None
Raises:
APIException: API调用失败时抛出
"""
try:
config = APIConstants.DEFAULT_CONFIG.copy()
config["requestId"] = str(randrange(20000000, 30000000))
payload = {
'type': 1,
'header': json.dumps(config)
}
params = self.crypto_utils.encrypt_params(APIConstants.QR_UNIKEY_API, payload)
response = await self.http_client.post_request_full(APIConstants.QR_UNIKEY_API, params, {})
result = json.loads(response.text)
# 非 dict 响应（null/数组）不再 AttributeError 逃出捕获
_ensure_dict(result, '生成二维码key响应异常（非标准 JSON 对象）')
if result.get('code') == 200:
return result.get('unikey')
else:
raise APIException(f"生成二维码key失败: {result.get('message', '未知错误')}")
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析二维码key响应失败: {e}") from e
async def create_qr_login(self) -> str | None:
"""创建登录二维码并在控制台显示
Returns:
成功返回unikey，失败返回None
"""
try:
import qrcode
unikey = await self.generate_qr_key()
if not unikey:
print("生成二维码key失败")
return None
# 创建二维码
qr = qrcode.QRCode()
qr.add_data(f'https://music.163.com/login?codekey={unikey}')
qr.make(fit=True)
# 在控制台显示二维码
qr.print_ascii(tty=True)
print("\n请使用网易云音乐APP扫描上方二维码登录")
return unikey
except ImportError:
print("请安装qrcode库: pip install qrcode")
return None
except Exception as e:
print(f"创建二维码失败: {e}")
return None
async def check_qr_login(self, unikey: str) -> tuple[int, dict[str, str]]:
"""检查二维码登录状态
Args:
unikey: 二维码key
Returns:
(登录状态码, cookie字典)
Raises:
APIException: API调用失败时抛出
"""
try:
config = APIConstants.DEFAULT_CONFIG.copy()
config["requestId"] = str(randrange(20000000, 30000000))
payload = {
'key': unikey,
'type': 1,
'header': json.dumps(config)
}
params = self.crypto_utils.encrypt_params(APIConstants.QR_LOGIN_API, payload)
response = await self.http_client.post_request_full(APIConstants.QR_LOGIN_API, params, {})
result = json.loads(response.text)
# 非 dict 响应不再 AttributeError 逃出捕获（同 generate_qr_key）
_ensure_dict(result, '解析登录状态响应失败（非标准 JSON 对象）')
cookie_dict = {}
if result.get('code') == 803:
# 登录成功，提取cookie。
# 此前只提取 MUSIC_U——__csrf 从未入 Cookie 串，
# weapi 写接口（发评论等）的 csrf_token 恒为空。改用 requests 的
# cookie jar（requests 已把多条 Set-Cookie 合并解析，不受逗号合并
# 的脆弱 split 影响）补齐 MUSIC_U 与 __csrf。
for name in ('MUSIC_U', '__csrf'):
val = response.cookies.get(name)
if val:
cookie_dict[name] = val
if not cookie_dict.get('MUSIC_U'):
# 兜底：jar 异常为空时回退 Set-Cookie 头解析 MUSIC_U
for cookie_str in response.headers.get('Set-Cookie', '').split(', '):
if 'MUSIC_U=' in cookie_str:
cookie_dict['MUSIC_U'] = cookie_str.split('MUSIC_U=')[1].split(';')[0]
break
return result.get('code', -1), cookie_dict
except (json.JSONDecodeError, KeyError) as e:
raise APIException(f"解析登录状态响应失败: {e}") from e
async def qr_login(self) -> str | None:
"""完整的二维码登录流程
Returns:
成功返回cookie字符串，失败返回None
"""
try:
unikey = await self.create_qr_login()
if not unikey:
return None
while True:
code, cookies = await self.check_qr_login(unikey)
if code == 803:
print("\n登录成功！")
# CLI 序列化须与 qr_login 口径一致，带上 __csrf
# （此串正是写往全局 cookie.txt 的内容）
serialized = f"MUSIC_U={cookies['MUSIC_U']};os=pc;appver=8.9.70;"
if cookies.get('__csrf'):
serialized += f"__csrf={cookies['__csrf']};"
return serialized
elif code == 801:
print("\r等待扫码...", end='')
elif code == 802:
print("\r扫码成功，请在手机上确认登录...", end='')
else:
print(f"\n登录失败，错误码：{code}")
return None
await asyncio.sleep(2)
except KeyboardInterrupt:
print("\n用户取消登录")
return None
except Exception as e:
print(f"\n登录过程中发生错误: {e}")
return None