refactor: get_playlist_detail 批次拉取拆解(94 行 / 圈 24 / 嵌套 3 → 60 行 / 圈 9 / 嵌套 1)

为什么:歌单解析的唯一上游通道,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)。
This commit is contained in:
snowy-yang committed 2026-09-27 10:53:48 +08:00
1 parent f0de0b65b1
commit d2e6856899
2 files changed
+270 -43

No files matched your search

+59 -43
View File
@@ -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 { return {
'type': 'song',
'id': item['id'], 'id': item['id'],
'name': item.get('name') or '未知歌曲', 'name': item.get('name') or '未知歌曲',
'artists': '/'.join(filter(None, (a.get('name') for a in item.get('ar') or []))) 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]: def _map_album_item(item: dict[str, Any]) -> dict[str, Any]:
"""专辑条目映射(artists 为空**不**兜底成"未知歌手"——与歌曲路不同口径)。""" """专辑条目映射(artists 为空**不**兜底成"未知歌手"——与歌曲路不同口径)。"""
return { return {
@@ -286,7 +336,7 @@ class NeteaseAPI:
cookies: 用户cookies cookies: 用户cookies
Returns: Returns:
歌单详情信息 歌单详情信息(曲目数与 trackIds 不一致时附 incomplete/missing/failed_batches)
Raises: Raises:
APIException: API调用失败时抛出 APIException: API调用失败时抛出
@@ -324,47 +374,13 @@ class NeteaseAPI:
'tracks': [] 'tracks': []
} }
# 获取所有trackIds并分批获取详细信息(明细批次的业务 code 也要校验, track_ids = _playlist_track_ids(playlist)
# 失败批次记入 failed_batches,不再静默缺歌) tracks, failed_batches = await _collect_playlist_tracks(track_ids, headers, cookies)
track_ids = [str(t['id']) for t in (playlist.get('trackIds') or []) if t.get('id')] info['tracks'] = tracks
failed_batches = 0
for i in range(0, len(track_ids), 100):
batch_ids = track_ids[i:i+100]
# trackIds 混入非数字脏 ID 时跳过该条(原实现 int 抛 ValueError
# 不在捕获列表,整歌单 500);明细缺 id 的歌跳过不拖整批
try:
song_data = {'c': json.dumps([{'id': int(sid), 'v': 0} for sid in batch_ids
if str(sid).isdigit()])}
except (TypeError, ValueError):
continue
song_resp = await requests.post(APIConstants.SONG_DETAIL_V3, data=song_data,
headers=headers, cookies=cookies, timeout=30)
song_resp.raise_for_status()
song_result = song_resp.json()
if not isinstance(song_result, dict):
# 上游 null/数组体按失败批次计,不让 AttributeError 失真
failed_batches += 1
continue
if song_result.get('code') != 200:
# 本批次业务失败:计为失败批次,交付方可见残缺状态
failed_batches += 1
continue
for song in (song_result.get('songs') or []):
if not song.get('id'):
continue
info['tracks'].append({
'id': song['id'],
'name': song.get('name') or '未知歌曲',
'artists': '/'.join(filter(None, (a.get('name') for a in song.get('ar') or []))) or '未知歌手',
'album': (song.get('al') or {}).get('name') or '未知专辑',
'picUrl': (song.get('al') or {}).get('picUrl') or ''
})
# trackIds 数与实际拿到的不一致时显式标记"不完整",不冒充完整结果 # trackIds 数与实际拿到的不一致时显式标记"不完整",不冒充完整结果
if len(info['tracks']) != len(track_ids): if len(tracks) != len(track_ids):
info['incomplete'] = True info['incomplete'] = True
info['missing'] = len(track_ids) - len(info['tracks']) info['missing'] = len(track_ids) - len(tracks)
info['failed_batches'] = failed_batches info['failed_batches'] = failed_batches
return info return info
+211
View File
@@ -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, {})