refactor: /song 端点拆解（99 行 / 圈 18 → 主函数 27 行 / 圈 8）

为什么：song 是端点层最重的函数——参数校验（4 条）、账号 Cookie 选择、四种 type 分发、json 分支的三源合并与降级全塞在一处；而四种 type 的返回体形状与降级口径 **没有任何测试记录**（golden 里的 /song 只有一条参数校验用例 song_badid）。改法： - _parse_song_query：提取 + 校验，返回 _SongQuery 或**错误消息字符串**。返回字符串而非响应对象，是为了让它不依赖响应层、可单独断言（校验段 4 条错误都是默认 400，无需状态码）。 - _SongQuery：frozen dataclass。music_id 保持 extract_id 的**字符串原值**——它原样进响应体 id 字段、golden 按字符串比对；上游要的 int 走 .mid property。 - _song_url_response / _song_json_response：两个实体分支各自成函数。 - song 主函数只剩「校验 → 选 Cookie → 四路分发」。刻意保留的语义（是决定，不是疏漏）： - 解析日志**只在 json 分支**记录，url/name/lyric 三路不记——历史实现如此。已写进 _song_json_response 的 docstring：不要为了「统一」把它上提到主函数，那会新增三路的可观察副作用。 - 校验顺序（缺参 → ID 非数字 → 音质 → 类型）原样保留，并用测试钉住「同时传错时报哪条」。顺序是用户可见行为。 - 三条 404 降级（上游 data 为空 / data=[null] / songs 为空）逐条保留。复杂度（同口径 AST 实测，重构前后）： song 99 行 / 圈 18 / 嵌套 4 → 27 行 / 圈 8 / 嵌套 2 _song_json_response 新增 → 49 行 / 圈 6 / 嵌套 1 _parse_song_query 新增 → 21 行 / 圈 5 / 嵌套 1 文件内 >80 行函数 1 → 0；最大圈 18 → 11（余下最大的是本次未动的 mask_server_nickname）注：上一轮口头报的「圈 44」与本轮口径不同，此处以本轮统一口径为准，重测为准绳。等价性证据（三道，互相独立）： 1. AST 字面量审计：字符串字面量 **227 处完全一致**；数值唯一差异是新增的 dataclass frozen=True。即没有一条用户可见文案或阈值被改动。 2. 反向注入验证：把 HEAD 的旧 song 换回去，用同一套 tests/test_song_types.py 跑 → 19 例**全过**，证明这批测试忠实编码了旧行为、对新实现构成等价性验证。 3. 新增 19 例行为契约测试：四种 type 正路、json 的三源合并与三种降级（歌词失败不阻断 / 直链失败置空 / 详情失败 404）、校验优先级、分享链接解析、 account=user 无 Cookie 401、全局 Cookie 冒用拦截。顺带修掉一个测试自身缺陷：新用例里「个人 Cookie 与全局账号一致」一条**单跑通过、全套失败**——根因是它依赖 GLOBAL_MUSIC_U_CACHE（5 分钟粘性的模块级缓存），别的文件先跑会把值填上，于是拿到缓存值而非本处 patch 的 Cookie。已改为 fixture 里用 monkeypatch.setitem 显式重置该缓存，让用例自足。验证：ruff 全过 · pyright 0/0 · pytest 233 passed（214→233）· 生产 data/ 未被触碰（全部用例走 tmp_path 沙箱）。
2026-09-27 07:41:16 +08:00
1 parent 1973d6a0e8
commit f27bd7dd74
@@ -18,9 +18,11 @@ import asyncio
 import hmac
 import logging
 import time
+from dataclasses import dataclass
 from typing import Annotated, Any

 from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse

 import music_api
 from core import media, repo
@@ -274,28 +276,128 @@ async def playlist(user: LoginUser, request: Request):
    return success({'status': 'success', 'playlist': result}, '获取歌单详情成功')


+@dataclass(frozen=True)
+class _SongQuery:
+    """`/song` 的已校验入参。
+
+    ``music_id`` 保持 ``extract_id`` 的原始返回（字符串）：它会**原样**进入响应体
+    的 ``id`` 字段，golden 快照按字符串比对。转 int 只在上游调用时做（见 ``mid``）。
+    """
+    music_id: str
+    level: str
+    info_type: str
+
+    @property
+    def mid(self) -> int:
+        """上游接口要的整型 ID（响应体仍用原字符串）。"""
+        return int(self.music_id)
+
+
+def _parse_song_query(data: dict[str, Any]) -> _SongQuery | str:
+    """提取并校验 /song 入参：成功给 _SongQuery，失败给**错误消息**。
+
+    返回消息字符串而非响应对象，是为了让本函数不依赖响应层、可单独断言。
+    校验顺序即用户可见的错误优先级（缺参 → ID 非数字 → 音质 → 类型），
+    与历史实现逐条一致——改顺序会改变「同时传错时先报哪个」。
+    """
+    song_ids = data.get('ids') or data.get('id')
+    url = data.get('url')
+    if not song_ids and not url:
+        return "必须提供 'ids'、'id' 或 'url' 参数"
+    music_id = _extract_id(song_ids or url)
+    if not music_id.strip().isdigit():
+        return '无效的歌曲ID，请输入数字ID或分享链接'
+    level = data.get('level', 'lossless')
+    if level not in VALID_QUALITY_LEVELS:
+        return f"无效的音质参数，支持: {', '.join(VALID_QUALITY_LEVELS)}"
+    info_type = data.get('type', 'url')
+    if info_type not in _VALID_SONG_TYPES:
+        return f"无效的类型参数，支持: {', '.join(_VALID_SONG_TYPES)}"
+    return _SongQuery(music_id, level, info_type)
+
+
+async def _song_url_response(api: Any, query: _SongQuery, cookies: dict[str, str]) -> JSONResponse:
+    """type=url：取直链，附带可读的大小与音质名。"""
+    result = await api.get_song_url(query.mid, query.level, cookies)
+    if result and result.get('data') and len(result['data']) > 0:
+        song_data = result['data'][0] or {}
+        if not song_data:
+            return error('获取音乐URL失败，可能是版权限制或音质不支持', 404)
+        return success({
+            'id': song_data.get('id'),
+            'url': song_data.get('url'),
+            'level': song_data.get('level'),
+            'quality_name': media.quality_display_name(song_data.get('level', query.level)),
+            'size': song_data.get('size'),
+            'size_formatted': media.format_file_size(song_data.get('size', 0)),
+            'type': song_data.get('type'),
+            'bitrate': song_data.get('br'),
+        }, '获取歌曲URL成功')
+    return error('获取音乐URL失败，可能是版权限制或音质不支持', 404)
+
+
+async def _song_json_response(api: Any, query: _SongQuery, cookies: dict[str, str],
+                              request: Request) -> JSONResponse:
+    """type=json：并行取详情/直链/歌词，歌词失败不阻断整次解析。
+
+    解析日志只在**本分支**记录——历史实现如此，url/name/lyric 三路都不记。不要
+    为了「统一」把它上提到主函数，那会新增其余三路的可观察副作用。
+    """
+    song_info, url_info, lyric_info = await asyncio.gather(
+        api.get_song_detail(query.mid),
+        api.get_song_url(query.mid, query.level, cookies),
+        api.get_lyric(query.mid, cookies),
+        return_exceptions=True,
+    )
+    if isinstance(lyric_info, BaseException):
+        lyric_info = None
+    if isinstance(url_info, BaseException):
+        url_info = None
+    if isinstance(song_info, BaseException) or not song_info or not song_info.get('songs'):
+        return error('未找到歌曲信息', 404)
+
+    song_data = song_info['songs'][0]
+    _al = song_data.get('al') or {}
+    _ar_names = ', '.join(filter(None, ((a or {}).get('name') for a in (song_data.get('ar') or []))))
+    response_data: dict[str, Any] = {
+        'id': query.music_id,
+        'name': song_data.get('name') or '',
+        'ar_name': _ar_names,
+        'al_name': _al.get('name') or '',
+        'pic': _al.get('picUrl') or '',
+        'level': query.level,
+        'lyric': ((lyric_info or {}).get('lrc') or {}).get('lyric') or '',
+        'tlyric': ((lyric_info or {}).get('tlyric') or {}).get('lyric') or '',
+        'yrc': ((lyric_info or {}).get('yrc') or {}).get('lyric') or '',
+    }
+    if url_info and url_info.get('data') and len(url_info['data']) > 0:
+        url_data = url_info['data'][0]
+        response_data.update({
+            'url': url_data.get('url', ''),
+            'size': media.format_file_size(url_data.get('size', 0)),
+            'level': url_data.get('level', query.level),
+        })
+    else:
+        response_data.update({'url': '', 'size': '获取失败'})
+
+    try:
+        _record_log('解析', ratelimit.client_ip(request), response_data.get('name') or query.music_id)
+    except Exception:
+        pass
+    return success(response_data, '获取歌曲信息成功')
+
+
@router.api_route('/song', methods=['GET', 'POST'])
 async def song(user: LoginUser, request: Request):
    """歌曲信息（type: url/name/lyric/json）。"""
    await ratelimit.guard(request, bucket='free', limit=30, window=60)
    data = await _request_data(request)
-    song_ids = data.get('ids') or data.get('id')
-    url = data.get('url')
-    level = data.get('level', 'lossless')
-    info_type = data.get('type', 'url')

-    if not song_ids and not url:
-        return error("必须提供 'ids'、'id' 或 'url' 参数")
-    music_id = _extract_id(song_ids or url)
-    if not str(music_id).strip().isdigit():
-        return error('无效的歌曲ID，请输入数字ID或分享链接')
-    if level not in VALID_QUALITY_LEVELS:
-        return error(f"无效的音质参数，支持: {', '.join(VALID_QUALITY_LEVELS)}")
-    if info_type not in _VALID_SONG_TYPES:
-        return error(f"无效的类型参数，支持: {', '.join(_VALID_SONG_TYPES)}")
+    query = _parse_song_query(data)
+    if isinstance(query, str):
+        return error(query)

-    account = data.get('account', 'admin')
-    if account == 'user':
+    if data.get('account', 'admin') == 'user':
        cookies = await _resolve_user_cookies(data, user, request)
        if not cookies:
            return error('请先扫码登录个人账号后再使用个人解析', 401)
@@ -304,73 +406,12 @@ async def song(user: LoginUser, request: Request):

    api = music_api.NeteaseAPI()
    try:
-        if info_type == 'url':
-            result = await api.get_song_url(int(music_id), level, cookies)
-            if result and result.get('data') and len(result['data']) > 0:
-                song_data = result['data'][0] or {}
-                if not song_data:
-                    return error('获取音乐URL失败，可能是版权限制或音质不支持', 404)
-                return success({
-                    'id': song_data.get('id'),
-                    'url': song_data.get('url'),
-                    'level': song_data.get('level'),
-                    'quality_name': media.quality_display_name(song_data.get('level', level)),
-                    'size': song_data.get('size'),
-                    'size_formatted': media.format_file_size(song_data.get('size', 0)),
-                    'type': song_data.get('type'),
-                    'bitrate': song_data.get('br'),
-                }, '获取歌曲URL成功')
-            return error('获取音乐URL失败，可能是版权限制或音质不支持', 404)
-
-        if info_type == 'name':
-            return success(await api.get_song_detail(int(music_id)), '获取歌曲信息成功')
-
-        if info_type == 'lyric':
-            return success(await api.get_lyric(int(music_id), cookies), '获取歌词成功')
-
-        # info_type == 'json'：并行取详情/URL，歌词失败不阻断
-        song_info, url_info, lyric_info = await asyncio.gather(
-            api.get_song_detail(int(music_id)),
-            api.get_song_url(int(music_id), level, cookies),
-            api.get_lyric(int(music_id), cookies),
-            return_exceptions=True,
-        )
-        if isinstance(lyric_info, BaseException):
-            lyric_info = None
-        if isinstance(url_info, BaseException):
-            url_info = None
-        if isinstance(song_info, BaseException) or not song_info or not song_info.get('songs'):
-            return error('未找到歌曲信息', 404)
-
-        song_data = song_info['songs'][0]
-        _al = song_data.get('al') or {}
-        _ar_names = ', '.join(filter(None, ((a or {}).get('name') for a in (song_data.get('ar') or []))))
-        response_data = {
-            'id': music_id,
-            'name': song_data.get('name') or '',
-            'ar_name': _ar_names,
-            'al_name': _al.get('name') or '',
-            'pic': _al.get('picUrl') or '',
-            'level': level,
-            'lyric': ((lyric_info or {}).get('lrc') or {}).get('lyric') or '',
-            'tlyric': ((lyric_info or {}).get('tlyric') or {}).get('lyric') or '',
-            'yrc': ((lyric_info or {}).get('yrc') or {}).get('lyric') or '',
-        }
-        if url_info and url_info.get('data') and len(url_info['data']) > 0:
-            url_data = url_info['data'][0]
-            response_data.update({
-                'url': url_data.get('url', ''),
-                'size': media.format_file_size(url_data.get('size', 0)),
-                'level': url_data.get('level', level),
-            })
-        else:
-            response_data.update({'url': '', 'size': '获取失败'})
-
-        try:
-            _record_log('解析', ratelimit.client_ip(request), response_data.get('name') or str(music_id))
-        except Exception:
-            pass
-        return success(response_data, '获取歌曲信息成功')
-
+        if query.info_type == 'url':
+            return await _song_url_response(api, query, cookies)
+        if query.info_type == 'name':
+            return success(await api.get_song_detail(query.mid), '获取歌曲信息成功')
+        if query.info_type == 'lyric':
+            return success(await api.get_lyric(query.mid, cookies), '获取歌词成功')
+        return await _song_json_response(api, query, cookies, request)
    except APIException:
        return error('服务暂时不可用，请稍后再试', 500)