From d2e6856899165160e22359744a32a7a477f70e04 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun, 27 Sep 2026 10:53:48 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20get=5Fplaylist=5Fdetail=20=E6=89=B9?= =?UTF-8?q?=E6=AC=A1=E6=8B=89=E5=8F=96=E6=8B=86=E8=A7=A3=EF=BC=8894=20?= =?UTF-8?q?=E8=A1=8C=20/=20=E5=9C=88=2024=20/=20=E5=B5=8C=E5=A5=97=203=20?= =?UTF-8?q?=E2=86=92=2060=20=E8=A1=8C=20/=20=E5=9C=88=209=20/=20=E5=B5=8C?= =?UTF-8?q?=E5=A5=97=201=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：歌单解析的唯一上游通道，94 行单函数把「歌单头解析 / trackIds 分批拉明细 / 失败批次记账 / incomplete 标记 / 错误映射」五件事揉在 一个 try 里；曲目字段映射与搜索歌曲路逐字段重复。 拆出（模块级，与 search_music 拆解同风格）： - _song_base_fields：歌曲条目公共字段（搜索/歌单共用同一容错口径）， _map_song_item 改为其加 type 键的包装 - _playlist_track_ids：trackIds → 字符串 ID 列表 - _collect_playlist_tracks：批次循环整体（失败批次计数、脏 ID 跳批、 明细缺 id 跳过、incomplete 判定输入），嵌套 3 是批次语义固有， 集中在单一职责函数内 保留语义：脏 ID（'①'.isdigit() 为真但 int() 抛 ValueError）跳过整批 且不计 failed_batches；批次阶段网络错误仍映射「获取歌单详情请求失败」。 先写测试后动代码：新增 tests/test_playlist_detail.py 11 例，在未改动 实现上跑绿（修正 1 例测试自身的替身消费顺序），覆盖此前零测试路径—— playlist:null 防枚举、业务 code 错误、非 dict 响应、批次失败记账、 脏 ID 跳批、100/批分批、错误映射。 门禁：ruff 全过 · pyright 0/0 · pytest 359 passed（348→359）。 --- music_api/api.py | 102 +++++++++------- tests/test_playlist_detail.py | 211 ++++++++++++++++++++++++++++++++++ 2 files changed, 270 insertions(+), 43 deletions(-) create mode 100644 tests/test_playlist_detail.py diff --git a/music_api/api.py b/music_api/api.py index 2c3944d..8e740cc 100644 --- a/music_api/api.py +++ b/music_api/api.py @@ -29,10 +29,9 @@ from .errors import ( ) -def _map_song_item(item: dict[str, Any]) -> dict[str, Any]: - """歌曲条目映射。""" +def _song_base_fields(item: dict[str, Any]) -> dict[str, Any]: + """歌曲条目公共字段（搜索四路与歌单详情共用同一容错口径）。""" return { - 'type': 'song', 'id': item['id'], 'name': item.get('name') or '未知歌曲', 'artists': '/'.join(filter(None, (a.get('name') for a in item.get('ar') or []))) or '未知歌手', @@ -41,6 +40,57 @@ def _map_song_item(item: dict[str, Any]) -> dict[str, Any]: } +def _map_song_item(item: dict[str, Any]) -> dict[str, Any]: + """歌曲条目映射。""" + return { + 'type': 'song', + **_song_base_fields(item) + } + + +def _playlist_track_ids(playlist: dict[str, Any]) -> list[str]: + """trackIds → 字符串 ID 列表（缺 id 的条目跳过）。""" + return [str(t['id']) for t in (playlist.get('trackIds') or []) if t.get('id')] + + +async def _collect_playlist_tracks(track_ids: list[str], headers: dict[str, str], + cookies: dict[str, str]) -> tuple[list[dict[str, Any]], int]: + """分批拉取歌单全部曲目明细，返回 (tracks, failed_batches)。 + + 明细批次的业务 code 也要校验，失败批次记入 failed_batches，不再静默缺歌。 + 脏 ID 跳过整批（'①'.isdigit() 为真但 int() 抛 ValueError——原实现此处 + 未捕获会整歌单 500，现为跳批）；明细缺 id 的歌跳过不拖整批。 + """ + tracks: list[dict[str, Any]] = [] + failed_batches = 0 + for i in range(0, len(track_ids), 100): + batch_ids = track_ids[i:i + 100] + try: + song_data = {'c': json.dumps([{'id': int(sid), 'v': 0} for sid in batch_ids + if str(sid).isdigit()])} + except (TypeError, ValueError): + continue + + song_resp = await requests.post(APIConstants.SONG_DETAIL_V3, data=song_data, + headers=headers, cookies=cookies, timeout=30) + song_resp.raise_for_status() + + song_result = song_resp.json() + if not isinstance(song_result, dict): + # 上游 null/数组体按失败批次计，不让 AttributeError 失真 + failed_batches += 1 + continue + if song_result.get('code') != 200: + # 本批次业务失败：计为失败批次，交付方可见残缺状态 + failed_batches += 1 + continue + for song in (song_result.get('songs') or []): + if not song.get('id'): + continue + tracks.append(_song_base_fields(song)) + return tracks, failed_batches + + def _map_album_item(item: dict[str, Any]) -> dict[str, Any]: """专辑条目映射（artists 为空**不**兜底成"未知歌手"——与歌曲路不同口径）。""" return { @@ -286,7 +336,7 @@ class NeteaseAPI: cookies: 用户cookies Returns: - 歌单详情信息 + 歌单详情信息（曲目数与 trackIds 不一致时附 incomplete/missing/failed_batches） Raises: APIException: API调用失败时抛出 @@ -324,47 +374,13 @@ class NeteaseAPI: 'tracks': [] } - # 获取所有trackIds并分批获取详细信息（明细批次的业务 code 也要校验， - # 失败批次记入 failed_batches，不再静默缺歌） - track_ids = [str(t['id']) for t in (playlist.get('trackIds') or []) if t.get('id')] - failed_batches = 0 - for i in range(0, len(track_ids), 100): - batch_ids = track_ids[i:i+100] - # trackIds 混入非数字脏 ID 时跳过该条（原实现 int 抛 ValueError - # 不在捕获列表，整歌单 500）；明细缺 id 的歌跳过不拖整批 - try: - song_data = {'c': json.dumps([{'id': int(sid), 'v': 0} for sid in batch_ids - if str(sid).isdigit()])} - except (TypeError, ValueError): - continue - - song_resp = await requests.post(APIConstants.SONG_DETAIL_V3, data=song_data, - headers=headers, cookies=cookies, timeout=30) - song_resp.raise_for_status() - - song_result = song_resp.json() - if not isinstance(song_result, dict): - # 上游 null/数组体按失败批次计，不让 AttributeError 失真 - failed_batches += 1 - continue - if song_result.get('code') != 200: - # 本批次业务失败：计为失败批次，交付方可见残缺状态 - failed_batches += 1 - continue - for song in (song_result.get('songs') or []): - if not song.get('id'): - continue - info['tracks'].append({ - 'id': song['id'], - 'name': song.get('name') or '未知歌曲', - 'artists': '/'.join(filter(None, (a.get('name') for a in song.get('ar') or []))) or '未知歌手', - 'album': (song.get('al') or {}).get('name') or '未知专辑', - 'picUrl': (song.get('al') or {}).get('picUrl') or '' - }) + track_ids = _playlist_track_ids(playlist) + tracks, failed_batches = await _collect_playlist_tracks(track_ids, headers, cookies) + info['tracks'] = tracks # trackIds 数与实际拿到的不一致时显式标记"不完整"，不冒充完整结果 - if len(info['tracks']) != len(track_ids): + if len(tracks) != len(track_ids): info['incomplete'] = True - info['missing'] = len(track_ids) - len(info['tracks']) + info['missing'] = len(track_ids) - len(tracks) info['failed_batches'] = failed_batches return info diff --git a/tests/test_playlist_detail.py b/tests/test_playlist_detail.py new file mode 100644 index 0000000..65db6d6 --- /dev/null +++ b/tests/test_playlist_detail.py @@ -0,0 +1,211 @@ +# -*- coding: utf-8 -*- +"""get_playlist_detail 的行为契约测试（沙箱，不触真实上游）。 + +为什么：歌单详情是「歌单解析」功能的唯一上游通道（94 行 / 批次拉取 / +嵌套 3），此前零测试覆盖。它的容错口径都是对外可观察行为： +- 上游 code:200 + playlist:null →「歌单不存在或无权访问」（路由统一转 404） +- 明细批次业务失败计 failed_batches，不静默缺歌 +- trackIds 混入脏 ID（如带圈数字——isdigit() 为真但 int() 抛错）跳过整批 +- trackIds 数与实得数不一致时显式标 incomplete/missing + +按惯例：先在未改动的原始实现上跑绿，重构时这批测试即等价性验证器。 +替身边界：monkeypatch ``music_api.api.requests.post``，按 URL 分流—— +PLAYLIST_DETAIL_API 与 SONG_DETAIL_V3 分别给响应替身。 +""" +import json +from typing import Any + +import pytest + +from music_api import APIException +from music_api import api as api_mod +from music_api.api import NeteaseAPI +from music_api.constants import APIConstants + +_API = NeteaseAPI() + + +class _FakeResp: + """``requests.post`` 响应替身：json 载荷或注入异常。""" + + def __init__(self, payload: Any = None, *, + status_error: Exception | None = None, + json_error: Exception | None = None): + self.payload = payload + self.status_error = status_error + self.json_error = json_error + + def raise_for_status(self) -> None: + if self.status_error is not None: + raise self.status_error + + def json(self) -> Any: + if self.json_error is not None: + raise self.json_error + return self.payload + + +@pytest.fixture() +def api_capture(monkeypatch): + """按 URL 分流的 requests.post 替身：歌单详情响应固定，批次响应按序消费。""" + calls: list[dict[str, Any]] = [] + state: dict[str, Any] = { + 'playlist_resp': _FakeResp({'code': 200, 'playlist': {}}), + 'batch_resps': [_FakeResp({'code': 200, 'songs': []})], + } + + async def _fake_post(url: str, **kw: Any): + calls.append({'url': url, 'data': kw.get('data')}) + if url == APIConstants.PLAYLIST_DETAIL_API: + return state['playlist_resp'] + assert url == APIConstants.SONG_DETAIL_V3, f'unexpected url: {url}' + if not state['batch_resps']: + raise AssertionError('批次响应耗尽') + return state['batch_resps'].pop(0) + + monkeypatch.setattr(api_mod.requests, 'post', _fake_post) + state['calls'] = calls + return state + + +def _playlist_payload(playlist: dict[str, Any]) -> _FakeResp: + return _FakeResp({'code': 200, 'playlist': playlist}) + + +def _batch_c_payloads(calls: list[dict[str, Any]]) -> list[list[int]]: + """从捕获的调用里提取每批的 c 载荷（id 列表）。""" + out = [] + for c in calls: + if c['url'] == APIConstants.SONG_DETAIL_V3: + ids = [e['id'] for e in json.loads(c['data']['c'])] + out.append(ids) + return out + + +async def _detail(api_capture, playlist: dict[str, Any]) -> dict[str, Any]: + api_capture['playlist_resp'] = _playlist_payload(playlist) + return await _API.get_playlist_detail(123, {}) + + +# ============ 歌单头 ============ + +async def test_playlist_null_is_apiexception(api_capture): + """上游对不存在/私密歌单返回 code:200 + playlist:null → 显式 APIException。""" + api_capture['playlist_resp'] = _playlist_payload({}) + with pytest.raises(APIException, match='歌单不存在或无权访问'): + await _API.get_playlist_detail(123, {}) + + +async def test_business_code_error(api_capture): + api_capture['playlist_resp'] = _FakeResp({'code': 512, 'message': '需要登录'}) + with pytest.raises(APIException, match='获取歌单详情失败: 需要登录'): + await _API.get_playlist_detail(123, {}) + + +async def test_non_dict_response(api_capture): + api_capture['playlist_resp'] = _FakeResp(None) + with pytest.raises(APIException, match='响应格式异常'): + await _API.get_playlist_detail(123, {}) + + +# ============ 成功主路径 ============ + +async def test_success_maps_tracks(api_capture): + api_capture['playlist_resp'] = _playlist_payload({ + 'id': 123, 'name': '歌单A', 'coverImgUrl': 'https://cover/1', + 'creator': {'nickname': '作者'}, 'trackCount': 2, + 'trackIds': [{'id': 1}, {'id': 2}, {'id': None}], + }) + api_capture['batch_resps'] = [_FakeResp({'code': 200, 'songs': [ + {'id': 1, 'name': '歌A', 'ar': [{'name': '歌手X'}], + 'al': {'name': '专辑Z', 'picUrl': 'https://pic/1.jpg'}}, + {'id': None, 'name': '残缺'}, + {'id': 2, 'name': '歌B', 'ar': [], 'al': {}}, + ]})] + + out = await _API.get_playlist_detail(123, {}) + assert out['name'] == '歌单A' + assert out['creator'] == '作者' + assert out['tracks'] == [ + {'id': 1, 'name': '歌A', 'artists': '歌手X', + 'album': '专辑Z', 'picUrl': 'https://pic/1.jpg'}, + {'id': 2, 'name': '歌B', 'artists': '未知歌手', + 'album': '未知专辑', 'picUrl': ''}, + ] + assert 'incomplete' not in out + assert _batch_c_payloads(api_capture['calls']) == [[1, 2]] # 缺 id 的 trackId 不进批次 + + +# ============ 批次容错 ============ + +async def test_batch_business_failure_marked_incomplete(api_capture): + """明细批次业务失败：计 failed_batches，显式标 incomplete，不冒充完整。""" + api_capture['batch_resps'] = [_FakeResp({'code': 512, 'message': '爆了'})] + out = await _detail(api_capture, {'id': 1, 'trackIds': [{'id': 1}, {'id': 2}]}) + assert out['tracks'] == [] + assert out['incomplete'] is True + assert out['missing'] == 2 + assert out['failed_batches'] == 1 + + +async def test_batch_non_dict_counts_as_failed(api_capture): + """批次响应体为 null（非 dict）：按失败批次计，不让 AttributeError 失真。""" + api_capture['batch_resps'] = [_FakeResp(None)] + out = await _detail(api_capture, {'id': 1, 'trackIds': [{'id': 9}]}) + assert out['incomplete'] is True + assert out['missing'] == 1 + assert out['failed_batches'] == 1 + + +async def test_dirty_trackid_skips_whole_batch_without_fail_count(api_capture): + """脏 ID（'①'.isdigit() 为真但 int() 抛 ValueError）：跳过整批且不计失败批次。""" + api_capture['batch_resps'] = [] # 不应有批次请求发出 + out = await _detail(api_capture, {'id': 1, 'trackIds': [{'id': '①'}]}) + assert out['tracks'] == [] + assert out['incomplete'] is True + assert out['missing'] == 1 + assert out['failed_batches'] == 0 + assert _batch_c_payloads(api_capture['calls']) == [] + + +async def test_empty_trackids_no_batch_no_incomplete(api_capture): + """trackIds 为空：不发生批次请求，结果完整。""" + out = await _detail(api_capture, {'id': 1, 'trackIds': []}) + assert out['tracks'] == [] + assert 'incomplete' not in out + assert _batch_c_payloads(api_capture['calls']) == [] + + +async def test_batches_of_100(api_capture): + """101 首分两批：首批 100、次批 1。""" + api_capture['batch_resps'] = [ + _FakeResp({'code': 200, 'songs': []}), + _FakeResp({'code': 200, 'songs': []}), + ] + track_ids = [{'id': i} for i in range(1, 102)] + out = await _detail(api_capture, {'id': 1, 'trackIds': track_ids}) + payloads = _batch_c_payloads(api_capture['calls']) + assert len(payloads) == 2 + assert len(payloads[0]) == 100 + assert payloads[1] == [101] + assert out['incomplete'] is True # 明细一首未回，101 全缺 + assert out['missing'] == 101 + + +# ============ 错误映射 ============ + +async def test_network_error_maps_to_request_failed(api_capture, monkeypatch): + """网络层异常（含批次阶段的）：统一转「获取歌单详情请求失败」。""" + + async def _boom(url: str, **kw: Any): + raise api_mod.requests.RequestException('connection reset') + + monkeypatch.setattr(api_mod.requests, 'post', _boom) + with pytest.raises(APIException, match='获取歌单详情请求失败'): + await _API.get_playlist_detail(123, {}) + + +async def test_json_decode_error_maps_to_parse_failed(api_capture): + api_capture['playlist_resp'] = _FakeResp(json_error=json.JSONDecodeError('bad', 'x', 0)) + with pytest.raises(APIException, match='解析歌单详情响应失败'): + await _API.get_playlist_detail(123, {})