# -*- coding: utf-8 -*- """音乐端点。 端点： - 免费解析类（需登录，不扣次数）： POST/GET /api/v1/toplist ↔ Flask /toplist POST/GET /api/v1/artist_songs ↔ Flask /artist_songs POST/GET /api/v1/comments ↔ Flask /comments POST/GET /api/v1/search ↔ Flask /search POST/GET /api/v1/album ↔ Flask /album POST/GET /api/v1/playlist ↔ Flask /playlist POST/GET /api/v1/song ↔ Flask /song（四种 type：url/name/lyric/json） 参数获取口径：GET query / POST json / POST form 三路兼容。 上游调用走 music_api 的**异步**方法（原生并发，无同步桥）。 """ 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 from core.config import VALID_QUALITY_LEVELS from core.state import GLOBAL_MUSIC_U_CACHE, SERVER_NICK_CACHE from music_api import APIException from web import ratelimit from web.deps import SiteUser, require_login from web.request_utils import cookies as _cookies from web.request_utils import extract_id as _extract_id from web.request_utils import request_data as _request_data from web.request_utils import safe_int as _safe_int from web.response import error, success from web.serializers import record_req_log as _record_log router = APIRouter() LoginUser = Annotated[SiteUser, Depends(require_login)] logger = logging.getLogger('web.music') _VALID_SONG_TYPES = ['url', 'name', 'lyric', 'json'] _SERVER_NICK_LOCK: asyncio.Lock | None = None def _nick_lock() -> asyncio.Lock: global _SERVER_NICK_LOCK if _SERVER_NICK_LOCK is None: _SERVER_NICK_LOCK = asyncio.Lock() return _SERVER_NICK_LOCK async def mask_server_nickname(nick: str, user: SiteUser) -> str: """非管理员响应中把服务器账号昵称替换为站点名。 - 管理员本人可见真实昵称； - 缓存 10 分钟内复用；上游拿不到时回落站点库里的最后已知昵称继续遮蔽。 """ if not nick or user.is_admin: return nick import time def _replace(n): return nick.replace(n, '云村小助手') if (n and n in nick) else nick if SERVER_NICK_CACHE['nick'] is None or time.time() - SERVER_NICK_CACHE['ts'] > 600: async with _nick_lock(): if SERVER_NICK_CACHE['nick'] is None or time.time() - SERVER_NICK_CACHE['ts'] > 600: fetched = '' try: fetched = (await music_api.NeteaseAPI().get_account_info(_cookies()) or {}).get('nickname', '') except Exception: fetched = '' if fetched: SERVER_NICK_CACHE['nick'] = fetched SERVER_NICK_CACHE['ts'] = time.time() try: await repo.set_setting('server_nickname', fetched) except Exception: pass else: try: persisted = await repo.get_setting('server_nickname', '') except Exception: persisted = '' if persisted: SERVER_NICK_CACHE['nick'] = persisted SERVER_NICK_CACHE['ts'] = time.time() else: # 库里也从未有过：缩短重试间隔，本次无遮蔽依据只能放行 SERVER_NICK_CACHE['ts'] = time.time() - 600 + 30 return _replace(SERVER_NICK_CACHE['nick']) @router.api_route('/toplist', methods=['GET', 'POST']) async def toplist(_user: LoginUser, request: Request): """官方榜单列表。""" await ratelimit.guard(request, bucket='free', limit=30, window=60) try: result = await music_api.NeteaseAPI().get_toplist() return success(result, '获取榜单列表成功') except APIException: return error('服务暂时不可用，请稍后再试', 500) @router.api_route('/artist_songs', methods=['GET', 'POST']) async def artist_songs(_user: LoginUser, request: Request): """歌手信息与热门歌曲。""" await ratelimit.guard(request, bucket='free', limit=30, window=60) data = await _request_data(request) artist_id = data.get('id') or data.get('artist_id') if artist_id in (None, ''): return error('缺少 artist_id 参数') artist_id = _extract_id(artist_id) if not str(artist_id).strip().isdigit(): return error('无效的歌手ID，请输入数字ID或分享链接') try: result = await music_api.NeteaseAPI().get_artist_hot_songs(int(artist_id)) return success(result, '获取歌手歌曲成功') except APIException: return error('服务暂时不可用，请稍后再试', 500) @router.api_route('/comments', methods=['GET', 'POST']) async def comments(_user: LoginUser, request: Request): """歌曲评论（分页）。""" await ratelimit.guard(request, bucket='free', limit=30, window=60) data = await _request_data(request) song_id = data.get('id') or data.get('song_id') limit = max(1, min(_safe_int(data.get('limit'), 20), 100)) offset = max(0, _safe_int(data.get('offset'), 0)) if song_id in (None, ''): return error('缺少 song_id 参数') song_id = _extract_id(song_id) if not str(song_id).strip().isdigit(): return error('无效的歌曲ID，请输入数字ID或分享链接') try: result = await music_api.NeteaseAPI().get_song_comments(int(song_id), limit, offset) return success(result, '获取评论成功') except APIException: return error('服务暂时不可用，请稍后再试', 500) @router.api_route('/search', methods=['GET', 'POST']) async def search(user: LoginUser, request: Request): """搜索（歌曲/专辑/歌手/歌单），offset 透传上游。""" await ratelimit.guard(request, bucket='free', limit=30, window=60) data = await _request_data(request) keyword = data.get('keyword') or data.get('keywords') or data.get('q') limit = max(1, _safe_int(data.get('limit'), 30)) offset = max(0, _safe_int(data.get('offset'), 0)) search_type = data.get('type', '1') if not keyword: return error('缺少 keyword 参数') if limit > 100: limit = 100 try: stype = int(search_type) except (TypeError, ValueError): return error('参数格式错误：type 必须为数字') try: result = await music_api.NeteaseAPI().search_music( str(keyword), _cookies(), limit, stype, offset=offset) except APIException: return error('服务暂时不可用，请稍后再试', 500) # 向后兼容字段 + 服务器昵称遮蔽 if result: for song in result: if song.get('type', 'song') == 'song' and 'artists' in song: song['artist_string'] = song['artists'] if song.get('type') == 'playlist': if song.get('creator'): song['creator'] = await mask_server_nickname(song['creator'], user) if song.get('name'): song['name'] = await mask_server_nickname(song['name'], user) return success(result, '搜索完成') # ============ 个人/全局账号 Cookie ============ def _global_music_u() -> str: """全局账号 MUSIC_U（5 分钟缓存）。""" now = time.time() if GLOBAL_MUSIC_U_CACHE['v'] is None or now - GLOBAL_MUSIC_U_CACHE['ts'] > 300: try: ck = _cookies() GLOBAL_MUSIC_U_CACHE['v'] = (ck.get('MUSIC_U') or '').strip() except Exception: GLOBAL_MUSIC_U_CACHE['v'] = '' GLOBAL_MUSIC_U_CACHE['ts'] = now return GLOBAL_MUSIC_U_CACHE['v'] async def _resolve_user_cookies( data: dict[str, Any], user: 'SiteUser | None', request: Request ) -> dict[str, str]: """解析个人账号 Cookie（浏览器传入的 user_cookie，不落服务器）。 全局账号（cookie.txt）零个人身份权限：浏览器提交的 MUSIC_U 若与全局账号一致 （全局 Cookie 泄露被当个人 Cookie 冒用），一律拒绝并记审计。 """ ck = media.parse_cookie_str(data.get('user_cookie')) mu = (ck.get('MUSIC_U') or '').strip() gmu = _global_music_u() if mu and gmu and hmac.compare_digest(mu, gmu): try: import auth_system await auth_system.add_log('fail', user.username if user else '匿名', ratelimit.client_ip(request), '全局账号冒用拦截', '浏览器提交的 MUSIC_U 与全局账号一致，已按未登录拒绝') except Exception: pass logger.warning('[全局账号冒用拦截] %s 提交的 MUSIC_U 与全局账号一致，已按未登录处理', request.url.path) return {} return ck # ============ 迁移端：专辑 / 歌单 / 歌曲 ============ @router.api_route('/album', methods=['GET', 'POST']) async def album(_user: LoginUser, request: Request): """专辑详情。""" await ratelimit.guard(request, bucket='free', limit=30, window=60) data = await _request_data(request) album_id = data.get('id') if not album_id: return error("参数 'album_id' 不能为空", 400) album_id = _extract_id(album_id) if not str(album_id).strip().isdigit(): return error('无效的专辑ID，请输入数字ID或分享链接') try: result = await music_api.NeteaseAPI().get_album_detail(int(album_id), _cookies()) except APIException: return error('服务暂时不可用，请稍后再试', 500) return success({'status': 200, 'album': result}, '获取专辑详情成功') @router.api_route('/playlist', methods=['GET', 'POST']) async def playlist(user: LoginUser, request: Request): """歌单详情（非管理员遮蔽服务器账号昵称）。""" await ratelimit.guard(request, bucket='free', limit=30, window=60) data = await _request_data(request) playlist_id = data.get('id') if not playlist_id: return error("参数 'playlist_id' 不能为空", 400) playlist_id = _extract_id(playlist_id) if not str(playlist_id).strip().isdigit(): return error('无效的歌单ID，请输入数字ID或分享链接') try: result = await music_api.NeteaseAPI().get_playlist_detail(int(playlist_id), _cookies()) except APIException: return error('服务暂时不可用，请稍后再试', 500) if isinstance(result, dict): if result.get('creator'): result['creator'] = await mask_server_nickname(result['creator'], user) if result.get('name'): result['name'] = await mask_server_nickname(result['name'], user) 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) query = _parse_song_query(data) if isinstance(query, str): return error(query) if data.get('account', 'admin') == 'user': cookies = await _resolve_user_cookies(data, user, request) if not cookies: return error('请先扫码登录个人账号后再使用个人解析', 401) else: cookies = _cookies() api = music_api.NeteaseAPI() try: 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)