refactor: search_music 四路映射拆解(101 行 / 圈 32 / 嵌套 6 → 42 行 / 圈 7 / 嵌套 1)

为什么:站内搜索的唯一上游通道,101 行里 if/elif 四路条目映射占了
2/3,错误映射与字段容错口径(歌曲 artists 空缺兜底未知歌手、专辑
不兜底、alias 缺省 '' 等)全埋在分支里,改一路要通读全函数。

拆出(模块级,与既有 _ensure_dict 同风格):
- _map_song_item / _map_album_item / _map_artist_item / _map_playlist_item
  ——四路口径原样,docstring 标注各路容错差异
- _SEARCH_RESULT_KEYS + _SEARCH_ITEM_MAPPERS 派发表(未知类型不在表内
  → 空结果,原 if/elif 无 else 行为一致)
- _collect_search_items 统一「取键 → 无 id 跳过 → 映射」

先写测试后动代码:新增 tests/test_search_music.py 12 例,在未改动
实现上跑绿(修正 1 例替身形状:网络异常必须抛垫片自己的
RequestException,aiohttp.ClientError 由 aiohttp_compat 内部翻译,
直接抛底层是不存在的形状)。契约测试不为违反 offset: int 签名的
None 入参背书(防御性 int(offset or 0) 保留,不测)。
门禁:ruff 全过 · pyright 0/0 · pytest 330 passed(319→330)。
This commit is contained in:
snowy-yang committed 2026-09-27 09:42:07 +08:00
1 parent 37f8d77405
commit e00ac14460
2 files changed
+289 -60

No files matched your search

+77 -60
View File
@@ -6,6 +6,7 @@
import json
import time
from collections.abc import Callable
from random import randrange
from typing import Any
@@ -28,6 +29,81 @@ from .errors import (
)
def _map_song_item(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 '未知歌手',
'album': (item.get('al') or {}).get('name') or '未知专辑',
'picUrl': (item.get('al') or {}).get('picUrl') or ''
}
def _map_album_item(item: dict[str, Any]) -> dict[str, Any]:
"""专辑条目映射(artists 为空**不**兜底成"未知歌手"——与歌曲路不同口径)。"""
return {
'type': 'album',
'id': item['id'],
'name': item.get('name') or '未知名称',
'artists': '/'.join(filter(None, (a.get('name') for a in item.get('artists') or []))),
'alias': '',
'album': '',
'picUrl': item.get('picUrl')
}
def _map_artist_item(item: dict[str, Any]) -> dict[str, Any]:
"""歌手条目映射(alias 缺失/为空 → '')。"""
return {
'type': 'artist',
'id': item['id'],
'name': item.get('name') or '未知名称',
'artists': '',
'alias': '/'.join(item.get('alias', [])) if item.get('alias') else '',
'album': '',
'picUrl': item.get('picUrl')
}
def _map_playlist_item(item: dict[str, Any]) -> dict[str, Any]:
"""歌单条目映射。"""
return {
'type': 'playlist',
'id': item['id'],
'name': item.get('name') or '未知名称',
'artists': '',
'creator': (item.get('creator') or {}).get('nickname', ''),
'trackCount': item.get('trackCount', 0),
'album': '',
'picUrl': item.get('coverImgUrl')
}
# search_type → (上游结果键, 条目映射);未知类型不在表内 → 空结果
_SEARCH_RESULT_KEYS: dict[int, str] = {
1: 'songs', 10: 'albums', 100: 'artists', 1000: 'playlists',
}
_SEARCH_ITEM_MAPPERS: dict[int, Callable[[dict[str, Any]], dict[str, Any]]] = {
1: _map_song_item, 10: _map_album_item, 100: _map_artist_item, 1000: _map_playlist_item,
}
def _collect_search_items(result_data: dict[str, Any], search_type: int) -> list[dict[str, Any]]:
"""按类型取上游结果键并逐条映射;残缺条目(无 id,下架/屏蔽)跳过,不拖死整页。"""
key = _SEARCH_RESULT_KEYS.get(search_type)
mapper = _SEARCH_ITEM_MAPPERS.get(search_type)
if key is None or mapper is None:
return []
items: list[dict[str, Any]] = []
for item in result_data.get(key, []):
if not item.get('id'):
continue
items.append(mapper(item))
return items
class NeteaseAPI:
"""网易云音乐API主类"""
@@ -196,66 +272,7 @@ class NeteaseAPI:
raise APIException(f"搜索失败: {result.get('message', '未知错误')}")
result_data = result.get('result') or {}
items: list[dict[str, Any]] = []
if search_type == 1:
# 歌曲(id 改 .get——上游单条残缺(下架/屏蔽)不再拖死整页搜索)
for item in result_data.get('songs', []):
if not item.get('id'):
continue
items.append({
'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 '未知歌手',
'album': (item.get('al') or {}).get('name') or '未知专辑',
'picUrl': (item.get('al') or {}).get('picUrl') or ''
})
elif search_type == 10:
# 专辑(同上)
for item in result_data.get('albums', []):
if not item.get('id'):
continue
items.append({
'type': 'album',
'id': item['id'],
'name': item.get('name') or '未知名称',
'artists': '/'.join(filter(None, (a.get('name') for a in item.get('artists') or []))),
'alias': '',
'album': '',
'picUrl': item.get('picUrl')
})
elif search_type == 100:
# 歌手(同上)
for item in result_data.get('artists', []):
if not item.get('id'):
continue
items.append({
'type': 'artist',
'id': item['id'],
'name': item.get('name') or '未知名称',
'artists': '',
'alias': '/'.join(item.get('alias', [])) if item.get('alias') else '',
'album': '',
'picUrl': item.get('picUrl')
})
elif search_type == 1000:
# 歌单(同上)
for item in result_data.get('playlists', []):
if not item.get('id'):
continue
items.append({
'type': 'playlist',
'id': item['id'],
'name': item.get('name') or '未知名称',
'artists': '',
'creator': (item.get('creator') or {}).get('nickname', ''),
'trackCount': item.get('trackCount', 0),
'album': '',
'picUrl': item.get('coverImgUrl')
})
return items
return _collect_search_items(result_data, search_type)
except requests.RequestException as e:
raise APIException(f"搜索请求失败: {e}") from e
except (json.JSONDecodeError, KeyError) as e: