diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..8fead57 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,43 @@ +name: CI +on: + push: + branches: [master] + pull_request: + workflow_dispatch: +permissions: + contents: read +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true +jobs: + test: + name: ${{ matrix.os }} / Python ${{ matrix.python }} + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + python: ['3.10', '3.12', '3.14'] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + - run: python -m pip install -e ".[test]" + - run: python -m pytest --cov=cloudmusic --cov-report=term-missing --cov-fail-under=90 + - run: python -m ruff check cloudmusic tests + package: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.14' + - run: python -m pip install build + - run: python -m build + - name: Install built wheel in isolation + run: | + python -m venv /tmp/cloudmusic-wheel + /tmp/cloudmusic-wheel/bin/pip install dist/*.whl + cd /tmp + /tmp/cloudmusic-wheel/bin/python -c "import cloudmusic; from importlib.metadata import version; assert cloudmusic.__version__ == version('cloudmusic'); print(cloudmusic.__file__)" diff --git a/.gitignore b/.gitignore index 7b0eb25..be5cdfb 100644 --- a/.gitignore +++ b/.gitignore @@ -89,3 +89,6 @@ ENV/ .ropeproject *.npy *.pkl +.venv/ +.pytest_cache/ +.ruff_cache/ diff --git a/MANIFEST.in b/MANIFEST.in index 64ad321..b1d872e 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1 +1,3 @@ include README.md LICENSE +recursive-include tests *.py +recursive-include docs *.md diff --git a/README.html b/README.html index fd4ffc5..501bcda 100644 --- a/README.html +++ b/README.html @@ -1,500 +1,9 @@ - - README - - - - - - - - - - - - - - - - - - -
-

cloudmusic: 网易云爬虫解决方案

-

一个功能强大的python库,使用最简单易用的方法获取一切你想要的信息。
-支持python3以上版本

-

安装

-
pip install cloudmusic
-

简单示例

- -
music = cloudmusic.getMusic(1347630432)
-
-print("歌名:{}".format(music.name))
-print("歌手:{}".format(music.artist))
-print("音频文件url:{}".format(music.url))
-
-# >>>
-# 歌名:白日 日剧《冤罪律师》主题曲
-# 歌手:['King Gnu']
-# 音频文件url:http://m10.music.126.net/.../.mp3
-
-
-
playlist = cloudmusic.getPlaylist(310729011)
-
-for music in playlist:
-    music.download(level = "lossless")
-
-
-
music = cloudmusic.getMusic(1347630432)
-
-coms = music.getHotComments()
-
-for com in coms:
-    print("发布者:{}".format(com['nickName']))
-    print("内容:{}".format(com['content']))
-    print("获赞数:{}".format(com['likeCount']))
-    print("------------")
-
-# >>>
-# 发布者:Akikonknk
-# 内容:特别喜欢日剧的一个原因 大概就是它帮你认清生活的真相之后 依然教会你如何热爱生活
-# 获赞数:19451
-# ------------
-# 发布者:我藏起来啦嘻嘻
-# 内容:最后一集的犯人真的是应证了那一句话“雪崩的时候没有一片雪花是无辜的”
-# 获赞数:6690
-# ------------
-# 发布者:Usio叔叔
-# ...
-
-
-
results = cloudmusic.search("白日", 100)
-
-for music in results:
-    print(music.getCommentsCount())
-
-# 5745
-# 77
-# 181
-# 560
-# 368
-# ...
-
-

-

参考文档

-

Music对象

- -

1.Music对象属性

- - -

2.Music对象方法

- - -

3.获取Music对象函数

- - -
-

User对象

- -

1.User对象属性

- - -

2.User对象方法

- - -

3.获取User对象函数

- - -
-

@p697

- -
- - - - - - - - - - - - \ No newline at end of file + + +cloudmusic 文档 + +

cloudmusic 文档

+

项目已恢复维护。原静态 HTML/PDF 文档对应 0.1.1,当前接口、安装方法和迁移说明请查看仓库文档。

+

阅读当前 README · 查看验收记录和已知限制

+ + diff --git a/README.md b/README.md index 152ea56..479a714 100644 --- a/README.md +++ b/README.md @@ -1,224 +1,188 @@ -cloudmusic: 网易云爬虫解决方案 -======================= +# cloudmusic -一个功能强大的python库,使用最简单易用的方法获取一切你想要的信息。 -支持python3以上版本 +一个轻量的网易云音乐 Python 客户端:查询歌曲、专辑、歌单、歌词、评论和用户公开资料,并下载当前账号有权访问的完整音频。 -安装 ------ -```bash -pip install cloudmusic -``` - -简单示例 ------ -- 通过歌曲id获取Music对象,并展示部分属性 -```python -music = cloudmusic.getMusic(1347630432) +支持 Python 3.10–3.14。保留 `getMusic()`、`getPlaylist()`、`search()` 等原有调用方式。 -print("歌名:{}".format(music.name)) -print("歌手:{}".format(music.artist)) -print("音频文件url:{}".format(music.url)) +> 仓库维护版本为 **0.2.0**。在发布到 PyPI 之前,`pip install cloudmusic` 仍可能安装旧的 0.1.1。请按下面的源码安装方式使用当前版本。历史 HTML/PDF 文档对应 0.1.1,当前接口以本 README 为准。 -# >>> -# 歌名:白日 日剧《冤罪律师》主题曲 -# 歌手:['King Gnu'] -# 音频文件url:http://m10.music.126.net/.../.mp3 +## 安装 +```bash +git clone https://github.com/p697/cloudmusic.git +cd cloudmusic +python -m pip install . ``` -- 通过歌单id获取Music对象列表,并下载整个歌单的无损品质音频 -```python -playlist = cloudmusic.getPlaylist(310729011) +运行时仅依赖 `requests` 和 `pycryptodomex`,不再要求 `lxml` 或 `threadpool`。 -for music in playlist: - music.download(level = "lossless") +## 快速开始 -``` - -- 获取一首歌的热评并展示部分信息 ```python +import cloudmusic + music = cloudmusic.getMusic(1347630432) +print(music.name, music.artist, music.album) +print(music.getLyrics()) # [原文歌词, 翻译歌词];缺失项为空字符串 -coms = music.getHotComments() - -for com in coms: - print("发布者:{}".format(com['nickName'])) - print("内容:{}".format(com['content'])) - print("获赞数:{}".format(com['likeCount'])) - print("------------") - -# >>> -# 发布者:Akikonknk -# 内容:特别喜欢日剧的一个原因 大概就是它帮你认清生活的真相之后 依然教会你如何热爱生活 -# 获赞数:19451 -# ------------ -# 发布者:我藏起来啦嘻嘻 -# 内容:最后一集的犯人真的是应证了那一句话“雪崩的时候没有一片雪花是无辜的” -# 获赞数:6690 -# ------------ -# 发布者:Usio叔叔 -# ... +results = cloudmusic.search("白日", number=5) +for song in results: + print(song.id, song.name) +playlist = cloudmusic.getPlaylist(310729011) +print(len(playlist)) ``` -- 通过关键字搜索获取前100首歌曲的Music对象,并输出每首歌的评论数量 -```python -results = cloudmusic.search("白日", 100) - -for music in results: - print(music.getCommentsCount()) - -# 5745 -# 77 -# 181 -# 560 -# 368 -# ... +歌曲信息与音频权限相互独立。获取歌曲名称、专辑、歌词或评论不需要先获取播放地址。访问 `music.url`、`size`、`type`、`level` 或 `freeTrialInfo` 时,才获取音频信息;同一批歌曲共享批量请求。音频信息缓存 60 秒,以减少重复请求和过期链接。 +```python +try: + print(music.url) # 没有播放权限时可能为 None;接口风控会抛出 APIError + print(music.level) # 服务实际返回的品质,可能低于请求品质 + path = music.download(dirs="./downloads") + print(path) # 已保存文件的绝对路径 +except cloudmusic.CloudMusicError as error: + print(error) ``` ----- - -参考文档 ------ -### Music对象 -#### 1.Music对象属性 -- **url**:歌曲音频文件链接 -- **id**:歌曲id -- **name**:歌曲名称 -- **artist**:歌手名称 -- **artistId**: 歌手id -- **album**:专辑名称 -- **albumId**: 专辑id -- **size**:音频文件大小 -- **type**:音频文件类型(mp3或m4a) -- **level**:歌曲品质。默认higher -- **picUrl**: 专辑图url - -#### 2.Music对象方法 -- #### `download(dirs, level)` - >***dirs***:可选。下载保存路径。默认为当前文件夹内创建的新的cloudmusic文件夹。 - ***level***:可选,字符型。默认higher。下载品质,有且只有四种选择:standard,higher,exhigh,lossless。 - - 下载歌曲,返回值为下载绝对路径 - - -- #### `getHotComments(number)` - >***number***:可选,整型,默认为15。希望获取的评论个数。上限为15个。 - 获取热评。返回值为列表,列表元素为字典,字典内容: - *"likeCount":获赞数* - *"content":评论内容* - *"time":评论时间* - *"userId":用户id* - *"nickNamd":用户昵称* - *"avatarUrl":用户头像url* - *"vipType":0表示未开通vip,10表示开通音乐包,11表示开通黑胶* - *"userType":0表示普通用户,非0表示特殊用户(明星,丁磊,网红,小秘书等)* +下载采用流式写入和临时文件替换。只有成功完成下载才覆盖同名目标文件;失败会保留已有文件并清理临时文件。仅有试听音频时抛出 `UnavailableError`,不会把试听片段当成完整歌曲保存。 +## 配置会话 -- #### `getComments(number)` - >***number***:整型,评论个数。数量无限制。 +默认匿名访问。需要账号权限时,可以显式传入自己的 cookie;库不包含任何内置登录凭据,不自动读取浏览器或环境中的凭据。 - 获取最新的评论。返回值与getHotReview相同。 - - -- #### `getLyrics()` - 获取歌词。返回值为列表。元素1为原始歌词,元素2为翻译歌词。 - - -- #### `getCommentsCount()` - 返回值为整型,本首歌的评论数量。 - -#### 3.获取Music对象函数 - -- #### `getMusic(id/id_list)` - >***id/id_list***: 必须,整型、字符型或列表。歌曲id或由歌曲id组成的列表。 - - 通过歌曲id或一个由id组成的列表生成music对象。当参数为单个id时返回值为单个music对象,当参数为id列表时返回值为music对象组成的列表。 - -- #### `getPlaylist(id)` - >***id***:必须,整型或字符型。歌单id。 - - 通过歌单id生成music对象,参数为歌单id,返回值为歌单内所有歌曲对应music对象组成的列表。 - -- #### `search(content, number)` - >***content***:必须,字符型。搜索关键字。 - ***number***:可选,整形,搜索结果个数,默认为5。 +```python +import os +import cloudmusic + +with cloudmusic.Client( + cookie=os.environ.get("NETEASE_COOKIE"), + timeout=(5, 20), # 连接超时、单次读取超时,单位秒 + level="higher", + # proxies={"https": "http://127.0.0.1:7890"}, +) as client: + music = client.getMusic(1347630432) + songs = client.search("白日", 5) + +# 返回的 Music/User 对象保留配置;关闭 Client 后仍可按需调用。 +print(music.getLyrics()) +``` - 通过关键词搜索获取music对象。返回值为music对象组成的列表。 +`cookie` 支持 cookie 字符串或字典。`Client` 复用查询连接并保留服务端更新的会话 cookie,使用完毕后应调用 `close()` 或用 `with` 管理。网络错误和风控错误不会自动反复重试;同批音频遇到验证/登录限制后,在 60 秒内复用该错误,避免批量任务重复触发限制。音频 CDN 请求不携带网易云登录 cookie。 -- #### `getAlbum(id)` - >***id***: 必须,整型或字符型。专辑id。 +该项目依赖网易云的非官方接口。接口返回内容受登录状态、隐私设置、地区和音频权限限制;升级客户端不能保证获取所有歌曲、无限评论或指定音质。不要将 cookie 提交到 Git 或输出到公开日志。 - 和getPlaylist一样。 +## API ---- +以下函数也可以通过 `Client` 调用: -### User对象 -#### 1.User对象属性 -- **id**:用户id -- **level**:用户等级 -- **listenSongs**:累计听歌数量 -- **createTime**:账号创建时间 -- **nickName**:用户昵称 -- **avatarUrl**:头像url -- **city**:所在城市的行政区划代码 -- **province**:所在省份的行政区划代码 -- **vipType**:vip类型。0表示未开通vip,10表示开通音乐包,11表示开通黑胶 -- **birthday**:生日时间戳 -- **signature**:个性签名 -- **fans**:粉丝数量 -- **follows**:关注的用户数量 -- **eventCount**:动态数量 -- **playlistCount**:创建的歌单数量 +| 函数 | 返回值 | +| --- | --- | +| `getMusic(id)` | 一个 `Music` 对象 | +| `getMusic([id, ...])` | `Music` 列表,保留输入顺序和重复 ID;单元素列表仍返回列表;空列表返回 `[]` | +| `getPlaylist(id)` | 按歌单顺序排列的 `Music` 列表 | +| `getAlbum(id)` | 按专辑顺序排列的 `Music` 列表 | +| `search(content, number=5)` | 最多 `number` 个不重复的搜索结果;支持分页;无结果返回 `[]` | +| `getUser(id)` | `User` 对象 | -#### 2.User对象方法 +ID 支持正整数或整数字符串。无效参数抛出 `ValueError`;歌曲 ID 不存在时抛出 `NotFoundError`,其 `ids` 字段包含缺失 ID。歌单返回数量与服务报告的总数不一致时抛出 `IncompleteResultError`,不会静默当作完整歌单。 -- #### `getPlaylist()` - 获取用户创建和收藏的全部歌单,返回值为一个列表,每个元素是一个歌单信息组成的字典。字典内容: - *id:歌单id;* - *name:歌单名称;* - *creatorId:创建者id;* - *createTime:创建时间;* - *coverImgUrl:歌单封面图片url;* - *updateTime:歌单最后一次更新时间;* - *tags:歌单所属音乐风格,列表形式* +```python +try: + songs = cloudmusic.getPlaylist(310729011) +except cloudmusic.IncompleteResultError as error: + print(error.actual_count, error.expected_count) + songs = error.partial_result # 调用方明确选择是否使用已获取部分 +``` -- #### `getRecord(recordType)` - >***recordType***: 可选。默认为0。为0返回所有时间排行,非0返回最近一周排行。 +### Music - 获取用户听歌排行,返回一个列表,元素为字典。字典内容: - *score:分值。100表示听歌次数最多,越小越少;* - *music:Music对象* +元数据属性:`id`(字符串)、`name`、`artist`、`artistId`、`album`、`albumId`、`picUrl`。 -#### 3.获取User对象函数 +按需获取的音频属性:`url`、`size`、`type`、`level`、`freeTrialInfo`。 -- #### `getUser(id)` - >***id***:必须,用户id。 +| 方法 | 行为 | +| --- | --- | +| `download(dirs="", level=None)` | 默认保存到当前目录的 `cloudmusic` 文件夹;`level=None` 沿用创建对象时的请求品质(默认 `higher`) | +| `getLyrics()` | 返回 `[原文, 翻译]`,缺失项为空字符串 | +| `getHotComments(number=15)` | 最多 15 条热评 | +| `getComments(number)` | 按时间从新到旧,使用游标分页;不足时返回可获取的数量 | +| `getCommentsCount()` | 服务报告的评论总数 | - 通过用户id获取User对象,返回值为User对象。 +品质参数保留 `standard`、`higher`、`exhigh`、`lossless`。实际品质由接口和账号权限决定。指定音质不会丢失下载目录,文件扩展名取服务实际返回值。 ---- -@p697 +评论字段:`commentId`、`likeCount`、`content`、`time`、`userId`、`nickName`、`avatarUrl`、`vipType`、`userType`。用户已注销或字段缺失时,昵称和头像可以为空。若服务表示还有数据但分页无法继续,抛出带 `partial_result` 的 `IncompleteResultError`,避免无限循环或重复结果。 +### User +属性:`id`、`level`、`listenSongs`、`createTime`、`nickname`(也支持 `nickName`)、`avatarUrl`、`city`、`province`、`vipType`、`birthday`、`signature`、`fans`、`follows`、`eventCount`、`playlistCount`、`sex`。部分资料可能因隐私设置缺失。 +- `getPlaylist()`:分页获取可访问的用户歌单,返回信息字典列表。字段包括 `id`、`name`、`creatorId`、`playCount`、`createTime`、`coverImgUrl`、`updateTime`、`tags`。 +- `getRecord(recordType=0)`:0 为所有时间,正整数为最近一周;返回 `{"score": ..., "music": Music}` 列表,只请求所需周期。 +### 批量下载 +```python +loader = cloudmusic.createLoader(procs=2, dirs="./downloads") +loader.data = cloudmusic.getPlaylist(310729011) +try: + loader.start() +except cloudmusic.DownloadError: + print(loader.errors) # 失败项:以 data 中的索引为键 +print(loader.results) # 成功保存的绝对路径,按输入顺序排列 +``` +最多同时使用 32 个工作线程。文件名会处理跨平台非法字符和过长名称;同名歌曲仍会覆盖同一目标文件,请选择适当的目录管理下载结果。 +### 错误类型 +均继承 `CloudMusicError`,无效调用参数使用标准 `ValueError`。 +| 类型 | 含义 | +| --- | --- | +| `RequestError` | 网络/超时/非 200 HTTP 响应;`status_code` 在 HTTP 失败时可用 | +| `APIError` | 网易云业务错误;保留 `code` 和 `message`,包括风控、验证或登录要求 | +| `ResponseError` | JSON 或响应结构异常 | +| `NotFoundError` | 指定歌曲不存在,保留 `ids` | +| `UnavailableError` | 没有完整音频可供下载 | +| `IncompleteResultError` | 歌单不完整或分页停止前进;保留 `partial_result` | +| `DownloadError` | 文件下载未完整完成或批量下载部分失败 | +## 测试与开发 +```bash +python -m pip install -e '.[test]' +python -m pytest --cov=cloudmusic --cov-branch --cov-report=term-missing +python -m ruff check cloudmusic tests +python -m build +``` +默认测试禁止外部联网,并通过本地 HTTP 服务验证流式下载、错误响应、中断、临时文件清理和已有文件保护。CI 在 Linux、macOS、Windows 上运行 Python 3.10、3.12、3.14 的测试,以及独立 wheel 安装检查。 +真实接口测试显式启用,依赖当前网络环境和服务状态,不参与必过 CI: +```bash +python -m pytest --live -m live -v +``` +真实测试默认匿名,不修改账号内容,也不批量下载音频。远端拒绝会表现为测试失败,不会被记为成功。若已在本机配置测试账号的 `NETEASE_COOKIE` 环境变量,可以明确启用登录态测试: +```bash +python -m pytest --live --cookie-env NETEASE_COOKIE -m live -v --tb=short +``` +未传 `--cookie-env` 时测试不会读取任何登录 cookie。具体实测范围和限制见 [验收记录](docs/verification.md)。 +## 从 0.1.1 迁移 +- Python 最低版本更新为 3.10。 +- 原有高层函数和对象属性保留,修正了单元素列表返回类型。 +- 音频属性按需读取;网络/权限错误可能在访问这些属性时出现。 +- 下载默认沿用对象品质;明确传入 `level="standard"` 时会请求该品质。 +- 错误不再以字符串、打印信息、`None` 或意外 `KeyError` 混合返回;请处理相应异常。 +- `query.py` 内部旧 HTML 抓取函数已移除。内部模块不是稳定 API。 +- 尚未自动发布到 PyPI。 +历史贡献参考:[PR #14](https://github.com/p697/cloudmusic/pull/14)、[PR #17](https://github.com/p697/cloudmusic/pull/17)、[PR #21](https://github.com/p697/cloudmusic/pull/21)。本轮吸收了跨平台路径和音质目录传递的修复方向;封面/标签写入功能暂未纳入。 +MIT License · @p697 diff --git a/cloudmusic/__init__.py b/cloudmusic/__init__.py index 47f2899..628d735 100644 --- a/cloudmusic/__init__.py +++ b/cloudmusic/__init__.py @@ -1 +1,36 @@ -from .cloudmusic import * \ No newline at end of file +from .__version__ import __version__ +from .cloudmusic import Client, createLoader, getAlbum, getMusic, getPlaylist, getUser, help, search +from .exceptions import ( + APIError, + CloudMusicError, + DownloadError, + IncompleteResultError, + NotFoundError, + RequestError, + ResponseError, + UnavailableError, +) +from .musicObj import Music +from .userObj import User + +__all__ = [ + "Client", + "Music", + "User", + "getMusic", + "getPlaylist", + "getAlbum", + "getUser", + "search", + "createLoader", + "help", + "__version__", + "CloudMusicError", + "APIError", + "RequestError", + "ResponseError", + "NotFoundError", + "UnavailableError", + "IncompleteResultError", + "DownloadError", +] diff --git a/cloudmusic/__version__.py b/cloudmusic/__version__.py index c5f1956..d3ec452 100644 --- a/cloudmusic/__version__.py +++ b/cloudmusic/__version__.py @@ -1,8 +1 @@ -# 8b d8 Yb dP 88""Yb db dP""b8 88 dP db dP""b8 888888 -# 88b d88 YbdP 88__dP dPYb dP `" 88odP dPYb dP `" 88__ -# 88YbdP88 8P 88""" dP__Yb Yb 88"Yb dP__Yb Yb "88 88"" -# 88 YY 88 dP 88 dP""""Yb YboodP 88 Yb dP""""Yb YboodP 888888 - -VERSION = (0, 1, 0) - -__version__ = '.'.join(map(str, VERSION)) +__version__ = "0.2.0" diff --git a/cloudmusic/api.py b/cloudmusic/api.py index 290f1be..9b2c53b 100644 --- a/cloudmusic/api.py +++ b/cloudmusic/api.py @@ -1,105 +1,159 @@ -from . import query -from . import encrypt +"""NetEase read-only endpoints and a reusable, caller-configured connection.""" import json -import random +import math +from collections.abc import Mapping +from http.cookies import SimpleCookie +import requests + +from . import encrypt, query -class Api: - def __init__(self): - self.userAgent = [ - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.36", - "Mozilla/5.0 (iPhone; CPU iPhone OS 9_1 like Mac OS X) AppleWebKit/601.1.46 (KHTML, like Gecko) Version/9.0 Mobile/13B143 Safari/601.1", - "Mozilla/5.0 (iPhone; CPU iPhone OS 9_1 like Mac OS X) AppleWebKit/601.1.46 (KHTML, like Gecko) Version/9.0 Mobile/13B143 Safari/601.1", - "Mozilla/5.0 (Linux; Android 5.0; SM-G900P Build/LRX21T) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Mobile Safari/537.36", - "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Mobile Safari/537.36", - "Mozilla/5.0 (Linux; Android 5.1.1; Nexus 6 Build/LYZ28E) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Mobile Safari/537.36", - "Mozilla/5.0 (iPhone; CPU iPhone OS 10_3_2 like Mac OS X) AppleWebKit/603.2.4 (KHTML, like Gecko) Mobile/14F89;GameHelper", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_5) AppleWebKit/603.2.4 (KHTML, like Gecko) Version/10.1.1 Safari/603.2.4", - "Mozilla/5.0 (iPhone; CPU iPhone OS 10_0 like Mac OS X) AppleWebKit/602.1.38 (KHTML, like Gecko) Version/10.0 Mobile/14A300 Safari/602.1", - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.103 Safari/537.36", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.12; rv:46.0) Gecko/20100101 Firefox/46.0", - "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:46.0) Gecko/20100101 Firefox/46.0", - "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0)", - "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0)", - "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)", - "Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.2; Win64; x64; Trident/6.0)", - "Mozilla/5.0 (Windows NT 6.3; Win64, x64; Trident/7.0; rv:11.0) like Gecko", - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/42.0.2311.135 Safari/537.36 Edge/13.10586", - "Mozilla/5.0 (iPad; CPU OS 10_0 like Mac OS X) AppleWebKit/602.1.38 (KHTML, like Gecko) Version/10.0 Mobile/14A300 Safari/602.1" - ] - musicU = 'e4ace7b13afdd88161175bda1c091445ad6bc398f36fb0f0315f1fa0787d4fbce7c70899629a7e58dcb329764b52b00341049cea1c6bb9b6' - self.CookiesList = [ - 'os=pc; osver=Microsoft-Windows-10-Professional-build-10586-64bit; appver=2.0.3.131777; channel=netease; __remember_me=true', - 'MUSIC_U=' + musicU + '; buildver=1506310743; resolution=1920x1080; mobilename=MI5; osver=7.0.1; channel=coolapk; os=android; appver=4.2.0', - 'osver=%E7%89%88%E6%9C%AC%2010.13.3%EF%BC%88%E7%89%88%E5%8F%B7%2017D47%EF%BC%89; os=osx; appver=1.5.9; MUSIC_U=' + - musicU + '; channel=netease;' - ] - self.headers = { - 'User-Agent': random.choice(self.userAgent), - 'cookie': random.choice(self.CookiesList), - 'referer': 'https://music.163.com' +class Api: + def __init__(self, *, cookie=None, timeout=(5, 20), proxies=None): + values = timeout if isinstance(timeout, (list, tuple)) else (timeout,) + if len(values) not in (1, 2) or any( + isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or v <= 0 + for v in values + ): + raise ValueError("timeout 必须是正数或 (连接超时, 读取超时)") + self.timeout = tuple(values) if len(values) == 2 else values[0] + self.proxies = dict(proxies or {}) + cookies = {"os": "pc"} + if isinstance(cookie, str): + parsed = SimpleCookie() + parsed.load(cookie) + if cookie.strip() and not parsed: + raise ValueError("cookie 格式无效") + cookies.update({key: value.value for key, value in parsed.items()}) + elif cookie is not None: + if not isinstance(cookie, Mapping): + raise ValueError("cookie 必须是字符串或字典") + cookies.update(cookie) + self.options = { + "cookie": dict(cookies), + "timeout": self.timeout, + "proxies": dict(self.proxies), } + self.session = requests.Session() + self.session.proxies.update(self.proxies) + for name, value in cookies.items(): + self.session.cookies.set(name, str(value), domain="music.163.com", path="/") + self.headers = {"User-Agent": "Mozilla/5.0", "Referer": "https://music.163.com/"} + self._closed = False + + def __enter__(self): + return self - def send(self, url, param={}, method=''): - return query.post(url, - self.headers, - encrypt.encrypted_request(param, method)) + def __exit__(self, *args): + self.close() + + def close(self): + self.session.close() + self._closed = True + + def send(self, url, param=None, method=""): + if self._closed: + raise RuntimeError("Client 已关闭") + if url.startswith("/"): + url = "https://music.163.com" + url + params = dict(param or {}) + params.setdefault("csrf_token", self.options["cookie"].get("__csrf", "")) + try: + return query.post( + url, + self.headers, + encrypt.encrypted_request(params, method), + session=self.session, + timeout=self.timeout, + ) + finally: + # Preserve server-issued session cookies for lazy requests on returned objects. + self.options["cookie"].update( + { + cookie.name: cookie.value + for cookie in self.session.cookies + if cookie.domain.lstrip(".") == "music.163.com" + } + ) def get_song_url(self, para): - url = "https://music.163.com/weapi/song/enhance/player/url/v1" - param = dict(ids=para["ID"], level=para["level"], encodeType="aac") - return self.send(url, param) + return self.send( + "/weapi/song/enhance/player/url/v1", + { + "ids": para["ID"], + "level": para["level"], + "encodeType": "aac", + }, + ) def search(self, para): - url = "https://music.163.com/weapi/cloudsearch/get/web" - param = '{"hlpretag":"","hlposttag":"","s":"' + \ - para["string"] + '","type":"1","offset":"0","total":"true","limit":"' + \ - para["number"] + '","csrf_token":""}' - return self.send(url, param) + return self.send( + "/weapi/search/get", + { + "s": para["string"], + "type": 1, + "offset": para.get("offset", 0), + "limit": para["number"], + }, + ) def get_commets(self, para): - url = "https://music.163.com/weapi/v1/resource/comments/R_SO_4_{}".format( - para["ID"]) - param = '{"rid":"R_SO_4_' + para["ID"] + '","offset":"' + para["offset"] + \ - '","total":"' + para["total"] + '","limit":"' + \ - para["limit"] + '","csrf_token":""}' - return self.send(url, param) + # Retain the historical internal spelling for integrations using Api. + return self.send( + "/weapi/v1/resource/comments/R_SO_4_{}".format(para["ID"]), + { + "rid": str(para["ID"]), + "offset": para.get("offset", 0), + "beforeTime": 0, + "limit": para.get("limit", 20), + }, + ) + + def get_comment_page(self, para): + return self.send( + "/weapi/v2/resource/comments", + { + "threadId": "R_SO_4_{}".format(para["ID"]), + "pageNo": para["page"], + "pageSize": para["limit"], + "sortType": 3, + "cursor": para["cursor"], + "showInner": True, + }, + ) def get_lyrics(self, para): - url = "https://music.163.com/weapi/song/lyric" - param = '{"id":"' + para["ID"] + '","lv":-1,"tv":-1,"csrf_token":""}' - return self.send(url, param) + return self.send("/weapi/song/lyric", {"id": para["ID"], "lv": -1, "tv": -1}) def get_song_detail(self, para): - url = "https://music.163.com/weapi/v3/song/detail" - param = dict(c=json.dumps( - [{"id": _id} for _id in para["ID"]]), ids=json.dumps(para["ID"])) - return self.send(url, param) + return self.send( + "/weapi/v3/song/detail", + { + "c": json.dumps([{"id": value} for value in para["ID"]]), + "ids": json.dumps(para["ID"]), + }, + ) - # 是个让人脑阔疼的api def get_playlist(self, para, method=""): - url = "http://music.163.com/api/v3/playlist/detail" - param = dict(id=para["ID"], total="true", limit=1000, n=1000, offest=0) - return self.send(url, param, method=para["method"]) + return self.send("/weapi/v6/playlist/detail", {"id": para["ID"], "n": 100000, "s": 8}) def get_album(self, para): - url = "https://music.163.com/weapi/v1/album/{}".format(para["ID"]) - return self.send(url) + return self.send("/weapi/v1/album/{}".format(para["ID"])) def get_userInfo(self, para): - url = "https://music.163.com/api/v1/user/detail/{}".format(para["ID"]) - return self.send(url) + return self.send("/weapi/v1/user/detail/{}".format(para["ID"])) def get_userPlayerlist(self, para): - url = "https://music.163.com/weapi/user/playlist" - params = dict(uid=para["ID"], wordwrap=7, - offset=0, total="true", limit=1000) - return self.send(url, params) + return self.send( + "/weapi/user/playlist", + { + "uid": para["ID"], + "offset": para.get("offset", 0), + "limit": para.get("limit", 100), + }, + ) def get_userRecord(self, para): - url = "https://music.163.com/weapi/v1/play/record" - params = dict(uid=para["ID"], type=-1, - limit=1000, offset=0, total="true") - return self.send(url, params) + return self.send("/weapi/v1/play/record", {"uid": para["ID"], "type": para.get("type", 0)}) diff --git a/cloudmusic/cloudmusic.py b/cloudmusic/cloudmusic.py index 93934b7..35c32d9 100644 --- a/cloudmusic/cloudmusic.py +++ b/cloudmusic/cloudmusic.py @@ -1,36 +1,43 @@ -from . import sessions +"""Convenience functions; use Client for explicit configuration and connection reuse.""" + +from .sessions import Session as Client def getMusic(para): - with sessions.Session() as session: - return session.request("song", para) + with Client() as client: + return client.getMusic(para) def getPlaylist(para): - with sessions.Session() as session: - return session.request("playlist", para) + with Client() as client: + return client.getPlaylist(para) def getAlbum(para): - with sessions.Session() as session: - return session.request("album", para) + with Client() as client: + return client.getAlbum(para) def search(para, number=5): - with sessions.Session() as session: - return session.search(para, number) + with Client() as client: + return client.search(para, number) def createLoader(procs=2, dirs=""): - with sessions.Session() as session: - return session.downloader(procs, dirs) + from .download import Downloader + + return Downloader(procs, dirs) def getUser(para): - with sessions.Session() as session: - return session.request("user", para) + with Client() as client: + return client.getUser(para) def help(): - print("这是帮助") - + print( + "cloudmusic: getMusic(id 或 id 列表), getPlaylist(id), getAlbum(id), " + "search(关键词, number=5), getUser(id), createLoader(procs=2, dirs='')。\n" + "使用 Client(cookie=..., timeout=(5, 20), proxies=...) 配置会话。\n" + "文档:https://github.com/p697/cloudmusic#readme" + ) diff --git a/cloudmusic/download.py b/cloudmusic/download.py index 3f6893b..f0c36e6 100644 --- a/cloudmusic/download.py +++ b/cloudmusic/download.py @@ -1,94 +1,116 @@ -from . import musicObj +"""Cross-platform, streamed and atomic downloads.""" -import urllib import os -import threadpool -import time - - - -def download(dirs, music): - if len(music.artist) == 1: - artist = music.artist[0] - else: - artist = "" - for ar in music.artist: - artist += ar + " " - name = music.name + " - " + artist + "." + music.type - - if not dirs: - dirs = "cloudmusic\\" + name - defalut_dirs = str(os.getcwd()) + '\\cloudmusic' - isExist = os.path.exists(defalut_dirs) - if not isExist: - os.makedirs(defalut_dirs) - else : - dirs += "\\" + name - - # 超时重连 - for t in range(5): - try: - resp = urllib.request.urlopen(music.url, timeout=5) - respHtml = resp.read() - break - except Exception as e: - if t == 4: - print("download failed - " + music.id) - return None - print("Error: " + str(e) + " - " + "reconnect time: " + str(t)) - time.sleep(1) - - - binfile = open(dirs, "wb") - binfile.write(respHtml) - binfile.close() - - print("dowload finish - " + music.id) - - return str(os.getcwd()) + dirs - - -class Downloader(): - def __init__(self, procs, dirs): - self.data = [] - self.dirs = dirs - self.procs = procs - - def start(self): - if not self.data: - print("data 为空") - return None - print("processing...") - - func_var = [] - for music in self.data: - if not isinstance(music, musicObj.Music): - print(str(music) + "is not Music object") - return None - var = [self.dirs, music] - func_var.append((var, None)) - - pool = threadpool.ThreadPool(self.procs) - requests = threadpool.makeRequests(download, func_var) - [pool.putRequest(req) for req in requests] - pool.wait() - - return self.dirs - - - - - # pool = multiprocessing.Pool(self.procs) - # for music in self.data: - # if not isinstance(music, musicObj.Music): - # print(str(music) + "is not Music object") - # continue - # pool.apply_async(download, (self.dirs, music)) - # pool.close() - # pool.join() - # print("finish!") - - - - - +import re +import tempfile +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path + +import requests + +from .exceptions import DownloadError, UnavailableError +from .validation import integer + + +def _filename(music, audio_type): + extension = str(audio_type or "").lower() + if not re.fullmatch(r"[a-z0-9]{1,8}", extension): + raise DownloadError("音频文件类型无效") + name = "{} - {}".format(music.name, " ".join(music.artist)) + name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", name).strip(" .") or music.id + reserved = {"CON", "PRN", "AUX", "NUL"} | { + prefix + str(i) for prefix in ("COM", "LPT") for i in range(1, 10) + } + if name.split(".")[0].upper() in reserved: + name = "_" + name + # Bound UTF-8 bytes as well as character length for macOS/Linux filenames. + name = name.encode("utf-8")[:180].decode("utf-8", errors="ignore").rstrip(" .") + return "{}.{}".format(name, extension) + + +def download(dirs, music, *, audio=None): + audio = audio if audio is not None else music._playback.get(music.id) + url = audio.get("url") + if not url: + raise UnavailableError("当前账号或地区无法获取音频:{}".format(music.id)) + if not isinstance(url, str) or not url.startswith(("https://", "http://")): + raise DownloadError("音频地址必须是 HTTP 或 HTTPS") + temporary = None + options = music._options + try: + target_dir = Path(dirs or "cloudmusic").expanduser().resolve() + target = target_dir / _filename(music, audio.get("type")) + target_dir.mkdir(parents=True, exist_ok=True) + # Never forward account cookies to a CDN or to a redirect destination. + with requests.Session() as session: + session.proxies.update(options.get("proxies") or {}) + session.headers["User-Agent"] = "Mozilla/5.0" + with session.get(url, stream=True, timeout=options.get("timeout", (5, 20))) as response: + response.raise_for_status() + content_type = response.headers.get("Content-Type", "").lower() + if "text/" in content_type or "json" in content_type: + raise DownloadError("下载地址返回了错误页面,未覆盖目标文件") + received = 0 + with tempfile.NamedTemporaryFile( + dir=target_dir, prefix=".cloudmusic-", suffix=".part", delete=False + ) as handle: + temporary = Path(handle.name) + for chunk in response.iter_content(chunk_size=64 * 1024): + if chunk: + handle.write(chunk) + received += len(chunk) + length = response.headers.get("Content-Length") + try: + expected = int(length) if length is not None else None + except ValueError: + raise DownloadError("音频响应的 Content-Length 无效") from None + if not received or ( + length and not response.headers.get("Content-Encoding") and received != expected + ): + raise DownloadError("音频下载不完整,未覆盖目标文件") + os.replace(temporary, target) + temporary = None + return str(target) + except requests.RequestException as exc: + raise DownloadError("音频下载失败或超时,未覆盖目标文件") from exc + except OSError as exc: + raise DownloadError("音频文件写入失败,请检查目录权限和剩余空间") from exc + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +class Downloader: + def __init__(self, procs=2, dirs=""): + self.data = [] + self.dirs = dirs + self.procs = integer(procs, "procs", 1) + self.results = [] + self.errors = {} + + def start(self): + from .musicObj import Music + + if any(not isinstance(music, Music) for music in self.data): + raise ValueError("data 中的每一项必须是 Music 对象") + self.results, self.errors = [], {} + with ThreadPoolExecutor(max_workers=min(self.procs, 32)) as pool: + futures = { + pool.submit(music.download, self.dirs): index + for index, music in enumerate(self.data) + } + completed = {} + for future in as_completed(futures): + index = futures[future] + try: + completed[index] = future.result() + except Exception as exc: + self.errors[index] = exc + self.results = [completed[index] for index in sorted(completed)] + if self.errors: + raise DownloadError( + "{} 首歌曲下载失败;详见 loader.errors,成功文件见 loader.results".format( + len(self.errors) + ) + ) + return str(Path(self.dirs or "cloudmusic").expanduser().resolve()) diff --git a/cloudmusic/encrypt.py b/cloudmusic/encrypt.py index 8f014fb..2c3c354 100644 --- a/cloudmusic/encrypt.py +++ b/cloudmusic/encrypt.py @@ -14,11 +14,13 @@ PUBKEY = "010001" NONCE = b"0CoJUm6Qyw8W8jud" -def encrypted_request(text={}, method=''): - # type: (str) -> dict - if method == 'linux': + +def encrypted_request(text=None, method=""): + text = {} if text is None else text + if method == "linux": return linuxEncrypt(text) - text = str(text) + if not isinstance(text, str): + text = json.dumps(text, ensure_ascii=False, separators=(",", ":")) data = text.encode("utf-8") secret = create_key(16) params = aes(aes(data, NONCE), secret) @@ -44,8 +46,5 @@ def create_key(size): return binascii.hexlify(os.urandom(size))[:16] -def linuxEncrypt(text={}): - # print(text) - return text - - +def linuxEncrypt(text=None): + return {} if text is None else text diff --git a/cloudmusic/exceptions.py b/cloudmusic/exceptions.py new file mode 100644 index 0000000..0050362 --- /dev/null +++ b/cloudmusic/exceptions.py @@ -0,0 +1,44 @@ +"""Errors callers can handle without depending on an upstream response shape.""" + + +class CloudMusicError(Exception): + """Base class for service and download errors.""" + + +class RequestError(CloudMusicError): + def __init__(self, message, status_code=None): + self.status_code = status_code + super().__init__(message) + + +class APIError(CloudMusicError): + def __init__(self, code, message=""): + self.code = code + self.message = message or "网易云接口拒绝了请求" + super().__init__("{} (code={})".format(self.message, code)) + + +class ResponseError(CloudMusicError): + """The service returned an unexpected or malformed response.""" + + +class NotFoundError(CloudMusicError): + def __init__(self, ids): + self.ids = list(ids) + super().__init__("未找到歌曲或用户:{}".format(", ".join(map(str, self.ids)))) + + +class UnavailableError(CloudMusicError): + """An audio file is unavailable for the current account or region.""" + + +class IncompleteResultError(CloudMusicError): + def __init__(self, message, partial_result, expected_count=None): + self.partial_result = partial_result + self.expected_count = expected_count + self.actual_count = len(partial_result) + super().__init__(message) + + +class DownloadError(CloudMusicError): + pass diff --git a/cloudmusic/musicObj.py b/cloudmusic/musicObj.py index 65d0c10..066fc69 100644 --- a/cloudmusic/musicObj.py +++ b/cloudmusic/musicObj.py @@ -1,118 +1,197 @@ -from . import download -from . import sessions -from . import api - -import re - - - -def createObj(ids, level): - api = sessions.api.Api() - musicUrl = api.get_song_url(dict(ID = ids, level = level))['data'] - musicInfo = api.get_song_detail(dict(ID = ids))['songs'] - musicOtp = [] - for i in range(len(ids)): - mu = musicUrl[i] - for mi in musicInfo: - if mi['id'] == mu['id']: - name = mi['name'] + " " + mi['alia'][0] if mi['alia'] else mi['name'] - artist = [ar['name'] for ar in mi['ar']] - artistId = [ar['id'] for ar in mi['ar']] - album = mi['al']['name'] - albumId = mi['al']['id'] - picUrl = mi['al']['picUrl'] - info = dict(name = name, artist = artist, album = album, picUrl = picUrl, artistId = artistId, albumId = albumId) - musicInfo.remove(mi) - break - musicOtp.append(Music(mu["id"], mu["url"], mu["level"], mu["size"], mu["type"], info)) - if len(musicUrl) == 1: - return musicOtp[0] - return musicOtp - - # musicOtp = [] - # info = {"name": "","artist": "","album": ""} - # musicOtp = [] - # for d in data: - # # if detail: - # # info = query.getSongInfo(d["id"]) - # # if not info: - # # print("歌曲不存在 id=" + str(d["id"])) - # # continue - # musicOtp.append(Music(d["id"], d["url"], d["level"], d["size"], d["type"], info)) - # if len(data) == 1: - # return musicOtp[0] - +"""Song metadata with lazy, batched audio lookup.""" +import threading +import time +from .api import Api +from .exceptions import APIError, NotFoundError, UnavailableError +from .validation import field -class Music: - def __init__(self, id_, url, level, size, type_, info): - self.url = url - self.id = str(id_) +LEVELS = ("standard", "higher", "exhigh", "lossless") +BATCH_SIZE = 200 + + +def validate_level(level): + if level not in LEVELS: + raise ValueError("level 必须是 {}".format(", ".join(LEVELS))) + return level + + +class _PlaybackBatch: + def __init__(self, ids, level, options, data=None): + self.ids = ids self.level = level - self.size = int(size) - self.type = type_ - self.name = info['name'] - self.artist = info['artist'] - self.album = info['album'] - self.artistId = info['artistId'] - self.albumId = info['albumId'] - self.picUrl = info['picUrl'] - self.para = { - "clas" : "", - "ID" : self.id, - "number" : 15, - "offset" : 0, - } - self.levels = ["standard", "higher", "exhigh", "lossless"] + self.options = options + self._data = data + self._loaded_at = time.monotonic() if data is not None else 0 + self._blocked = None + self._blocked_at = 0 + self._lock = threading.Lock() + + def get(self, song_id): + with self._lock: + if self._blocked and time.monotonic() - self._blocked_at < 60: + raise APIError(*self._blocked) + # Avoid retaining expired CDN URLs in long-running scripts. + if self._data is None or time.monotonic() - self._loaded_at >= 60: + try: + with Api(**self.options) as api: + rows = field( + api.get_song_url({"ID": self.ids, "level": self.level}), "data", list + ) + except APIError as exc: + if exc.code in (-460, -462, 301, 401, 403, 429): + self._blocked = (exc.code, exc.message) + self._blocked_at = time.monotonic() + raise + self._data = {str(field(row, "id", int)): row for row in rows} + self._loaded_at = time.monotonic() + self._blocked = None + return self._data.get(str(song_id), {}) + + +def createObj(ids, level="higher", *, api=None, options=None, details=None): + """Always return a list, in input order, including repeated IDs.""" + validate_level(level) + if not ids: + return [] + if api is None: + with Api(**(options or {})) as owned: + return createObj(ids, level, api=owned, options=owned.options, details=details) + options = options or api.options + known = {field(song, "id", int): song for song in (details or [])} + unique = list(dict.fromkeys(ids)) + missing = [song_id for song_id in unique if song_id not in known] + for offset in range(0, len(missing), BATCH_SIZE): + rows = field( + api.get_song_detail({"ID": missing[offset : offset + BATCH_SIZE]}), "songs", list + ) + known.update({field(song, "id", int): song for song in rows}) + absent = [song_id for song_id in unique if song_id not in known] + if absent: + raise NotFoundError(absent) + songs = {} + for offset in range(0, len(unique), BATCH_SIZE): + batch = unique[offset : offset + BATCH_SIZE] + playback = _PlaybackBatch(batch, level, options) + for song_id in batch: + detail = known[song_id] + album = field(detail, "al", dict) + artists = field(detail, "ar", list) + name = field(detail, "name", str) + aliases = detail.get("alia") or [] + info = { + "name": name + (" " + str(aliases[0]) if aliases else ""), + "artist": [ar.get("name") or "" for ar in artists], + "artistId": [ar.get("id") for ar in artists], + "album": album.get("name") or "", + "albumId": album.get("id"), + "picUrl": album.get("picUrl") or "", + } + songs[song_id] = Music( + song_id, level=level, info=info, options=options, playback=playback + ) + return [songs[song_id] for song_id in ids] + +class Music: + def __init__( + self, + id_, + url=None, + level="higher", + size=0, + type_=None, + info=None, + *, + options=None, + playback=None, + ): + self.id = str(id_) + self._options = options or {} + self._requested_level = level + self._playback = playback or _PlaybackBatch( + [int(id_)], + level, + self._options, + {self.id: {"url": url, "level": level, "size": size, "type": type_}}, + ) + info = info or {} + for key, default in ( + ("name", ""), + ("artist", []), + ("artistId", []), + ("album", ""), + ("albumId", None), + ("picUrl", ""), + ): + setattr(self, key, info.get(key, default)) def __repr__(self): - return "" - - def download(self, dirs="", level="standard"): - if self.type: - if level == "standard": - return download.download(dirs, self) - elif level in self.levels: - return createObj([self.id], level).download() - else: - print("没有这个level, 默认standard") - return download.download(dirs, self) - else: - print("download failed - " + self.id) - return None + return "".format(self.id) + + @property + def url(self): + return self._playback.get(self.id).get("url") + + @property + def size(self): + return int(self._playback.get(self.id).get("size") or 0) + + @property + def type(self): + return self._playback.get(self.id).get("type") + + @property + def level(self): + """Actual audio quality returned by NetEase, which may be lower than requested.""" + return self._playback.get(self.id).get("level") + + @property + def freeTrialInfo(self): + return self._playback.get(self.id).get("freeTrialInfo") + + def download(self, dirs="", level=None): + from .download import download + + selected = validate_level(self._requested_level if level is None else level) + playback = ( + self._playback + if selected == self._requested_level + else _PlaybackBatch([int(self.id)], selected, self._options) + ) + audio = playback.get(self.id) + if not audio.get("url") or not audio.get("type"): + raise UnavailableError("当前账号或地区无法获取这首歌的音频:{}".format(self.id)) + if audio.get("freeTrialInfo"): + raise UnavailableError("接口仅返回试听音频,未下载为完整歌曲:{}".format(self.id)) + return download(dirs, self, audio=audio) + + def _comment(self, clas, number=15): + from .sessions import Session + + with Session(**self._options) as session: + return session.comment({"ID": self.id, "clas": clas, "number": number}) - # 获取评论数量 def getCommentsCount(self): - self.para["clas"] = "count" - with sessions.Session() as session: - return session.comment(self.para) + return self._comment("count") - # 获取热评,上限15 def getHotComments(self, number=15): - self.para["clas"] = "hot" - self.para["number"] = number - with sessions.Session() as session: - return session.comment(self.para) - - # 获取评论,时间顺序,从最近的一直向后 + return self._comment("hot", number) + def getComments(self, number): - self.para["clas"] = "new" - self.para["number"] = str(number) - with sessions.Session() as session: - return session.comment(self.para) + return self._comment("new", number) - # 获取歌词 def getLyrics(self): - lrc = api.Api().get_lyrics(dict(ID = self.id)) - lyric = "" - tlyric = "" - - if "lrc" in lrc: - lyric = lrc["lrc"]["lyric"] - if "lyric" in lrc["tlyric"]: - tlyric = lrc["tlyric"]["lyric"] - - return [lyric, tlyric] - + with Api(**self._options) as api: + data = api.get_lyrics({"ID": self.id}) + output = [] + for key in ("lrc", "tlyric"): + if data.get(key) is None: + output.append("") + else: + lyrics = field(data, key, dict) + output.append( + field(lyrics, "lyric", str) if lyrics.get("lyric") is not None else "" + ) + return output diff --git a/cloudmusic/query.py b/cloudmusic/query.py index 125df53..f6f9c03 100644 --- a/cloudmusic/query.py +++ b/cloudmusic/query.py @@ -1,158 +1,32 @@ -from . import musicObj -from . import api +"""HTTP handling, deliberately independent of individual music endpoints.""" import requests -import json -from lxml import etree -import time -import re +from .exceptions import APIError, RequestError, ResponseError -def post(url, headers, data): - res = requests.post(url=url, headers=headers, data=data) - - if res.status_code == 200: - if not res.text: - return "请求出错" - - if "cloudsearch" in url: - ids = [] - for li in json.loads(res.text)['result']['songs']: - ids.append(li["id"]) - return ids - else: - # print(res.text) - return json.loads(res.text) - else: - return "请求出错" - - -# 老方法获取用户信息 -def getUserInfo(ID): - url='https://music.163.com/user/home?id={}'.format(ID) - session = requests.Session() - session.headers = api.Api().headers - res = session.get(url) - - html = etree.HTML(res.text) - name = html.xpath('//*[@id="j-name-wrap"]/span[1]')[0].text - level = html.xpath('//*[@id="j-name-wrap"]/span[3]')[0].text - sexText = etree.tostring(html.xpath('//*[@id="j-name-wrap"]/i')[0]).decode("utf-8") - if sexText.split('"/>')[-2:] == '01': - sex = "male" - elif sexText.split('"/>')[-2:] == '02': - sex = "female" - else: - sex = None - eventCount = html.xpath('//*[@id="event_count"]')[0].text - followCount = html.xpath('//*[@id="follow_count"]')[0].text - fanCount = html.xpath('//*[@id="fan_count"]')[0].text - musicCount = re.split("歌|首", html.xpath('//*[@id="rHeader"]/h4')[0].text)[1] - createdCount = re.split("(|)", html.xpath('//*[@id="cHeader"]/h3/span/text()[2]')[0])[1] - collectCount = re.split("(|)", html.xpath('//*[@id="sHeader"]/h3/span/text()[2]')[0])[1] - - avatarUrl = html.xpath('//*[@id="ava"]/img')[0].get("src") - - div2 = html.xpath('//*[@id="head-box"]/dd/div[2]') - if div2: - if "个人介绍" in div2[0].text: - print("是个人介绍") - print(div2[0].text) - # if html.xpath('//*[@id="head-box"]/dd/div[3]/span[1]') and not html.xpath('//*[@id="age"]/span'): - # print("有介绍和地区") - # print(html.xpath('//*[@id="head-box"]/dd/div[3]/span[1]')[0].text) - # elif html.xpath('//*[@id="head-box"]/dd/div[3]/span[1]'): - # print("有介绍和年龄") - # print(html.xpath('//*[@id="age"]/span')[0].text) - if html.xpath('//*[@id="head-box"]/dd/div[3]/span[1]') and html.xpath('//*[@id="age"]/span'): - print("三个搜有") - print(etree.tostring(html.xpath('//*[@id="age"]')[0])) - print(html.xpath('//*[@id="head-box"]/dd/div[3]/span[1]')[0].text) - else: - print("是地区") - a = html.xpath('//*[@id="head-box"]/dd/div[2]/span[1]')[0].text - print(a) - - input() - - - location = html.xpath('//*[@id="head-box"]/dd/div[3]/span[1]') - location = location[0].text[5:] if location else None - - introduation = html.xpath('//*[@id="head-box"]/dd/div[2]') - introduation = introduation[0].text[5:] if introduation else None - - age = html.xpath('//*[@id="age"]/span') - age = age[0].text if age else None - print(introduation) - - - print(location) - print(age) - print(musicCount) - - input() - - - title = html.xpath("/html/head/title")[0].text +def post(url, headers, data, *, session=None, timeout=(5, 20)): + if session is None: + with requests.Session() as owned: + return post(url, headers, data, session=owned, timeout=timeout) try: - name = title.split(" -" )[0] - artist = title.split(" - ")[1] - except: - return "" - album = html.xpath("/html/body/div[3]/div[1]/div/div/div[1]/div[1]/div[2]/p[2]/a")[0].text - info = { - "name" : name, - "artist" : artist, - "album" : album, - } - - return info - - -# 老方法获取歌曲详细信息 -def getSongInfo(id_): - headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36', - 'Referer': 'http://music.163.com/' - } - url='http://music.163.com/song?id=' + str(id_) - session = requests.Session() - session.headers=headers - res = session.get(url) - - html = etree.HTML(res.text) - title = html.xpath("/html/head/title")[0].text - try: - name = title.split(" -" )[0] - artist = title.split(" - ")[1] - except: - return "" - album = html.xpath("/html/body/div[3]/div[1]/div/div/div[1]/div[1]/div[2]/p[2]/a")[0].text - info = { - "name" : name, - "artist" : artist, - "album" : album, - } - print("creep --- " + str(id_)) - return info - - -# 老方法获取歌单信息 -def getPlayList(id_): - headers = { - 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36', - 'Referer': 'http://music.163.com/' - } - url='http://music.163.com/playlist?id=' + str(id_) - session = requests.Session() - session.headers=headers - res = session.get(url) - - html = etree.HTML(res.text) - lists = html.xpath('//ul[@class="f-hide"]/li/a/@href') - - musicList = map(lambda x: x.split("=")[1], lists) - - return list(musicList) + with session.post(url, headers=headers, data=data, timeout=timeout) as response: + if response.status_code != 200: + raise RequestError( + "网易云 HTTP 请求失败 ({})".format(response.status_code), + status_code=response.status_code, + ) + try: + result = response.json() + except ValueError: + raise ResponseError("网易云返回了无效的 JSON") from None + except requests.RequestException as exc: + raise RequestError("网易云网络请求失败或超时") from exc + if not isinstance(result, dict): + raise ResponseError("网易云响应不是 JSON 对象") + if "code" not in result: + raise ResponseError("网易云响应缺少业务状态码") + if result["code"] not in (200, "200"): + message = result.get("message") or result.get("msg") or "" + raise APIError(result["code"], str(message)) + return result diff --git a/cloudmusic/sessions.py b/cloudmusic/sessions.py index aa7ae92..decdd74 100644 --- a/cloudmusic/sessions.py +++ b/cloudmusic/sessions.py @@ -1,157 +1,178 @@ -from . import encrypt -from . import query -from . import download -from . import api -from . import musicObj -from . import userObj +"""Reusable public client; returned objects retain configuration, not open sockets.""" +from .api import Api +from .exceptions import IncompleteResultError, ResponseError +from .musicObj import createObj, validate_level +from .validation import field, ids, integer -class Session(): - def __init__(self): - - self.error = "错误参数 --- 运行cloudmusic.help()查看支持" - - self.api = api.Api() - - # 默认音质 - self.level = "higher" - +class Session: + def __init__(self, *, cookie=None, timeout=(5, 20), proxies=None, level="higher"): + self.level = validate_level(level) + self.api = Api(cookie=cookie, timeout=timeout, proxies=proxies) def __enter__(self): return self - def __exit__(self, *args): - pass - + self.close() + + def close(self): + self.api.close() + + def _music(self, song_ids, details=None): + return createObj( + song_ids, self.level, api=self.api, options=self.api.options, details=details + ) + + def getMusic(self, para): + songs = self._music(ids(para)) + return songs if isinstance(para, (list, tuple)) else songs[0] + + def getPlaylist(self, para): + playlist_id = integer(para, "id", 1) + data = field(self.api.get_playlist({"ID": playlist_id}), "playlist", dict) + tracks = field(data, "tracks", list) + track_ids = field(data, "trackIds", list) + song_ids = [field(track, "id", int) for track in track_ids] + result = self._music(song_ids, details=tracks) + expected = field(data, "trackCount", int) + if len(song_ids) != expected: + raise IncompleteResultError( + "网易云只返回了歌单的部分歌曲;可能需要登录或访问权限", + result, + expected, + ) + return result + + def getAlbum(self, para): + album_id = integer(para, "id", 1) + songs = field(self.api.get_album({"ID": album_id}), "songs", list) + return self._music([field(song, "id", int) for song in songs], details=songs) + + def getUser(self, para): + from .userObj import User + + user_id = integer(para, "id", 1) + return User(self.api.get_userInfo({"ID": user_id}), options=self.api.options) def request(self, clas, para): - # 先判断数据是否合法 - if isinstance(para, list): - try: - ids = [int(p) for p in para] - except: - return self.error - else: - try: - ids = [int(para)] - except: - return self.error - - - if clas == "song": - return musicObj.createObj(ids, self.level) - - # 歌单 - elif clas == "playlist" : - playlist = self.api.get_playlist(dict(ID = ids[0], method = "linux"))["playlist"]["tracks"] - ids = [ml["id"] for ml in playlist] - musicList = self.request("song", ids) - musicListc = [] - for ID in ids: - for music in musicList: - if music.id == str(ID): - musicListc.append(music) - musicList.remove(music) - continue - return musicListc - - # 专辑 - elif clas == "album": - playlist = self.api.get_album(dict(ID = ids[0]))["songs"] - ids = [ml["id"] for ml in playlist] - musicList = self.request("song", ids) - musicList = [musicList] if len(ids) == 1 else musicList - musicListc = [] - for ID in ids: - for music in musicList: - if music.id == str(ID): - musicListc.append(music) - musicList.remove(music) - continue - return musicListc - - # 用户 - elif clas == "user": - return userObj.createUser(ids[0]) - - else : - return self.error - - - # 搜索 - def search(self, content, number): - para = { - "string" : str(content), - "number" : str(number) - } - ids = self.api.search(para) - - musicList = self.request("song", ids) - musicList = [musicList] if not isinstance(musicList, list) else musicList - musicListc = [] - for ID in ids: - for music in musicList: - if music.id == str(ID): - musicListc.append(music) - musicList.remove(music) - continue - return musicListc - - - # 评论 - def comment(self, mpara): - para = { - "ID" : mpara["ID"], - "offset" : "0", - "total" : "true", - "limit" : "20" + methods = { + "song": self.getMusic, + "playlist": self.getPlaylist, + "album": self.getAlbum, + "user": self.getUser, } - if mpara["clas"] == "count": - return self.api.get_commets(para)["total"] - if mpara["clas"] == "hot": - comments = self.api.get_commets(para)["hotComments"] - return self.datalizeComment(comments, mpara["number"]) - if mpara["clas"] == "new": - offset = 0 - number = int(mpara["number"]) - comments = [] - for _ in range(int(number / 20)): - para["offset"] = str(offset) - para["limit"] = str(offset + 20) - comments.extend(self.api.get_commets(para)["comments"]) - offset += 20 - para["offset"] = str(offset) - para["limit"] = str(number - int(number / 20) * 20) - comments.extend(self.api.get_commets(para)["comments"]) - return self.datalizeComment(comments, mpara["number"]) - - - # 评论格式化 - def datalizeComment(self, comments, number): - number = len(comments) if int(number) > len(comments) else int(number) - comList = [] - for com in comments[0:number]: - comDic = dict( likeCount = com['likedCount'], - content = com['content'], - time = com['time'], - nickName = com['user']['nickname'], - userId = str(com['user']['userId']), - avatarUrl = com['user']['avatarUrl'], - vipType = com['user']['vipType'], - userType = com['user']['userType'] - ) - comList.append(comDic) - return comList - - - # 多进程下载 - def downloader(self, procs, dirs): - return download.Downloader(procs, dirs) - - - - - + if clas not in methods: + raise ValueError("未知请求类型:{}".format(clas)) + return methods[clas](para) + + def search(self, content, number=5): + if not isinstance(content, str) or not content.strip(): + raise ValueError("搜索关键词不能为空") + number = integer(number, "number") + song_ids, seen, offset = [], set(), 0 + while len(song_ids) < number: + limit = min(100, number) + result = field( + self.api.search({"string": content, "number": limit, "offset": offset}), + "result", + dict, + ) + if "songs" not in result and result.get("songCount") != 0: + raise ResponseError("搜索响应缺少歌曲列表或零结果计数") + page = result.get("songs", []) + if not isinstance(page, list): + raise ResponseError("搜索结果中的 songs 格式无效") + added = 0 + for song in page: + song_id = field(song, "id", int) + if song_id not in seen: + song_ids.append(song_id) + seen.add(song_id) + added += 1 + offset += len(page) + more = ( + field(result, "hasMore", bool) + if "hasMore" in result + else (offset < field(result, "songCount", int)) + ) + if not more: + break + if added == 0: + raise IncompleteResultError( + "搜索分页没有继续返回新歌曲", self._music(song_ids), number + ) + return self._music(song_ids[:number]) + + def comment(self, para): + song_id = str(integer(para["ID"], "id", 1)) + clas = para["clas"] + if clas == "count": + return field(self.api.get_commets({"ID": song_id, "limit": 20}), "total", int) + number = integer(para["number"], "number") + if not number: + return [] + if clas == "hot": + data = self.api.get_commets({"ID": song_id, "limit": 20}) + return self.datalizeComment(field(data, "hotComments", list), min(number, 15)) + if clas != "new": + raise ValueError("未知评论类型") + comments, seen, cursors = [], set(), {"0"} + cursor, page_number = "0", 1 + while len(comments) < number: + data = field( + self.api.get_comment_page( + { + "ID": song_id, + "page": page_number, + "limit": min(20, number), + "cursor": cursor, + } + ), + "data", + dict, + ) + page = field(data, "comments", list) + previous_count = len(comments) + for comment in page: + comment_id = field(comment, "commentId", int) + if comment_id not in seen: + seen.add(comment_id) + comments.append(comment) + if len(comments) >= number or not field(data, "hasMore", bool): + break + next_cursor = str(data.get("cursor") or "") + if not next_cursor or next_cursor in cursors or len(comments) == previous_count: + raise IncompleteResultError( + "评论分页未继续前进,已停止以避免重复请求", + self.datalizeComment(comments, number), + number, + ) + cursors.add(next_cursor) + cursor, page_number = next_cursor, page_number + 1 + return self.datalizeComment(comments, number) + def datalizeComment(self, comments, number): + output = [] + for comment in comments[:number]: + user = comment.get("user") or {} + output.append( + { + "commentId": comment.get("commentId"), + "likeCount": comment.get("likedCount", 0), + "content": comment.get("content") or "", + "time": comment.get("time"), + "nickName": user.get("nickname") or "", + "userId": str(user.get("userId", "")), + "avatarUrl": user.get("avatarUrl") or "", + "vipType": user.get("vipType", 0), + "userType": user.get("userType", 0), + } + ) + return output + + def downloader(self, procs=2, dirs=""): + from .download import Downloader + + return Downloader(procs, dirs) diff --git a/cloudmusic/userObj.py b/cloudmusic/userObj.py index cf4713d..bc86399 100644 --- a/cloudmusic/userObj.py +++ b/cloudmusic/userObj.py @@ -1,74 +1,91 @@ -from . import query -from . import api -from . import sessions +"""User metadata and read-only playlist/listening-history access.""" +from .api import Api +from .exceptions import IncompleteResultError +from .musicObj import createObj +from .validation import field, integer -def createUser(ID): - info = api.Api().get_userInfo(dict(ID = ID)) - return User(info) +def createUser(ID, **options): + with Api(**options) as api: + return User(api.get_userInfo({"ID": integer(ID, "id", 1)}), options=api.options) -class User: - def __init__(self, info): - self.id = str(info['userPoint']['userId']) - self.level = info['level'] - self.sex = info['profile']['gender'] - self.listenSongs = info['listenSongs'] - self.createTime = info['profile']['createTime'] - self.nickname = info['profile']['nickname'] - self.avatarUrl = info['profile']['avatarUrl'] - self.city = info['profile']['city'] - self.province = info['profile']['province'] - self.vipType = info['profile']['vipType'] - self.birthday = info['profile']['birthday'] - self.signature = info['profile']['signature'] - self.fans = info['profile']['followeds'] - self.follows = info['profile']['follows'] - self.eventCount = info['profile']['eventCount'] - self.playlistCount = info['profile']['playlistCount'] +class User: + def __init__(self, info, *, options=None): + self._options = options or {} + profile = field(info, "profile", dict) + self.id = str(field(profile, "userId", int)) + self.nickname = profile.get("nickname") or "" + self.nickName = self.nickname + self.sex = profile.get("gender", 0) + self.level = info.get("level", 0) + self.listenSongs = info.get("listenSongs", 0) + for key in ( + "createTime", + "avatarUrl", + "city", + "province", + "vipType", + "birthday", + "signature", + "follows", + "eventCount", + "playlistCount", + ): + setattr(self, key, profile.get(key)) + self.fans = profile.get("followeds", 0) def getPlaylist(self): - info = api.Api().get_userPlayerlist(dict(ID = self.id))['playlist'] - outp = [] - for pl in info: - playlist = dict( - id = pl['id'], - creatorId = pl['userId'], - playCount = pl['playCount'], - createTime = pl['createTime'], - coverImgUrl = pl['coverImgUrl'], - name = pl['name'], - updateTime = pl['updateTime'], - tags = pl['tags'] - ) - outp.append(playlist) - return outp - + result, seen, offset = [], set(), 0 + with Api(**self._options) as api: + while True: + data = api.get_userPlayerlist({"ID": self.id, "offset": offset, "limit": 100}) + page = field(data, "playlist", list) + added = 0 + for playlist in page: + playlist_id = field(playlist, "id", int) + if playlist_id in seen: + continue + seen.add(playlist_id) + result.append( + { + "id": playlist_id, + "creatorId": playlist.get("userId"), + **{ + key: playlist.get(key) + for key in ( + "playCount", + "createTime", + "coverImgUrl", + "name", + "updateTime", + "tags", + ) + }, + } + ) + added += 1 + if not data.get("more", False): + return result + if not added: + raise IncompleteResultError("用户歌单分页未继续前进", result) + offset += len(page) def getRecord(self, recordType=0): - info = api.Api().get_userRecord(dict(ID = self.id)) - if info['weekData']: - weekIds = [wd['song']['id'] for wd in info['weekData']] - weekObj = sessions.Session().request("song", weekIds) - weekOtp = [] - for wi in range(len(weekIds)): - weekOtp.append(dict(score = info['weekData'][wi]['score'], music = weekObj[wi])) - else: - weekOtp = [] - - if info['allData']: - allIds = [ad['song']['id'] for ad in info['allData']] - allObj = sessions.Session().request("song", allIds) - allOtp = [] - for ai in range(len(allIds)): - allOtp.append(dict(score = info['allData'][ai]['score'], music = allObj[ai])) - else: - allOtp = [] - - if recordType: - return weekOtp - else: - return allOtp - - + record_type = 1 if integer(recordType, "recordType") else 0 + with Api(**self._options) as api: + data = api.get_userRecord({"ID": self.id, "type": record_type}) + rows = field(data, "weekData" if record_type else "allData", list) + details = [field(row, "song", dict) for row in rows] + songs = createObj( + [field(song, "id", int) for song in details], + api=api, + options=api.options, + details=details, + ) + by_id = {song.id: song for song in songs} + return [ + {"score": field(row, "score", int), "music": by_id[str(row["song"]["id"])]} + for row in rows + ] diff --git a/cloudmusic/validation.py b/cloudmusic/validation.py new file mode 100644 index 0000000..ce84382 --- /dev/null +++ b/cloudmusic/validation.py @@ -0,0 +1,26 @@ +"""Shared helpers for public input and upstream response validation.""" + +from .exceptions import ResponseError + + +def integer(value, name, minimum=0): + if isinstance(value, bool) or not isinstance(value, (int, str)): + raise ValueError("{} 必须是整数".format(name)) + try: + result = int(value) + except (ValueError, TypeError): + raise ValueError("{} 必须是整数".format(name)) from None + if result < minimum: + raise ValueError("{} 必须大于或等于 {}".format(name, minimum)) + return result + + +def ids(value): + values = value if isinstance(value, (list, tuple)) else [value] + return [integer(item, "id", 1) for item in values] + + +def field(data, key, expected_type): + if not isinstance(data, dict) or not isinstance(data.get(key), expected_type): + raise ResponseError("网易云响应缺少有效的 {} 字段".format(key)) + return data[key] diff --git a/docs/verification.md b/docs/verification.md new file mode 100644 index 0000000..3aa988e --- /dev/null +++ b/docs/verification.md @@ -0,0 +1,58 @@ +# 0.2.0 验收记录 + +日期:2026-09-29。基线:主分支 0.1.1,`7bb714b`。本次没有发布到 PyPI。 + +## 已修复并由自动化测试覆盖 + +- 单个 ID / 单元素列表 / 空列表的返回类型、输入顺序、重复 ID、缺失歌曲处理。 +- 歌单按 trackIds 补齐详情,检测服务返回的数量是否完整;复用已有元数据。 +- 搜索参数转义、标准 JSON 加密、分页、去重和无结果。 +- 评论改用时间游标,处理短页、重叠页、重复游标和没有进展的响应;100 条结果不重复。 +- 歌词缺少翻译、用户资料缺少可选字段、听歌记录按 ID 对齐、用户歌单分页。 +- 歌曲信息与音频权限分离;音频批量延迟加载、过期刷新、并发共用一次请求;同批验证失败暂停重复请求。 +- 连接复用、明确超时、HTTP/业务/响应结构错误分类;显式 cookie 配置和服务端会话保留。 +- 跨平台文件名、指定目录、音质切换目录传递、流式下载、原子替换和批量错误报告。 +- 本地 HTTP 服务返回正常音频、503、HTML、空响应和中途中断:成功文件内容完整;失败保留旧文件、不遗留临时文件;目录无权限和文件替换失败也返回明确错误;CDN 请求不携带账号 cookie。 + +## 本地检查 + +使用隔离虚拟环境。Python 3.14.6 当前已通过 98 项离线/本地 HTTP 测试,语句与分支合计覆盖率约 99%。6 项外部接口测试默认明确跳过,不计入通过数。Python 3.12.12 的最终 98 项用例也已通过。Ruff 检查通过。已成功构建 wheel 和 sdist,并在没有 lxml/threadpool 的全新环境中安装 wheel、从仓库外导入成功;sdist 已检查包含测试 fixture 和验收文档。 + +```bash +python -m pytest --cov=cloudmusic --cov-branch --cov-report=term-missing +python -m ruff check cloudmusic tests +python -m build +``` + +CI 新增 Linux / macOS / Windows × Python 3.10 / 3.12 / 3.14 共 9 组测试,以及独立安装构建 wheel 的检查。最终 CI 结果以本次 PR 的检查记录为准。 + +## 实际网易云接口 + +本轮采用当前机器的匿名网络环境;没有使用仓库历史内置登录凭据,没有尝试绕过验证。 + +| 能力 | 已观察到的结果 | 验收边界 | +| --- | --- | --- | +| 歌曲详情 | 修正序列化和参数后成功获取示例歌曲信息 | 后续相同查询也观察到 -462,不能声称匿名访问持续稳定 | +| 搜索 | 更新接口后成功返回 5 条结果 | 已通过一次正式联网测试 | +| 歌单 | 示例歌单返回完整 123 首,而旧版只有 1 首详情 | 已通过正式联网测试;后续下载探针中的歌单请求也出现 -462 | +| 歌词 | 成功返回歌词 | 已通过正式联网测试 | +| 专辑 | 正式联网测试返回 -462 | 未通过实网验收 | +| 评论 | 调研探针中 v1 热评及 v2 游标接口成功;正式测试阶段出现 -462 | 100 条去重、顺序与结束条件已离线验证,未完成稳定实网验收 | +| 音频地址 | 观察到成功业务响应但 url 为空,以及 -462 | 没有完整音频可用性保证;没有完成真实网易云文件下载 | +| 用户资料、歌单、记录 | 匿名探针返回 -462 | 逻辑已离线覆盖,登录态未提供,未完成实网验收 | + +一次完整联网测试为 4 项通过、2 项失败(专辑、评论,均为上游 -462);另一次在歌曲详情 fixture 阶段被 -462 拒绝。后续音频下载探针也被拒绝。失败保留为失败,没有转为 skip 或算作成功。参数修正和外部验证限制可能同时存在,不能以一次成功/失败证明唯一原因。 + +因此,本次可确认本地逻辑、接口错误处理和已成功返回的查询链路得到了修复;不能承诺所有地区、匿名会话和用户权限下均能取得全部数据或音频。登录态验收需要有权限的测试账号会话:显式通过 `--cookie-env` 指定本机环境变量,勿把 cookie 发到 issue、PR 或日志。 + +```bash +python -m pytest --live -m live -v --tb=short +python -m pytest --live --cookie-env NETEASE_COOKIE -m live -v --tb=short +``` + +## 发布与兼容性 + +- 保留主要公共函数、对象属性;修正返回类型和异常语义,见 README 的迁移说明。 +- 删除未被当前接口调用的 HTML 抓取实现及其 lxml 依赖;线程池改用标准库。 +- 源码版本更新到 0.2.0;旧的直接上传脚本移除。本轮只构建和验证发行包,不上传 PyPI。 +- 不自动合并历史 PR。参考并感谢 #14、#17、#21 的下载路径与音质目录修复思路;标签功能没有混入这次可靠性修复。 diff --git a/index.html b/index.html index a681642..501bcda 100644 --- a/index.html +++ b/index.html @@ -1,500 +1,9 @@ - - README - - - - - - - - - - - - - - - - - - -
-

cloudmusic: 网易云爬虫解决方案

-

一个功能强大的python库,使用最简单易用的方法获取一切你想要的信息。
-支持python3以上版本

-

安装

-
pip install cloudmusic
-

简单示例

-
    -
  • 通过歌曲id获取Music对象,并展示部分属性
  • -
-
music = cloudmusic.getMusic(1347630432)
-
-print("歌名:{}".format(music.name))
-print("歌手:{}".format(music.artist))
-print("音频文件url:{}".format(music.url))
-
-# >>>
-# 歌名:白日 日剧《冤罪律师》主题曲
-# 歌手:['King Gnu']
-# 音频文件url:http://m10.music.126.net/.../.mp3
-
-
    -
  • 通过歌单id获取Music对象列表,并下载整个歌单的无损品质音频
  • -
-
playlist = cloudmusic.getPlaylist(310729011)
-
-for music in playlist:
-    music.download(level = "lossless")
-
-
    -
  • 获取一首歌的热评并展示部分信息
  • -
-
music = cloudmusic.getMusic(1347630432)
-
-coms = music.getHotComments()
-
-for com in coms:
-    print("发布者:{}".format(com['nickName']))
-    print("内容:{}".format(com['content']))
-    print("获赞数:{}".format(com['likeCount']))
-    print("------------")
-
-# >>>
-# 发布者:Akikonknk
-# 内容:特别喜欢日剧的一个原因 大概就是它帮你认清生活的真相之后 依然教会你如何热爱生活
-# 获赞数:19451
-# ------------
-# 发布者:我藏起来啦嘻嘻
-# 内容:最后一集的犯人真的是应证了那一句话“雪崩的时候没有一片雪花是无辜的”
-# 获赞数:6690
-# ------------
-# 发布者:Usio叔叔
-# ...
-
-
    -
  • 通过关键字搜索获取前100首歌曲的Music对象,并输出每首歌的评论数量
  • -
-
results = cloudmusic.search("白日", 100)
-
-for music in results:
-    print(music.getCommentsCount())
-
-# 5745
-# 77
-# 181
-# 560
-# 368
-# ...
-
-

-

参考文档

-

Music对象

- -

1.Music对象属性

- -
    -
  • url:歌曲音频文件链接
  • -
  • id:歌曲id
  • -
  • name:歌曲名称
  • -
  • rtist:歌手名称
  • -
  • artistId: 歌手id
  • -
  • album:专辑名称
  • -
  • albumId: 专辑id
  • -
  • size:音频文件大小
  • -
  • type:音频文件类型(mp3或m4a)
  • -
  • level:歌曲品质。默认higher
  • -
  • picUrl: 专辑图url
  • -
-

2.Music对象方法

- -
    -
  • -

    download(dirs, level)

    -
    -

    dirs:可选。下载保持路径。默认为当前文件夹内创建的新的cloudmusic文件夹。
    -level:可选,字符型。默认higher。下载品质,有且只有四种选择:standard,higher,exhigh,lossless。

    -
    -

    下载歌曲,返回值为下载绝对路径

    -
  • -
  • -

    getHotComments(number)

    -
    -

    number:可选,整型,默认为15。希望获取的评论个数。上限为15个。

    -
    -

    获取热评。返回值为列表,列表元素为字典,字典内容:
    -"likeCount":获赞数
    -"content":评论内容
    -"time":评论时间
    -"userId":用户id
    -"nickNamd":用户昵称
    -"avatarUrl":用户头像url
    -"vipType":0表示未开通vip,10表示开通音乐包,11表示开通黑胶
    -"userType":0表示普通用户,非0表示特殊用户(明星,丁磊,网红,小秘书等)

    -
  • -
  • -

    getComments(number)

    -
    -

    number:整型,评论个数。数量无限制。

    -
    -

    获取最新的评论。返回值与getHotReview相同。

    -
  • -
  • -

    getLyrics()

    -

    获取歌词。返回值为列表。元素1为原始歌词,元素2为翻译歌词。

    -
  • -
  • -

    getCommentsCount()

    -

    返回值为整型,本首歌的评论数量。

    -
  • -
-

3.获取Music对象函数

- -
    -
  • -

    getMusic(id/id_list)

    -
    -

    id/id_list: 必须,整型、字符型或列表。歌曲id或由歌曲id组成的列表。

    -
    -

    通过歌曲id或一个由id组成的列表生成music对象。当参数为单个id时返回值为单个music对象,当参数为id列表时返回值为music对象组成的列表。

    -
  • -
  • -

    getPlaylist(id)

    -
    -

    id:必须,整型或字符型。歌单id。

    -
    -

    通过歌单id生成music对象,参数为歌单id,返回值为歌单内所有歌曲对应music对象组成的列表。

    -
  • -
  • -

    search(content, number)

    -
    -

    content:必须,字符型。搜索关键字。
    -number:可选,整形,搜索结果个数,默认为5。

    -
    -

    通过关键词搜索获取music对象。返回值为music对象组成的列表。

    -
  • -
  • -

    getAlbum(id)

    -
    -

    id: 必须,整型或字符型。专辑id。

    -
    -

    和getPlaylist一样。

    -
  • -
-
-

User对象

- -

1.User对象属性

- -
    -
  • id:用户id
  • -
  • level:用户等级
  • -
  • listenSongs:累计听歌数量
  • -
  • createTime:账号创建时间
  • -
  • nickName:用户昵称
  • -
  • avatarUrl:头像url
  • -
  • city:所在城市的行政区划代码
  • -
  • province:所在省份的行政区划代码
  • -
  • vipType:vip类型。0表示未开通vip,10表示开通音乐包,11表示开通黑胶
  • -
  • birthday:生日时间戳
  • -
  • signature:个性签名
  • -
  • fans:粉丝数量
  • -
  • follows:关注的用户数量
  • -
  • eventCount:动态数量
  • -
  • playlistCount:创建的歌单数量
  • -
-

2.User对象方法

- -
    -
  • -

    getPlaylist()

    -

    获取用户创建和收藏的全部歌单,返回值为一个列表,每个元素是一个歌单信息组成的字典。字典内容:
    -id:歌单id;
    -name:歌单名称;
    -creatorId:创建者id;
    -createTime:创建时间;
    -coverImgUrl:歌单封面图片url;
    -updateTime:歌单最后一次更新时间;
    -tags:歌单所属音乐风格,列表形式

    -
  • -
  • -

    getRecord(recordType)

    -
    -

    recordType: 可选。默认为0。为0返回所有时间排行,非0返回最近一周排行。

    -
    -

    获取用户听歌排行,返回一个列表,元素为字典。字典内容:
    -score:分值。100表示听歌次数最多,越小越少;
    -music:Music对象

    -
  • -
-

3.获取User对象函数

- -
    -
  • -

    getUser(id)

    -
    -

    id:必须,用户id。

    -
    -

    通过用户id获取User对象,返回值为User对象。

    -
  • -
-
-

@p697

- -
- - - - - - - - - - - - \ No newline at end of file + + +cloudmusic 文档 + +

cloudmusic 文档

+

项目已恢复维护。原静态 HTML/PDF 文档对应 0.1.1,当前接口、安装方法和迁移说明请查看仓库文档。

+

阅读当前 README · 查看验收记录和已知限制

+ + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..06b275b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,46 @@ +[build-system] +requires = ["setuptools>=77"] +build-backend = "setuptools.build_meta" + +[project] +name = "cloudmusic" +dynamic = ["version"] +description = "A small Python client for NetEase Cloud Music" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +authors = [{name = "p697"}] +dependencies = ["requests>=2.32.4,<3", "pycryptodomex>=3.23,<4"] +classifiers = [ + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", +] + +[project.urls] +Homepage = "https://github.com/p697/cloudmusic" +Issues = "https://github.com/p697/cloudmusic/issues" + +[project.optional-dependencies] +test = ["pytest>=8,<10", "pytest-cov>=5,<8", "build>=1.2,<2", "ruff>=0.11,<1"] + +[tool.setuptools.dynamic] +version = {attr = "cloudmusic.__version__.__version__"} + +[tool.setuptools.packages.find] +include = ["cloudmusic*"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra" +markers = ["live: opt-in read-only checks against NetEase (requires --live)"] + +[tool.ruff] +target-version = "py310" +line-length = 100 + +[tool.ruff.lint] +select = ["E4", "E7", "E9", "F"] diff --git a/setup.py b/setup.py index d492635..18b7209 100644 --- a/setup.py +++ b/setup.py @@ -1,131 +1,4 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- +"""Compatibility entry point; package metadata lives in pyproject.toml.""" +from setuptools import setup -# Note: To use the 'upload' functionality of this file, you must: -# $ pipenv install twine --dev - -import io -import os -import sys -from shutil import rmtree - -from setuptools import find_packages, setup, Command - -# Package meta-data. -NAME = 'cloudmusic' -DESCRIPTION = '网易云爬虫解决方案' -URL = 'https://github.com/p697/cloudmusic' -EMAIL = '254139147@qq.com' -AUTHOR = 'p697' -REQUIRES_PYTHON = '>=3.0.0' -VERSION = '0.1.1' - -# What packages are required for this module to be executed? -REQUIRED = [ - 'requests', 'lxml', 'pycryptodomex', 'threadpool' -] - -# What packages are optional? -EXTRAS = { - # 'fancy feature': ['django'], -} - -# The rest you shouldn't have to touch too much :) -# ------------------------------------------------ -# Except, perhaps the License and Trove Classifiers! -# If you do change the License, remember to change the Trove Classifier for that! - -here = os.path.abspath(os.path.dirname(__file__)) - -# Import the README and use it as the long-description. -# Note: this will only work if 'README.md' is present in your MANIFEST.in file! -try: - with io.open(os.path.join(here, 'README.md'), encoding='utf-8') as f: - long_description = '\n' + f.read() -except FileNotFoundError: - long_description = DESCRIPTION - -# Load the package's __version__.py module as a dictionary. -about = {} -if not VERSION: - project_slug = NAME.lower().replace("-", "_").replace(" ", "_") - with open(os.path.join(here, project_slug, '__version__.py')) as f: - exec(f.read(), about) -else: - about['__version__'] = VERSION - - -class UploadCommand(Command): - """Support setup.py upload.""" - - description = 'Build and publish the package.' - user_options = [] - - @staticmethod - def status(s): - """Prints things in bold.""" - print('\033[1m{0}\033[0m'.format(s)) - - def initialize_options(self): - pass - - def finalize_options(self): - pass - - def run(self): - try: - self.status('Removing previous builds…') - rmtree(os.path.join(here, 'dist')) - except OSError: - pass - - self.status('Building Source and Wheel (universal) distribution…') - os.system('{0} setup.py sdist bdist_wheel --universal'.format(sys.executable)) - - self.status('Uploading the package to PyPI via Twine…') - os.system('twine upload dist/*') - - self.status('Pushing git tags…') - os.system('git tag v{0}'.format(about['__version__'])) - os.system('git push --tags') - - sys.exit() - - -# Where the magic happens: -setup( - name=NAME, - version=about['__version__'], - description=DESCRIPTION, - long_description=long_description, - long_description_content_type='text/markdown', - author=AUTHOR, - author_email=EMAIL, - python_requires=REQUIRES_PYTHON, - url=URL, - packages=find_packages(exclude=["tests", "*.tests", "*.tests.*", "tests.*"]), - # If your package is a single module, use this instead of 'packages': - # py_modules=['mypackage'], - - # entry_points={ - # 'console_scripts': ['mycli=mymodule:cli'], - # }, - install_requires=REQUIRED, - extras_require=EXTRAS, - include_package_data=True, - license='MIT', - classifiers=[ - # Trove classifiers - # Full list: https://pypi.python.org/pypi?%3Aaction=list_classifiers - 'License :: OSI Approved :: MIT License', - 'Programming Language :: Python', - 'Programming Language :: Python :: 3', - 'Programming Language :: Python :: 3.6', - 'Programming Language :: Python :: Implementation :: CPython', - 'Programming Language :: Python :: Implementation :: PyPy' - ], - # $ setup.py publish support. - cmdclass={ - 'upload': UploadCommand, - }, -) +setup() diff --git a/test.py b/test.py index f1a1b65..37e353c 100644 --- a/test.py +++ b/test.py @@ -1,53 +1,11 @@ -import cloudmusic -import time -import re -import urllib.request - -# playlist = api.getSong(445702291) -# 601776277 是我的一个歌单id -# print(playlist) -# [, , , , , , , , , , , , , , ] -# 每个Music object包含属性:url(音频文件链接),level(品质),tyoe(格式),id(歌曲id),name(歌曲名),aitist(歌手),album(所属专辑),size(音频文件大小) -# 每个Music object包含方法:1.download()下载到本地 2.review()获取详细评论 - - -# music = cloudmusic.getMusic(1381755293) - -# print(music.getCommentCount()) - -# comments = music.getHotComments() - -# for com in comments: -# print(com['content']) -# print(com['likeCount']) -musiclist = cloudmusic.getPlaylist(362246091) - -# 循环输出每个Music对象的评论数 -for music in musiclist: - print(music.name) - print(music.getCommentsCount()) - -print(len(musiclist)) - - - - - - -# loader = api.createLoader() - -# loader.data = musicList - -# if __name__ == "__main__": -# loader.start() - - - - - - - - +"""Small manual example. Automated tests live in tests/ and never run this file.""" +import cloudmusic +if __name__ == "__main__": + try: + for music in cloudmusic.search("白日", 5): + print(music.id, music.name) + except cloudmusic.CloudMusicError as error: + print(error) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..b0fc7c2 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,78 @@ +import copy + +import pytest +import requests + +from cloudmusic.api import Api + + +def pytest_addoption(parser): + parser.addoption("--live", action="store_true", help="Run opt-in NetEase network checks") + parser.addoption( + "--cookie-env", + default=None, + help="Explicit environment variable containing a cookie for live tests", + ) + + +def pytest_collection_modifyitems(config, items): + if not config.getoption("--live"): + for item in items: + if "live" in item.keywords: + item.add_marker( + pytest.mark.skip(reason="Pass --live to enable real network checks") + ) + + +@pytest.fixture(autouse=True) +def block_external_network(monkeypatch, request): + if "live" in request.keywords: + return + original = requests.Session.request + + def guarded(self, method, url, *args, **kwargs): + if url.startswith("http://127.0.0.1:"): + return original(self, method, url, *args, **kwargs) + pytest.fail("Unexpected external request: {} {}".format(method, url)) + + monkeypatch.setattr(requests.Session, "request", guarded) + + +def song(song_id): + return { + "id": song_id, + "name": "歌曲{}".format(song_id), + "alia": [], + "ar": [{"id": 10, "name": "歌手"}], + "al": {"id": 20, "name": "专辑", "picUrl": "https://example.invalid/cover"}, + } + + +def audio(song_id, **overrides): + return { + "id": song_id, + "url": "https://example.invalid/audio", + "level": "higher", + "size": 4, + "type": "mp3", + "code": 200, + **overrides, + } + + +@pytest.fixture +def api_send(monkeypatch): + routes, calls = {}, [] + + def send(self, url, param=None, method=""): + params = copy.deepcopy(param or {}) + calls.append((url, params)) + value = routes[url] + if callable(value): + return value(params) + if isinstance(value, Exception): + raise value + return copy.deepcopy(value) + + monkeypatch.setattr(Api, "send", send) + return routes, calls diff --git a/tests/test_collections.py b/tests/test_collections.py new file mode 100644 index 0000000..117a1e8 --- /dev/null +++ b/tests/test_collections.py @@ -0,0 +1,205 @@ +import pytest + +import cloudmusic +from cloudmusic import IncompleteResultError, ResponseError +from cloudmusic.sessions import Session +from conftest import song + +PLAYLIST = "/weapi/v6/playlist/detail" +SEARCH = "/weapi/search/get" +COMMENT = "/weapi/v2/resource/comments" + + +def comment(comment_id): + return { + "commentId": comment_id, + "content": str(comment_id), + "time": comment_id, + "likedCount": 3, + "user": {"userId": 10, "nickname": "用户"}, + } + + +@pytest.mark.parametrize("track_ids", [[], [1], [3, 1, 2]]) +def test_playlist_complete_empty_single_and_order(api_send, track_ids): + routes, calls = api_send + routes[PLAYLIST] = { + "code": 200, + "playlist": { + "trackCount": len(track_ids), + "trackIds": [{"id": i} for i in track_ids], + "tracks": [song(i) for i in reversed(track_ids)], + }, + } + assert [m.id for m in cloudmusic.getPlaylist(1)] == list(map(str, track_ids)) + assert len(calls) == 1 # Existing details reused, no audio or duplicate metadata request. + + +def test_playlist_hydrates_track_ids_and_detects_truncation(api_send): + routes, calls = api_send + routes[PLAYLIST] = { + "code": 200, + "playlist": { + "trackCount": 3, + "trackIds": [{"id": 1}, {"id": 2}], + "tracks": [song(1)], + }, + } + routes["/weapi/v3/song/detail"] = {"code": 200, "songs": [song(2)]} + with pytest.raises(IncompleteResultError) as caught: + cloudmusic.getPlaylist(1) + assert caught.value.expected_count == 3 + assert caught.value.actual_count == 2 + assert [m.id for m in caught.value.partial_result] == ["1", "2"] + assert len(calls) == 2 + + +@pytest.mark.parametrize("song_ids", [[], [1], [3, 2, 1]]) +def test_album_preserves_order_without_extra_fetch(api_send, song_ids): + routes, calls = api_send + routes["/weapi/v1/album/20"] = {"code": 200, "songs": [song(i) for i in song_ids]} + assert [m.id for m in cloudmusic.getAlbum(20)] == list(map(str, song_ids)) + assert len(calls) == 1 + + +def test_search_pages_and_deduplicates(api_send): + routes, calls = api_send + pages = iter( + [ + {"songs": [{"id": 1}, {"id": 2}], "hasMore": True}, + {"songs": [{"id": 2}, {"id": 3}], "hasMore": True}, + {"songs": [{"id": 4}], "hasMore": False}, + ] + ) + routes[SEARCH] = lambda _: {"code": 200, "result": next(pages)} + routes["/weapi/v3/song/detail"] = {"code": 200, "songs": [song(i) for i in range(1, 5)]} + assert [m.id for m in cloudmusic.search('白日 "test"', 4)] == ["1", "2", "3", "4"] + search_calls = [p for path, p in calls if path == SEARCH] + assert [p["offset"] for p in search_calls] == [0, 2, 4] + assert search_calls[0]["s"] == '白日 "test"' + + +def test_search_empty_and_no_progress(api_send): + routes, calls = api_send + assert cloudmusic.search("anything", 0) == [] + assert calls == [] + routes[SEARCH] = {"code": 200, "result": {"songCount": 0}} + assert cloudmusic.search("no matches") == [] + routes[SEARCH] = {"code": 200, "result": {"songs": [], "hasMore": True}} + with pytest.raises(IncompleteResultError): + cloudmusic.search("broken") + routes[SEARCH] = {"code": 200, "result": {"songs": "invalid"}} + with pytest.raises(ResponseError): + cloudmusic.search("broken") + + +@pytest.mark.parametrize("keyword,number", [("", 1), (" ", 1), (None, 1), ("a", -1), ("a", 1.2)]) +def test_bad_search_input(keyword, number): + with pytest.raises(ValueError): + cloudmusic.search(keyword, number) + + +def test_comments_100_unique_with_short_pages_and_overlap(api_send): + routes, calls = api_send + + def page(params): + cursor = int(params["cursor"]) + start = max(1, cursor) # One overlapping item between pages. + end = min(101, start + params["pageSize"]) + if cursor == 0: + end = 19 # Service can return fewer items while hasMore remains true. + rows = [comment(i) for i in range(start, end)] + return { + "code": 200, + "data": {"comments": rows, "hasMore": end < 101, "cursor": str(end - 1)}, + } + + routes[COMMENT] = page + with Session() as client: + result = client.comment({"ID": 1, "clas": "new", "number": 100}) + assert [c["commentId"] for c in result] == list(range(1, 101)) + assert all(0 < params["pageSize"] <= 20 for _, params in calls) + assert all(params["sortType"] == 3 for _, params in calls) + + +def test_comments_end_before_requested_count(api_send): + api_send[0][COMMENT] = { + "code": 200, + "data": {"comments": [comment(1)], "hasMore": False, "cursor": "1"}, + } + with Session() as client: + assert len(client.comment({"ID": 1, "clas": "new", "number": 100})) == 1 + + +@pytest.mark.parametrize("cursor,rows", [("0", [comment(1)]), (None, []), ("", [comment(1)])]) +def test_comment_pagination_stops_when_stalled(api_send, cursor, rows): + api_send[0][COMMENT] = { + "code": 200, + "data": {"comments": rows, "hasMore": True, "cursor": cursor}, + } + with Session() as client, pytest.raises(IncompleteResultError): + client.comment({"ID": 1, "clas": "new", "number": 100}) + assert len(api_send[1]) == 1 + + +def test_comment_wrappers_count_hot_deleted_user_and_zero(api_send): + routes, calls = api_send + routes["/weapi/v3/song/detail"] = {"code": 200, "songs": [song(1)]} + routes["/weapi/v1/resource/comments/R_SO_4_1"] = { + "code": 200, + "total": 123, + "hotComments": [{**comment(1), "user": None}], + } + routes[COMMENT] = {"code": 200, "data": {"comments": [], "hasMore": False}} + music = cloudmusic.getMusic(1) + assert music.getCommentsCount() == 123 + assert music.getHotComments()[0]["nickName"] == "" + assert music.getComments(1) == [] + calls.clear() + assert music.getHotComments(0) == [] + assert music.getComments(0) == [] + assert calls == [] + with pytest.raises(ValueError): + music.getComments(-1) + + +def test_dispatch_and_invalid_operations(api_send): + routes, _ = api_send + routes["/weapi/v3/song/detail"] = {"code": 200, "songs": [song(1)]} + with Session() as client: + assert client.request("song", 1).id == "1" + assert client.downloader().procs == 2 + with pytest.raises(ValueError): + client.request("unknown", 1) + with pytest.raises(ValueError): + client.comment({"ID": 1, "clas": "unknown", "number": 1}) + with pytest.raises(ValueError): + Session(level="magic") + + +def test_search_rejects_missing_payload_and_uses_total_without_more_flag(api_send): + routes, _ = api_send + routes[SEARCH] = {"code": 200, "result": {}} + with pytest.raises(ResponseError): + cloudmusic.search("test") + pages = iter( + [ + {"songs": [{"id": 1}], "songCount": 2}, + {"songs": [{"id": 2}], "songCount": 2}, + ] + ) + routes[SEARCH] = lambda _: {"code": 200, "result": next(pages)} + routes["/weapi/v3/song/detail"] = {"code": 200, "songs": [song(1), song(2)]} + assert [m.id for m in cloudmusic.search("test", 2)] == ["1", "2"] + + +def test_duplicate_comment_page_with_new_cursor_does_not_loop(api_send): + routes, calls = api_send + routes[COMMENT] = lambda params: { + "code": 200, + "data": {"comments": [comment(1)], "hasMore": True, "cursor": str(params["pageNo"])}, + } + with Session() as client, pytest.raises(IncompleteResultError) as caught: + client.comment({"ID": 1, "clas": "new", "number": 10}) + assert len(calls) == 2 + assert len(caught.value.partial_result) == 1 diff --git a/tests/test_download.py b/tests/test_download.py new file mode 100644 index 0000000..e451816 --- /dev/null +++ b/tests/test_download.py @@ -0,0 +1,162 @@ +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest + +import cloudmusic +from cloudmusic.download import _filename, download +from cloudmusic.exceptions import DownloadError, UnavailableError + + +@pytest.fixture +def server(): + seen = [] + content = b"audio-data" * 20000 + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + seen.append(dict(self.headers)) + if self.path == "/error": + self.send_error(503) + return + self.send_response(200) + self.send_header("Content-Type", "text/html" if self.path == "/html" else "audio/mpeg") + payload = b"" if self.path == "/empty" else content + self.send_header( + "Content-Length", str(len(payload) + (1 if self.path == "/short" else 0)) + ) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *args): + pass + + httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + worker = threading.Thread(target=httpd.serve_forever, daemon=True) + worker.start() + try: + yield "http://127.0.0.1:{}".format(httpd.server_port), content, seen + finally: + httpd.shutdown() + httpd.server_close() + worker.join(timeout=5) + + +def music(url): + return cloudmusic.Music( + 1, + url, + "higher", + 10, + "mp3", + { + "name": "歌/曲:*?", + "artist": ["歌手"], + "album": "专辑", + }, + options={"cookie": {"MUSIC_U": "do-not-forward"}, "timeout": (1, 2)}, + ) + + +def test_streamed_download_to_custom_directory_and_no_cookie(server, tmp_path): + base, expected, seen = server + destination = tmp_path / "new" / "directory" + value = music(base + "/ok") + result = Path(value.download(destination)) + assert result.is_absolute() and result.parent == destination + assert result.read_bytes() == expected + assert result.name == "歌_曲___ - 歌手.mp3" + assert value.name == "歌/曲:*?" + assert not list(destination.glob("*.part")) + assert "Cookie" not in seen[0] + + +def test_default_directory(server, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + result = Path(music(server[0] + "/ok").download()) + assert result.parent == tmp_path / "cloudmusic" + + +@pytest.mark.parametrize("path", ["/error", "/html", "/empty", "/short"]) +def test_failed_download_does_not_replace_existing_file(server, tmp_path, path): + value = music(server[0] + path) + target = tmp_path / _filename(value, "mp3") + target.write_bytes(b"keep existing") + with pytest.raises(DownloadError): + value.download(tmp_path) + assert target.read_bytes() == b"keep existing" + assert sorted(p.name for p in tmp_path.iterdir()) == [target.name] + + +def test_long_unicode_and_reserved_filenames(): + value = music("https://example.invalid") + value.name = "长" * 500 + assert len(_filename(value, "mp3").encode()) <= 184 + value.name = "CON.txt" + assert _filename(value, "mp3").startswith("_") + with pytest.raises(DownloadError): + _filename(value, "../mp3") + value.name = ".." + value.artist = [] + assert "/" not in _filename(value, "mp3") + + +def test_invalid_audio_address(tmp_path): + with pytest.raises(UnavailableError): + download(tmp_path, music(None)) + with pytest.raises(DownloadError): + download(tmp_path, music("file:///etc/passwd")) + + +def test_batch_loader_success_and_failure(monkeypatch, tmp_path): + loader = cloudmusic.createLoader(2, tmp_path) + loader.data = [music("a"), music("b")] + monkeypatch.setattr(cloudmusic.Music, "download", lambda self, dirs: str(Path(dirs) / self.id)) + assert loader.start() == str(tmp_path) + assert len(loader.results) == 2 + loader.data[1].id = "2" + + def fail_one(self, dirs): + if self.id == "2": + raise UnavailableError("unavailable") + return "saved" + + monkeypatch.setattr(cloudmusic.Music, "download", fail_one) + with pytest.raises(DownloadError): + loader.start() + assert loader.results == ["saved"] and 1 in loader.errors + loader.data = ["not music"] + with pytest.raises(ValueError): + loader.start() + loader.data = [] + assert loader.start() == str(tmp_path) + assert not loader.results and not loader.errors + with pytest.raises(ValueError): + cloudmusic.createLoader(0) + + +def test_failed_atomic_replace_preserves_old_file(server, tmp_path, monkeypatch): + value = music(server[0] + "/ok") + target = tmp_path / _filename(value, "mp3") + target.write_bytes(b"keep existing") + + def fail_replace(*args): + raise PermissionError("destination locked") + + monkeypatch.setattr("cloudmusic.download.os.replace", fail_replace) + with pytest.raises(DownloadError) as caught: + value.download(tmp_path) + assert isinstance(caught.value.__cause__, PermissionError) + assert target.read_bytes() == b"keep existing" + assert list(tmp_path.iterdir()) == [target] + + +def test_unwritable_directory_has_actionable_error(tmp_path, monkeypatch): + def deny(*args, **kwargs): + raise PermissionError("read only") + + monkeypatch.setattr(Path, "mkdir", deny) + with pytest.raises(DownloadError) as caught: + music("https://example.invalid/audio").download(tmp_path / "denied") + assert isinstance(caught.value.__cause__, PermissionError) diff --git a/tests/test_live.py b/tests/test_live.py new file mode 100644 index 0000000..23b0f92 --- /dev/null +++ b/tests/test_live.py @@ -0,0 +1,67 @@ +"""Opt-in, low-volume checks. Remote restrictions are failures, never silent passes.""" + +import os + +import pytest + +import cloudmusic + +pytestmark = pytest.mark.live +SONG_ID = 1347630432 +PLAYLIST_ID = 310729011 + + +@pytest.fixture(scope="module") +def live_client(request): + cookie_env = request.config.getoption("--cookie-env") + if cookie_env and not os.environ.get(cookie_env): + raise pytest.UsageError("Missing cookie environment variable: {}".format(cookie_env)) + with cloudmusic.Client(cookie=os.environ[cookie_env] if cookie_env else None) as client: + yield client + + +@pytest.fixture(scope="module") +def sample(live_client): + return live_client.getMusic(SONG_ID) + + +def test_live_song_metadata(sample): + assert sample.id == str(SONG_ID) + assert sample.name and sample.artist and sample.albumId + + +def test_live_search(live_client): + results = live_client.search("白日", 5) + assert len(results) == 5 + assert len({song.id for song in results}) == 5 + assert all(song.name for song in results) + + +def test_live_playlist_complete(live_client): + # Client checks trackCount against the returned IDs before returning success. + songs = live_client.getPlaylist(PLAYLIST_ID) + assert len(songs) > 1 + assert len({song.id for song in songs}) == len(songs) + assert all(song.name for song in songs) + + +def test_live_album(sample, live_client): + songs = live_client.getAlbum(sample.albumId) + assert sample.id in {song.id for song in songs} + + +def test_live_lyrics(sample): + original, translation = sample.getLyrics() + assert original and isinstance(translation, str) + + +def test_live_comments(sample): + assert sample.getCommentsCount() >= 100 + hot = sample.getHotComments(3) + assert len(hot) == 3 and all(row["content"] for row in hot) + comments = sample.getComments(100) + assert len(comments) == 100 + assert len({row["commentId"] for row in comments}) == 100 + assert [row["time"] for row in comments] == sorted( + (row["time"] for row in comments), reverse=True + ) diff --git a/tests/test_music.py b/tests/test_music.py new file mode 100644 index 0000000..f652276 --- /dev/null +++ b/tests/test_music.py @@ -0,0 +1,188 @@ +import json +from concurrent.futures import ThreadPoolExecutor + +import pytest + +import cloudmusic +from cloudmusic import APIError, NotFoundError, ResponseError, UnavailableError +from cloudmusic.api import Api +from cloudmusic.musicObj import _PlaybackBatch, createObj +from conftest import audio, song + +DETAIL = "/weapi/v3/song/detail" +URL = "/weapi/song/enhance/player/url/v1" + + +def default_details(params): + return {"code": 200, "songs": [song(item["id"]) for item in reversed(json.loads(params["c"]))]} + + +def test_input_shape_order_duplicates_and_empty(api_send): + routes, calls = api_send + routes[DETAIL] = default_details + assert cloudmusic.getMusic(1).id == "1" + assert [m.id for m in cloudmusic.getMusic([2, 1, 2])] == ["2", "1", "2"] + assert isinstance(cloudmusic.getMusic([1]), list) + assert isinstance(cloudmusic.getMusic((1,)), list) + calls.clear() + assert cloudmusic.getMusic([]) == [] + assert calls == [] + + +@pytest.mark.parametrize("bad", [None, True, 0, -1, 1.5, "abc", [1, False], [0], {}]) +def test_invalid_ids_never_make_request(bad, api_send): + with pytest.raises(ValueError): + cloudmusic.getMusic(bad) + assert api_send[1] == [] + + +def test_missing_metadata_cannot_reuse_another_song(api_send): + api_send[0][DETAIL] = {"code": 200, "songs": [song(1)]} + with pytest.raises(NotFoundError) as caught: + cloudmusic.getMusic([1, 2]) + assert caught.value.ids == [2] + + +def test_unexpected_metadata_schema(api_send): + api_send[0][DETAIL] = {"code": 200, "songs": [{"id": 1}]} + with pytest.raises(ResponseError): + cloudmusic.getMusic(1) + + +def test_metadata_does_not_require_audio_and_lazy_batch_reuses_request(api_send): + routes, calls = api_send + routes[DETAIL] = default_details + routes[URL] = {"code": 200, "data": [audio(2, level="standard"), audio(1)]} + music = cloudmusic.getMusic([1, 2]) + assert [m.name for m in music] == ["歌曲1", "歌曲2"] + assert len(calls) == 1 + assert music[1].url == "https://example.invalid/audio" + assert music[1].level == "standard" + assert music[0].size == 4 and music[0].type == "mp3" + assert music[0].freeTrialInfo is None + assert len(calls) == 2 + assert calls[1][1]["ids"] == [1, 2] + assert repr(music[0]) == "" + + +def test_batching_for_large_queries(api_send): + routes, calls = api_send + routes[DETAIL] = default_details + result = cloudmusic.getMusic(list(range(1, 452))) + assert len(result) == 451 + assert [len(json.loads(params["c"])) for _, params in calls] == [200, 200, 51] + + +def test_concurrent_audio_lookup_uses_one_batch(api_send): + routes, calls = api_send + routes[DETAIL] = default_details + routes[URL] = {"code": 200, "data": [audio(i) for i in range(1, 11)]} + songs = cloudmusic.getMusic(list(range(1, 11))) + with ThreadPoolExecutor(10) as pool: + assert all(pool.map(lambda m: m.url, songs)) + assert len(calls) == 2 + + +def test_expired_audio_is_refreshed(api_send, monkeypatch): + routes, calls = api_send + routes[URL] = {"code": 200, "data": [audio(1)]} + batch = _PlaybackBatch([1], "higher", {}) + monkeypatch.setattr("cloudmusic.musicObj.time.monotonic", lambda: 10) + assert batch.get(1)["url"] + monkeypatch.setattr("cloudmusic.musicObj.time.monotonic", lambda: 71) + assert batch.get(1)["url"] + assert len(calls) == 2 + + +def test_audio_challenge_preserves_metadata_and_is_not_repeated(api_send, monkeypatch): + routes, calls = api_send + monkeypatch.setattr("cloudmusic.musicObj.time.monotonic", lambda: 10) + routes[DETAIL] = default_details + routes[URL] = APIError(-460, "network risk") + music = cloudmusic.getMusic(1) + with pytest.raises(APIError): + _ = music.url + assert music.name == "歌曲1" + routes[URL] = {"code": 200, "data": [audio(1)]} + with pytest.raises(APIError): + _ = music.url + assert len(calls) == 2 + monkeypatch.setattr("cloudmusic.musicObj.time.monotonic", lambda: 71) + assert music.url + + +def test_audio_missing_null_or_trial_is_explicit(api_send): + routes, _ = api_send + routes[DETAIL] = default_details + for data in ([], [audio(1, url=None)], [audio(1, freeTrialInfo={"start": 0, "end": 30})]): + routes[URL] = {"code": 200, "data": data} + music = cloudmusic.getMusic(1) + with pytest.raises(UnavailableError): + music.download() + + +def test_quality_change_keeps_directory_and_metadata(api_send, monkeypatch, tmp_path): + routes, calls = api_send + routes[DETAIL] = default_details + routes[URL] = {"code": 200, "data": [audio(1, type="flac", level="lossless")]} + captured = [] + monkeypatch.setattr( + "cloudmusic.download.download", + lambda dirs, music, **kwargs: captured.append((dirs, music, kwargs)) or "saved", + ) + music = cloudmusic.getMusic(1) + assert music.download(tmp_path, "lossless") == "saved" + assert captured[0][0] == tmp_path and captured[0][1] is music + assert captured[0][2]["audio"]["type"] == "flac" + assert calls[-1][1]["level"] == "lossless" + with pytest.raises(ValueError): + music.download(level="magic") + + +@pytest.mark.parametrize( + "payload,expected", + [ + ({"code": 200}, ["", ""]), + ({"code": 200, "lrc": {"lyric": "原文"}}, ["原文", ""]), + ({"code": 200, "lrc": None, "tlyric": None}, ["", ""]), + ({"code": 200, "lrc": {"lyric": "原文"}, "tlyric": {"lyric": "译文"}}, ["原文", "译文"]), + ], +) +def test_lyrics_without_audio_or_translation(api_send, payload, expected): + routes, calls = api_send + routes[DETAIL] = default_details + routes["/weapi/song/lyric"] = payload + assert cloudmusic.getMusic(1).getLyrics() == expected + assert len(calls) == 2 + + +def test_objects_retain_configuration_after_client_close(api_send): + routes, _ = api_send + routes[DETAIL] = default_details + with cloudmusic.Client(cookie="MUSIC_U=test", timeout=2) as client: + music = client.getMusic(1) + assert music._options["cookie"]["MUSIC_U"] == "test" + assert music._options["timeout"] == 2 + routes[URL] = {"code": 200, "data": [audio(1)]} + assert music.url + + +def test_helpers_and_alias_metadata(api_send, capsys): + detail = song(1) + detail["alia"] = ["别名"] + api_send[0][DETAIL] = {"code": 200, "songs": [detail]} + assert createObj([1])[0].name == "歌曲1 别名" + with Api() as api: + assert createObj([], api=api) == [] + cloudmusic.help() + assert "getMusic" in capsys.readouterr().out + + +def test_invalid_lyrics_shape_is_response_error(api_send): + routes, _ = api_send + routes[DETAIL] = default_details + routes["/weapi/song/lyric"] = {"code": 200, "lrc": "not an object"} + with pytest.raises(ResponseError): + cloudmusic.getMusic(1).getLyrics() + with pytest.raises(ValueError): + cloudmusic.getMusic(1).download(level="") diff --git a/tests/test_transport.py b/tests/test_transport.py new file mode 100644 index 0000000..08e2078 --- /dev/null +++ b/tests/test_transport.py @@ -0,0 +1,172 @@ +import base64 +import json +from unittest.mock import MagicMock + +import pytest +import requests +from Cryptodome.Cipher import AES + +from cloudmusic import APIError, RequestError, ResponseError +from cloudmusic import encrypt, query +from cloudmusic.api import Api + + +def response(payload=None, status=200): + value = MagicMock(status_code=status) + value.__enter__.return_value = value + value.json.return_value = payload + return value + + +def test_json_encryption_unicode_quotes_and_types(monkeypatch): + secret = b"0123456789abcdef" + monkeypatch.setattr(encrypt, "create_key", lambda _: secret) + payload = {"s": '白日 "hello" \\ path', "ids": [1], "flag": True, "nothing": None} + encrypted = encrypt.encrypted_request(payload) + + def decrypt(value, key): + plain = AES.new(key, AES.MODE_CBC, b"0102030405060708").decrypt(base64.b64decode(value)) + return plain[: -plain[-1]] + + assert json.loads(decrypt(decrypt(encrypted["params"], secret), encrypt.NONCE)) == payload + assert len(encrypted["encSecKey"]) == 256 + assert len(encrypt.create_key(16)) == 16 + assert encrypt.encrypted_request(payload, "linux") is payload + assert encrypt.linuxEncrypt() == {} + assert encrypt.encrypted_request() and encrypt.encrypted_request('{"id":1}') + + +def test_query_success_and_closes_response(): + reply = response({"code": 200, "songs": []}) + session = MagicMock() + session.post.return_value = reply + assert ( + query.post("https://example.invalid", {}, {}, session=session, timeout=(1, 2))["songs"] + == [] + ) + assert session.post.call_args.kwargs["timeout"] == (1, 2) + reply.__exit__.assert_called_once() + + +@pytest.mark.parametrize( + "payload,expected", + [ + ([], ResponseError), + ({}, ResponseError), + ({"code": -460, "message": "网络环境存在风险"}, APIError), + ({"code": -462, "message": "需要验证"}, APIError), + ({"code": 301}, APIError), + ({"code": 50000005}, APIError), + ], +) +def test_business_errors_never_turn_into_keyerror(payload, expected): + session = MagicMock() + session.post.return_value = response(payload) + with pytest.raises(expected) as caught: + query.post("https://example.invalid", {}, {}, session=session) + if expected is APIError: + assert caught.value.code == payload["code"] + + +@pytest.mark.parametrize("status", [401, 403, 405, 429, 500]) +def test_http_error_keeps_status(status): + session = MagicMock() + session.post.return_value = response(status=status) + with pytest.raises(RequestError) as caught: + query.post("https://example.invalid", {}, {}, session=session) + assert caught.value.status_code == status + + +def test_invalid_json_and_timeout(): + session = MagicMock() + session.post.return_value = response() + session.post.return_value.json.side_effect = ValueError("invalid") + with pytest.raises(ResponseError): + query.post("https://example.invalid", {}, {}, session=session) + session.post.side_effect = requests.Timeout() + with pytest.raises(RequestError): + query.post("https://example.invalid", {}, {}, session=session) + + +def test_owned_http_session(monkeypatch): + session = MagicMock() + session.__enter__.return_value = session + session.post.return_value = response({"code": "200"}) + monkeypatch.setattr(query.requests, "Session", lambda: session) + assert query.post("https://example.invalid", {}, {})["code"] == "200" + session.__exit__.assert_called_once() + + +@pytest.mark.parametrize( + "timeout", [0, -1, True, None, float("nan"), float("inf"), (), (1, 2, 3), (1, 0)] +) +def test_invalid_timeout(timeout): + with pytest.raises(ValueError): + Api(timeout=timeout) + + +def test_client_cookie_csrf_config_and_close(monkeypatch): + captured = [] + monkeypatch.setattr( + query, "post", lambda *args, **kw: captured.append((args, kw)) or {"code": 200} + ) + monkeypatch.setattr(encrypt, "encrypted_request", lambda data, method="": data) + cookie = {"MUSIC_U": "test-token", "__csrf": "test-csrf"} + with Api(cookie=cookie, timeout=3, proxies={"https": "http://localhost:8888"}) as api: + cookie["MUSIC_U"] = "changed" + api.send("/weapi/test", {"id": 1}) + assert api.session.cookies.get("MUSIC_U", domain="music.163.com") == "test-token" + assert captured[0][0][0] == "https://music.163.com/weapi/test" + assert captured[0][0][2]["csrf_token"] == "test-csrf" + assert captured[0][1]["session"] is api.session + assert "Cookie" not in api.headers and "cookie" not in api.headers + with pytest.raises(RuntimeError): + api.send("/weapi/test") + with Api(cookie="MUSIC_U=test; __csrf=csrf") as api: + assert api.options["cookie"]["__csrf"] == "csrf" + with Api() as api: + assert "MUSIC_U" not in api.options["cookie"] + + +@pytest.mark.parametrize("cookie", [False, 1, [], "not a cookie"]) +def test_bad_cookie(cookie): + with pytest.raises(ValueError): + Api(cookie=cookie) + + +def test_random_key_generation(): + first, second = encrypt.create_key(16), encrypt.create_key(16) + assert len(first) == len(second) == 16 + assert first != second + + +def test_required_endpoint_parameters(api_send): + routes, calls = api_send + routes["/weapi/v3/song/detail"] = {"code": 200, "songs": []} + routes["/weapi/v1/resource/comments/R_SO_4_1"] = {"code": 200} + routes["/weapi/song/enhance/player/url/v1"] = {"code": 200, "data": []} + with Api() as api: + api.get_song_detail({"ID": [1]}) + api.get_commets({"ID": 1}) + api.get_song_url({"ID": [1], "level": "standard"}) + assert json.loads(calls[0][1]["ids"]) == [1] + assert json.loads(calls[0][1]["c"]) == [{"id": 1}] + assert calls[1][1] == {"rid": "1", "offset": 0, "beforeTime": 0, "limit": 20} + assert calls[2][1] == {"ids": [1], "level": "standard", "encodeType": "aac"} + + +def test_server_issued_cookies_reach_followup_objects(monkeypatch): + def reply(url, headers, data, *, session, timeout): + session.cookies.set("NMTID", "anonymous-session", domain=".music.163.com", path="/") + session.cookies.set("foreign", "not-copied", domain="example.invalid", path="/") + return {"code": 200} + + monkeypatch.setattr(query, "post", reply) + with Api() as api: + api.send("/weapi/test") + assert api.options["cookie"]["NMTID"] == "anonymous-session" + assert "foreign" not in api.options["cookie"] + with Api(**api.options) as followup: + assert ( + followup.session.cookies.get("NMTID", domain="music.163.com") == "anonymous-session" + ) diff --git a/tests/test_user.py b/tests/test_user.py new file mode 100644 index 0000000..b6e116b --- /dev/null +++ b/tests/test_user.py @@ -0,0 +1,61 @@ +import pytest + +import cloudmusic +from cloudmusic import IncompleteResultError, ResponseError +from cloudmusic.userObj import createUser +from conftest import song + + +def user(): + return cloudmusic.User({"profile": {"userId": 1, "nickname": "用户"}}) + + +def test_user_optional_fields_and_public_alias(api_send): + api_send[0]["/weapi/v1/user/detail/1"] = { + "code": 200, + "profile": {"userId": 1, "nickname": "用户"}, + } + value = cloudmusic.getUser(1) + assert value.id == "1" + assert value.nickname == value.nickName == "用户" + assert value.level == value.listenSongs == value.fans == 0 + assert createUser(1).id == "1" + with pytest.raises(ResponseError): + cloudmusic.User({"profile": None}) + + +def test_user_playlist_paginates_without_duplicates(api_send): + routes, calls = api_send + pages = iter( + [ + {"playlist": [{"id": 10, "name": "a"}], "more": True}, + {"playlist": [{"id": 10}, {"id": 11, "name": "b"}], "more": False}, + ] + ) + routes["/weapi/user/playlist"] = lambda _: next(pages) + assert [p["id"] for p in user().getPlaylist()] == [10, 11] + assert [p["offset"] for _, p in calls] == [0, 1] + + +def test_user_playlist_stalled_page(api_send): + api_send[0]["/weapi/user/playlist"] = {"playlist": [], "more": True} + with pytest.raises(IncompleteResultError): + user().getPlaylist() + + +@pytest.mark.parametrize("record_type,key", [(0, "allData"), (1, "weekData"), (2, "weekData")]) +def test_record_only_fetches_requested_period_and_preserves_scores(api_send, record_type, key): + routes, calls = api_send + routes["/weapi/v1/play/record"] = { + "code": 200, + key: [ + {"song": song(2), "score": 100}, + {"song": song(1), "score": 20}, + ], + } + result = user().getRecord(record_type) + assert [(r["music"].id, r["score"]) for r in result] == [("2", 100), ("1", 20)] + assert len(calls) == 1 + assert calls[0][1]["type"] == (1 if record_type else 0) + routes["/weapi/v1/play/record"] = {"code": 200, key: []} + assert user().getRecord(record_type) == []