From f27bd7dd748b28e0750092e814a69538a8f1a2a9 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun, 27 Sep 2026 07:41:16 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20/song=20=E7=AB=AF=E7=82=B9=E6=8B=86?= =?UTF-8?q?=E8=A7=A3=EF=BC=8899=20=E8=A1=8C=20/=20=E5=9C=88=2018=20?= =?UTF-8?q?=E2=86=92=20=E4=B8=BB=E5=87=BD=E6=95=B0=2027=20=E8=A1=8C=20/=20?= =?UTF-8?q?=E5=9C=88=208=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：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 沙箱）。 --- tests/test_song_types.py | 307 +++++++++++++++++++++++++++++++++++++++ web/routers/music.py | 207 +++++++++++++++----------- 2 files changed, 431 insertions(+), 83 deletions(-) create mode 100644 tests/test_song_types.py diff --git a/tests/test_song_types.py b/tests/test_song_types.py new file mode 100644 index 0000000..99b86fa --- /dev/null +++ b/tests/test_song_types.py @@ -0,0 +1,307 @@ +# -*- coding: utf-8 -*- +"""`/song` 端点四种 type 的行为契约。 + +为什么需要它：golden 快照里的 /song 只有 **一条**（song_badid，走参数校验）， +url / name / lyric / json 四条业务分支零覆盖。把 99 行 / 圈 44 的单函数拆成 +_parse_song_query + 三个响应函数后，需要行为级测试钉住： + +- 四种 type 各自打到哪个上游方法、返回体形状（含 `id` 必须是**字符串**： + 它来自 extract_id、原样进响应，golden 按字符串比对） +- 校验顺序（缺参 → ID 非数字 → 音质 → 类型）——同时传错时报哪个 +- json 分支的三个上游任一失败时的降级口径（歌词不阻断、详情失败转 404） +- 解析日志**只在 json 分支**记录（历史实现如此，不是遗漏） + +断言只看「调用后世界里发生了什么」：状态码 / 关键字段 / 上游收到了什么参数， +不引用任何私有函数名，故可跨后续重构复用。 +""" +import sqlite3 +import time + +import pytest +from fastapi.testclient import TestClient + +import auth_system +import core.db as core_db +import music_api +from core import session as cs +from core.state import GLOBAL_MUSIC_U_CACHE +from web import ratelimit + +SECRET = 'song-test-secret' +# create_app(api_prefix='') 时路由挂在**根**（与 test_golden 同一挂载方式）， +# 故这里是 /song 而不是 /api/v1/song。 +PATH = '/song' + + +class FakeAPI: + """music_api.NeteaseAPI 的替身：记录调用，返回预置结果或抛预置异常。""" + + def __init__(self) -> None: + self.calls: list[tuple] = [] + self.url_result: object = {'data': [{ + 'id': 1, 'url': 'http://x/a.mp3', 'level': 'lossless', + 'size': 1024, 'type': 'mp3', 'br': 999, + }]} + self.detail_result: object = {'songs': [{ + 'name': '某首歌', + 'al': {'name': '某专辑', 'picUrl': 'http://x/p.jpg'}, + 'ar': [{'name': '甲'}, {'name': '乙'}], + }]} + self.lyric_result: object = {'lrc': {'lyric': '歌词本体'}, + 'tlyric': {'lyric': '译词'}, + 'yrc': {'lyric': '逐字'}} + self.url_exc: Exception | None = None + self.detail_exc: Exception | None = None + self.lyric_exc: Exception | None = None + + async def get_song_url(self, mid, level, cookies): + self.calls.append(('get_song_url', mid, level)) + if self.url_exc is not None: + raise self.url_exc + return self.url_result + + async def get_song_detail(self, mid): + self.calls.append(('get_song_detail', mid)) + if self.detail_exc is not None: + raise self.detail_exc + return self.detail_result + + async def get_lyric(self, mid, cookies): + self.calls.append(('get_lyric', mid)) + if self.lyric_exc is not None: + raise self.lyric_exc + return self.lyric_result + + +@pytest.fixture() +def fake_api(monkeypatch): + fake = FakeAPI() + monkeypatch.setattr(music_api, 'NeteaseAPI', lambda *a, **kw: fake) + return fake + + +@pytest.fixture() +def logs(monkeypatch): + """替换全局 Cookie 与解析日志，返回日志调用记录。 + + 同时清掉 GLOBAL_MUSIC_U_CACHE 并置空限流闸： + - 该缓存是 5 分钟粘性的模块级状态，别的测试先跑会把值填上，导致本文件的 + 「个人 Cookie 与全局账号一致 → 拒绝」用例拿到缓存值而非本处 patch 的 + Cookie（表现为单跑通过、全套失败）。monkeypatch.setitem 保证 teardown 还原。 + - ratelimit 是进程内计数，让它参与会让用例受其它文件的请求数影响而随机 429。 + """ + recs: list[tuple] = [] + + async def _noop_guard(*a, **kw): + return None + + monkeypatch.setitem(GLOBAL_MUSIC_U_CACHE, 'v', None) + monkeypatch.setitem(GLOBAL_MUSIC_U_CACHE, 'ts', 0.0) + monkeypatch.setattr('web.routers.music._cookies', lambda: {'MUSIC_U': 'global'}) + monkeypatch.setattr('web.routers.music._record_log', lambda *a: recs.append(a)) + monkeypatch.setattr(ratelimit, 'guard', _noop_guard) + return recs + + +@pytest.fixture() +def client(tmp_path, monkeypatch): + """自建站点库 + 造一份合法会话 Cookie（不触碰生产 data/）。""" + path = tmp_path / 'site.db' + conn = sqlite3.connect(str(path)) + conn.executescript(""" + CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, role TEXT DEFAULT 'user', + status INTEGER DEFAULT 1, login_seq INTEGER DEFAULT 0, + duration_seconds INTEGER DEFAULT 0, ban_reason TEXT DEFAULT ''); + INSERT INTO users (id, username, role, status, login_seq, duration_seconds) + VALUES (1,'tester','user',1,7,100); + """) + conn.commit() + conn.close() + monkeypatch.setattr(core_db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(core_db, 'DB_PATH', path) + monkeypatch.setattr(auth_system.db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(auth_system.db, 'DB_PATH', path) + monkeypatch.setattr('web.deps.secret_key', lambda: SECRET) + from web.app import create_app + app = TestClient(create_app(api_prefix='')) + cookie = cs.dumps(SECRET, {'site_uid': 1, 'site_uid_seq': 7, 'site_uid_ts': int(time.time())}) + return app, {'Cookie': f'session={cookie}'} + + +def _post(client, headers, payload): + return client.post(PATH, json=payload, headers=headers) + + +# ============ 四种 type 的正路 ============ + + +def test_type_url_returns_direct_link(client, fake_api, logs): + app, h = client + r = _post(app, h, {'id': 123, 'type': 'url', 'level': 'lossless'}) + assert r.status_code == 200, r.text + body = r.json() + assert body['message'] == '获取歌曲URL成功' + assert body['data']['url'] == 'http://x/a.mp3' + assert body['data']['size'] == 1024 + assert body['data']['size_formatted'] # 上游 size 转可读串 + assert body['data']['quality_name'] # 音质显示名 + assert fake_api.calls == [('get_song_url', 123, 'lossless')] + assert logs == [], 'url 分支不记解析日志' + + +def test_type_url_empty_data_is_404(client, fake_api, logs): + """上游返回空 data 列表 → 404（与「拿不到直链」同一口径）。""" + app, h = client + fake_api.url_result = {'data': []} + r = _post(app, h, {'id': 123, 'type': 'url'}) + assert r.status_code == 404 + assert '获取音乐URL失败' in r.json()['message'] + + +def test_type_url_none_first_element_is_404(client, fake_api, logs): + """data 首元素为 null → song_data 退化为 {} → 404（不是 500）。""" + app, h = client + fake_api.url_result = {'data': [None]} + r = _post(app, h, {'id': 123, 'type': 'url'}) + assert r.status_code == 404 + + +def test_type_name_uses_detail(client, fake_api, logs): + app, h = client + r = _post(app, h, {'id': 456, 'type': 'name'}) + assert r.status_code == 200 + assert r.json()['message'] == '获取歌曲信息成功' + assert r.json()['data'] == fake_api.detail_result + assert fake_api.calls == [('get_song_detail', 456)] + assert logs == [] + + +def test_type_lyric_uses_lyric(client, fake_api, logs): + app, h = client + r = _post(app, h, {'id': 789, 'type': 'lyric'}) + assert r.status_code == 200 + assert r.json()['message'] == '获取歌词成功' + assert fake_api.calls == [('get_lyric', 789)] + assert logs == [] + + +def test_type_is_default_url(client, fake_api, logs): + """不传 type 时默认 url —— 与历史实现一致。""" + app, h = client + r = _post(app, h, {'id': 123}) + assert r.status_code == 200 + assert fake_api.calls == [('get_song_url', 123, 'lossless')] + + +# ============ json 分支：三源合并与降级 ============ + + +def test_type_json_merges_three_sources(client, fake_api, logs): + app, h = client + r = _post(app, h, {'id': 123, 'type': 'json'}) + assert r.status_code == 200 + d = r.json()['data'] + assert d['id'] == '123' and isinstance(d['id'], str), 'id 必须是 extract_id 的字符串原值' + assert d['name'] == '某首歌' + assert d['ar_name'] == '甲, 乙' + assert d['al_name'] == '某专辑' + assert d['pic'] == 'http://x/p.jpg' + assert d['lyric'] == '歌词本体' + assert d['tlyric'] == '译词' + assert d['yrc'] == '逐字' + assert d['url'] == 'http://x/a.mp3' + assert {c[0] for c in fake_api.calls} == {'get_song_detail', 'get_song_url', 'get_lyric'} + assert len(logs) == 1, 'json 分支记一条解析日志' + assert logs[0][0] == '解析' + + +def test_type_json_lyric_failure_does_not_break(client, fake_api, logs): + """歌词取失败不阻断整次解析（return_exceptions 的关键语义）。""" + app, h = client + fake_api.lyric_exc = music_api.APIException('歌词挂了') + r = _post(app, h, {'id': 123, 'type': 'json'}) + assert r.status_code == 200 + d = r.json()['data'] + assert d['lyric'] == '' and d['tlyric'] == '' and d['yrc'] == '' + assert d['name'] == '某首歌' and d['url'] == 'http://x/a.mp3' + + +def test_type_json_url_failure_yields_empty_url(client, fake_api, logs): + """直链取失败 → url 置空、size 记「获取失败」，其余照常返回。""" + app, h = client + fake_api.url_exc = music_api.APIException('直链挂了') + r = _post(app, h, {'id': 123, 'type': 'json'}) + assert r.status_code == 200 + d = r.json()['data'] + assert d['url'] == '' and d['size'] == '获取失败' + assert d['name'] == '某首歌' + + +def test_type_json_detail_failure_is_404(client, fake_api, logs): + """详情取失败 → 404「未找到歌曲信息」（不是 500，也不静默返回空壳）。""" + app, h = client + fake_api.detail_exc = music_api.APIException('详情挂了') + r = _post(app, h, {'id': 123, 'type': 'json'}) + assert r.status_code == 404 + assert r.json()['message'] == '未找到歌曲信息' + assert logs == [], '未产出响应数据则不记解析日志' + + +def test_type_json_empty_songs_is_404(client, fake_api, logs): + """详情返回但 songs 为空 → 同样 404。""" + app, h = client + fake_api.detail_result = {'songs': []} + r = _post(app, h, {'id': 123, 'type': 'json'}) + assert r.status_code == 404 + + +# ============ 参数校验与优先级 ============ + + +@pytest.mark.parametrize('payload,expected', [ + ({}, "必须提供 'ids'、'id' 或 'url' 参数"), + ({'id': 'abc'}, '无效的歌曲ID，请输入数字ID或分享链接'), + ({'id': 1, 'level': 'bad'}, '无效的音质参数'), + ({'id': 1, 'type': 'bad'}, '无效的类型参数'), +]) +def test_validation_messages(client, fake_api, logs, payload, expected): + app, h = client + r = _post(app, h, payload) + assert r.status_code == 400 + assert expected in r.json()['message'] + assert fake_api.calls == [], '校验不通过不应触达上游' + + +def test_validation_precedence_id_before_level(client, fake_api, logs): + """ID 与音质同时非法时报**先校验的那条**（顺序是用户可见行为）。""" + app, h = client + r = _post(app, h, {'id': 'abc', 'level': 'bad', 'type': 'bad'}) + assert r.status_code == 400 + assert '无效的歌曲ID' in r.json()['message'] + + +def test_url_param_via_share_link(client, fake_api, logs): + """用分享链接代替 id 也能解析出数字 ID（extract_id 的既有能力）。""" + app, h = client + r = _post(app, h, {'url': 'https://music.163.com/song?id=999', 'type': 'name'}) + assert r.status_code == 200 + assert fake_api.calls == [('get_song_detail', 999)] + + +def test_account_user_without_cookie_is_401(client, fake_api, logs): + """account=user 但未提供个人 Cookie → 401（不静默回落全局账号）。""" + app, h = client + r = _post(app, h, {'id': 1, 'type': 'name', 'account': 'user'}) + assert r.status_code == 401 + assert fake_api.calls == [] + + +def test_account_user_rejects_global_cookie(client, fake_api, logs, monkeypatch): + """个人 Cookie 与全局账号一致（全局 Cookie 泄露被冒用）→ 401 并记审计。""" + app, h = client + monkeypatch.setattr('web.routers.music.media.parse_cookie_str', + lambda raw: {'MUSIC_U': 'global'}) + r = _post(app, h, {'id': 1, 'type': 'name', 'account': 'user', + 'user_cookie': 'MUSIC_U=global'}) + assert r.status_code == 401 + assert fake_api.calls == [] diff --git a/web/routers/music.py b/web/routers/music.py index d7f3638..d6025f4 100644 --- a/web/routers/music.py +++ b/web/routers/music.py @@ -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)