# -*- coding: utf-8 -*- """音乐查询端点（FastAPI 栈）——与 Flask 侧同名端点响应契约逐字段对齐。 已迁移（免费解析类，需登录，不扣次数）： - 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 参数获取沿用 Flask 口径：GET query / POST json / POST form 三路兼容。 上游调用走 music_api 的**异步**方法（原生并发，无同步桥）。 """ import asyncio import re from typing import Annotated, Any from fastapi import APIRouter, Depends, Request import music_api from core import repo from core.state import SERVER_NICK_CACHE from music_api import APIException from web import ratelimit from web.deps import SiteUser, require_login from web.response import error, success router = APIRouter() LoginUser = Annotated[SiteUser, Depends(require_login)] _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 _request_data(request: Request) -> dict[str, Any]: """三路兼容取参：GET query 优先，其次 POST JSON，最后表单。""" data: dict[str, Any] = {} if request.method == 'GET': data.update(dict(request.query_params)) return data try: body = await request.json() if isinstance(body, dict): data.update(body) except Exception: pass if not data: try: form = await request.form() data.update(dict(form.items())) except Exception: pass return data def _safe_int(value: Any, default: int) -> int: try: return int(value) except (TypeError, ValueError): return default def _extract_id(raw: Any) -> str: """URL/纯数字 → ID（与 Flask _extract_music_id 的查询参数口径一致）。""" s = str(raw or '').strip() if 'music.163.com' in s: from urllib.parse import parse_qs, urlsplit try: q = parse_qs(urlsplit(s).query) for key in ('id', 'songid'): if q.get(key): return str(q[key][0]).strip() except Exception: pass m = re.search(r'[?&#]id=(\d+)', s) if m: return m.group(1) return s def _cookies() -> dict[str, str]: """全局账号 Cookie（管理员 cookie.txt）。""" return music_api.load_cookies() async def mask_server_nickname(nick: str, user: SiteUser) -> str: """非管理员响应中把服务器账号昵称替换为站点名（与 Flask _mask_server_nickname 同口径）。 - 管理员本人可见真实昵称； - 缓存 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) # 向后兼容字段 + 服务器昵称遮蔽（与 Flask 侧同口径） 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, '搜索完成')