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, {})