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
-
简单示例
-
-- 通过歌曲id获取Music对象,并展示部分属性
-
-
music = cloudmusic.getMusic(1347630432)
-
-print("歌名:{}".format(music.name))
-print("歌手:{}".format(music.artist))
-print("音频文件url:{}".format(music.url))
-
-
-
-
-
-
-
-- 通过歌单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("------------")
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-- 通过关键字搜索获取前100首歌曲的Music对象,并输出每首歌的评论数量
-
-
results = cloudmusic.search("白日", 100)
-
-for music in results:
- print(music.getCommentsCount())
-
-
-
-
-
-
-
-
-
-
参考文档
-
-
-
-
-
-- url:歌曲音频文件链接
-- id:歌曲id
-- name:歌曲名称
-- artist:歌手名称
-- artistId: 歌手id
-- album:专辑名称
-- albumId: 专辑id
-- size:音频文件大小
-- type:音频文件类型(mp3或m4a)
-- level:歌曲品质。默认higher
-- picUrl: 专辑图url
-
-
-
-
--
-
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()
-返回值为整型,本首歌的评论数量。
-
-
-
-
-
--
-
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一样。
-
-
-
-
-
-
-
-
-- id:用户id
-- level:用户等级
-- listenSongs:累计听歌数量
-- createTime:账号创建时间
-- nickName:用户昵称
-- avatarUrl:头像url
-- city:所在城市的行政区划代码
-- province:所在省份的行政区划代码
-- vipType:vip类型。0表示未开通vip,10表示开通音乐包,11表示开通黑胶
-- birthday:生日时间戳
-- signature:个性签名
-- fans:粉丝数量
-- follows:关注的用户数量
-- eventCount:动态数量
-- playlistCount:创建的歌单数量
-
-
-
-
--
-
getPlaylist()
-获取用户创建和收藏的全部歌单,返回值为一个列表,每个元素是一个歌单信息组成的字典。字典内容:
-id:歌单id;
-name:歌单名称;
-creatorId:创建者id;
-createTime:创建时间;
-coverImgUrl:歌单封面图片url;
-updateTime:歌单最后一次更新时间;
-tags:歌单所属音乐风格,列表形式
-
--
-
getRecord(recordType)
-
-recordType: 可选。默认为0。为0返回所有时间排行,非0返回最近一周排行。
-
-获取用户听歌排行,返回一个列表,元素为字典。字典内容:
-score:分值。100表示听歌次数最多,越小越少;
-music:Music对象
-
-
-
-
-
-
-
@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))
-
-
-
-
-
-
-
-- 通过歌单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("------------")
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-- 通过关键字搜索获取前100首歌曲的Music对象,并输出每首歌的评论数量
-
-
results = cloudmusic.search("白日", 100)
-
-for music in results:
- print(music.getCommentsCount())
-
-
-
-
-
-
-
-
-
-
参考文档
-
-
-
-
-
-- url:歌曲音频文件链接
-- id:歌曲id
-- name:歌曲名称
-- rtist:歌手名称
-- artistId: 歌手id
-- album:专辑名称
-- albumId: 专辑id
-- size:音频文件大小
-- type:音频文件类型(mp3或m4a)
-- level:歌曲品质。默认higher
-- picUrl: 专辑图url
-
-
-
-
--
-
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()
-返回值为整型,本首歌的评论数量。
-
-
-
-
-
--
-
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一样。
-
-
-
-
-
-
-
-
-- id:用户id
-- level:用户等级
-- listenSongs:累计听歌数量
-- createTime:账号创建时间
-- nickName:用户昵称
-- avatarUrl:头像url
-- city:所在城市的行政区划代码
-- province:所在省份的行政区划代码
-- vipType:vip类型。0表示未开通vip,10表示开通音乐包,11表示开通黑胶
-- birthday:生日时间戳
-- signature:个性签名
-- fans:粉丝数量
-- follows:关注的用户数量
-- eventCount:动态数量
-- playlistCount:创建的歌单数量
-
-
-
-
--
-
getPlaylist()
-获取用户创建和收藏的全部歌单,返回值为一个列表,每个元素是一个歌单信息组成的字典。字典内容:
-id:歌单id;
-name:歌单名称;
-creatorId:创建者id;
-createTime:创建时间;
-coverImgUrl:歌单封面图片url;
-updateTime:歌单最后一次更新时间;
-tags:歌单所属音乐风格,列表形式
-
--
-
getRecord(recordType)
-
-recordType: 可选。默认为0。为0返回所有时间排行,非0返回最近一周排行。
-
-获取用户听歌排行,返回一个列表,元素为字典。字典内容:
-score:分值。100表示听歌次数最多,越小越少;
-music:Music对象
-
-
-
-
-
-
-
@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) == []