Archived
snowy-yang
6c3181341b
迁移路由（/api/v1/*，与 Flask 同端点逐字段对齐）： - account/vip：admin/user 双路径 + 5 分钟缓存（降级结果 30s）， VIP 探测单飞锁与 Flask 同源跨栈共享；未登录可访问（同 Flask） - global/playlists、global/playlist：管理员闸（require_admin，403 对齐） - web/deps.py 新增 require_admin；_resolve_user_cookies 支持可选用户（匿名审计保底） 测试：46/46（新增 10 项鉴权依赖：有效/管理员/封禁/禁用/顶号/伪造 Cookie、 401/403 闸）；ruff All checks passed；pyright 0 errors。 契约奇偶校验：16 项全 OK（新增 account:vip、global:playlists/playlist、 非管理员 403）；uvicorn 实跑双栈 account/vip 均 200、未授权歌单 403。
182 lines
7.7 KiB
Python
182 lines
7.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""账号与全局歌单端点（FastAPI 栈）——与 Flask 侧同名端点响应契约逐字段对齐。
|
||
|
||
已迁移：
|
||
- POST/GET /api/v1/account/vip ↔ Flask /account/vip（admin/user 双路径 + 缓存）
|
||
- POST/GET /api/v1/global/playlists ↔ Flask /global/playlists（管理员）
|
||
- POST/GET /api/v1/global/playlist ↔ Flask /global/playlist（管理员）
|
||
"""
|
||
import hashlib
|
||
import time
|
||
from typing import Annotated, Any
|
||
|
||
from fastapi import APIRouter, Depends, Request
|
||
|
||
import music_api
|
||
from core.state import VIP_CACHE, VIP_PROBE_LOCK, VIP_USER_CACHE
|
||
from music_api import APIException
|
||
from web import ratelimit
|
||
from web.deps import SiteUser, current_user, require_admin
|
||
from web.response import error, success
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
AdminUser = Annotated[SiteUser, Depends(require_admin)]
|
||
|
||
_VIP_NAMES = {
|
||
0: '普通用户', 1: '音乐包', 11: '黑胶VIP', 100: '黑胶VIP', 10: '黑胶SVIP', 110: '黑胶SVIP',
|
||
}
|
||
|
||
# 探测单飞：与 Flask _VIP_PROBE_LOCK 同源（跨栈共享，避免两栈各自串行 8 级探测）
|
||
_VIP_PROCESS_LOCK: Any = VIP_PROBE_LOCK
|
||
|
||
|
||
def _vip_name(vip_type: int) -> str:
|
||
return _VIP_NAMES.get(vip_type, '普通用户')
|
||
|
||
|
||
def _cookies() -> dict[str, str]:
|
||
return music_api.load_cookies()
|
||
|
||
|
||
async def _detect_vip(cookies: dict[str, str]) -> tuple[int, str, list[str]]:
|
||
"""检测账号会员类型与可解析音质（与 Flask _detect_vip 同口径）。
|
||
|
||
单飞锁为 threading.Lock：异步上下文里用 to_thread 有界等待 0.5s，
|
||
拿不到锁的请求直接用账号接口给降级结果（不等 8 级探测）。
|
||
"""
|
||
import asyncio
|
||
|
||
api = music_api.NeteaseAPI()
|
||
acquired = await asyncio.to_thread(_VIP_PROCESS_LOCK.acquire, True, 0.5)
|
||
if not acquired:
|
||
info = await api.get_account_info(cookies)
|
||
if info:
|
||
return info.get('vipType', 0), 'account', []
|
||
return 0, 'probe', []
|
||
try:
|
||
probe = await api.detect_vip_by_probe(cookies)
|
||
levels_ok = probe.get('levels_ok', [])
|
||
info = await api.get_account_info(cookies)
|
||
if info:
|
||
return info.get('vipType', 0), 'account', levels_ok
|
||
return probe.get('vip_type', 0), 'probe', levels_ok
|
||
finally:
|
||
_VIP_PROCESS_LOCK.release()
|
||
|
||
|
||
async def _request_data(request: Request) -> dict[str, Any]:
|
||
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
|
||
|
||
|
||
@router.api_route('/account/vip', methods=['GET', 'POST'])
|
||
async def account_vip(
|
||
user: Annotated[SiteUser | None, Depends(current_user)],
|
||
request: Request,
|
||
):
|
||
"""当前解析账号的 VIP 类型（admin=全局账号 / user=个人账号）。无需登录。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
data = await _request_data(request)
|
||
account = data.get('account', 'admin')
|
||
api = music_api.NeteaseAPI()
|
||
|
||
try:
|
||
if account == 'user':
|
||
from web.routers.music import _resolve_user_cookies
|
||
user_cookies = await _resolve_user_cookies(data, user, request)
|
||
if not user_cookies:
|
||
return success({'account': 'user', 'logged_in': False, 'vip_type': None,
|
||
'vip_name': '未登录', 'levels_ok': []}, '个人账号未登录')
|
||
ck = '|'.join(f'{k}={v}' for k, v in sorted(user_cookies.items()))
|
||
ck_hash = hashlib.md5(ck.encode('utf-8')).hexdigest()
|
||
cached_user = VIP_USER_CACHE.get(ck_hash)
|
||
if cached_user is not None and time.time() - cached_user['ts'] <= cached_user.get('ttl', 300):
|
||
return success(dict(cached_user['data'], account='user'), '获取个人账号信息成功（缓存）')
|
||
vip_type, method, levels_ok = await _detect_vip(user_cookies)
|
||
info = await api.get_account_info(user_cookies)
|
||
user_payload = {
|
||
'account': 'user', 'logged_in': True, 'vip_type': vip_type,
|
||
'vip_name': _vip_name(vip_type), 'detected_by': method, 'levels_ok': levels_ok,
|
||
'nickname': (info or {}).get('nickname', ''),
|
||
}
|
||
# 降级结果（levels_ok 空）短缓存 30s，正常结果 300s（与 Flask 同口径）
|
||
ttl = 30 if not levels_ok else 300
|
||
if len(VIP_USER_CACHE) > 500:
|
||
VIP_USER_CACHE.clear()
|
||
VIP_USER_CACHE[ck_hash] = {'ts': time.time(), 'ttl': ttl, 'data': user_payload}
|
||
return success(user_payload, '获取个人账号信息成功')
|
||
|
||
cached = VIP_CACHE['data']
|
||
if cached is None or time.time() - VIP_CACHE['ts'] > VIP_CACHE.get('ttl', 300):
|
||
admin_cookies = _cookies()
|
||
vip_type, method, levels_ok = await _detect_vip(admin_cookies)
|
||
info = await api.get_account_info(admin_cookies)
|
||
cached = {
|
||
'vip_type': vip_type, 'vip_name': _vip_name(vip_type),
|
||
'detected_by': method, 'levels_ok': levels_ok,
|
||
'nickname': (info or {}).get('nickname', ''),
|
||
'logged_in': info is not None,
|
||
'note': '' if info else '账号接口未返回会员信息，已通过歌曲音质探测接口检测',
|
||
}
|
||
VIP_CACHE['data'] = cached
|
||
VIP_CACHE['ts'] = time.time()
|
||
VIP_CACHE['ttl'] = 30 if not levels_ok else 300
|
||
payload = dict(cached)
|
||
payload['account'] = 'admin'
|
||
payload['nickname'] = cached['nickname'] if (user and user.is_admin) else ''
|
||
return success(payload, '获取全局账号信息成功')
|
||
except Exception:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|
||
|
||
|
||
@router.api_route('/global/playlists', methods=['GET', 'POST'])
|
||
async def global_playlists(_admin: AdminUser, request: Request):
|
||
"""全局账号收藏/创建的歌单（含「我喜欢的音乐」）。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
api = music_api.NeteaseAPI()
|
||
cookies = _cookies()
|
||
try:
|
||
account = await api.get_account_info(cookies)
|
||
if not account or not account.get('userId'):
|
||
return error('全局账号未登录或 Cookie 已失效，请先在「管理员设置」扫码登录', 401)
|
||
playlists = await api.get_user_playlists(account['userId'], cookies)
|
||
return success({'user': account, 'playlists': playlists}, '获取全局账号歌单成功')
|
||
except APIException:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|
||
|
||
|
||
@router.api_route('/global/playlist', methods=['GET', 'POST'])
|
||
async def global_playlist(_admin: AdminUser, request: Request):
|
||
"""全局账号歌单详情（全局 Cookie，支持「我喜欢的音乐」等私有歌单）。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
from web.routers.music import _extract_id
|
||
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())
|
||
return success({'playlist': result}, '获取全局账号歌单详情成功')
|
||
except APIException:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|