Archived
snowy-yang
22f4d77abd
- web/routers/admin.py：admin/qr/create、admin/qr/check、admin/cookie/status、 admin/cookie/delete、admin/logs - 管理员闸用本地实现，文案与「无 error_code」包体与 Flask _verify_admin_password 逐字对齐（与 auth_system.admin_required 的 forbidden 字段不同） - QRLoginManager await；文件类操作(save_cookie/logout/_get_cookie_info) 走 to_thread - Cookie 变更后同步清 VIP 与全局 MUSIC_U 缓存 契约奇偶校验：36 项全 OK（新增 admin:logs、admin:cookie_status、非管理员 403）； ruff All checks passed；pyright 0 errors；冒烟 10/10。
205 lines
7.6 KiB
Python
205 lines
7.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""管理员扫码登录 / Cookie 管理 / 操作记录（FastAPI 栈）。
|
||
|
||
已迁移：
|
||
- POST /api/v1/admin/qr/create ↔ Flask /admin/qr/create
|
||
- POST /api/v1/admin/qr/check ↔ Flask /admin/qr/check
|
||
- POST /api/v1/admin/cookie/status ↔ Flask /admin/cookie/status
|
||
- POST /api/v1/admin/cookie/delete ↔ Flask /admin/cookie/delete
|
||
- POST /api/v1/admin/logs ↔ Flask /admin/logs
|
||
|
||
鉴权：Flask 这些端点用 _verify_admin_password（文案"需要管理员权限，请使用管理员账号登录"），
|
||
与 auth_system.admin_required 的文案不同，故本模块用本地 admin 闸保持逐字对齐。
|
||
|
||
说明：QRLoginManager 已异步化（直接 await）；QRLoginClient 的 save_cookie/logout/
|
||
_get_cookie_info 仍为同步文件操作，经 asyncio.to_thread 执行。
|
||
"""
|
||
import asyncio
|
||
import logging
|
||
from typing import Annotated, Any
|
||
|
||
from fastapi import APIRouter, Depends, Request
|
||
|
||
from core.state import GLOBAL_MUSIC_U_CACHE, REQ_LOG, VIP_CACHE
|
||
from web import ratelimit
|
||
from web.deps import SiteUser, current_user
|
||
from web.errors import AppError
|
||
from web.response import error, success
|
||
|
||
router = APIRouter()
|
||
|
||
logger = logging.getLogger('web.admin')
|
||
|
||
|
||
async def _require_admin(user: Annotated[SiteUser | None, Depends(current_user)]) -> SiteUser:
|
||
"""管理员闸（文案与 Flask _verify_admin_password 逐字对齐）。"""
|
||
if user is None or not user.is_admin:
|
||
raise AppError('需要管理员权限，请使用管理员账号登录', 403)
|
||
return user
|
||
|
||
|
||
AdminUser = Annotated[SiteUser, Depends(_require_admin)]
|
||
|
||
|
||
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
|
||
|
||
|
||
def _safe_int(value: Any, default: int) -> int:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _admin_client():
|
||
from qr_login import QRLoginClient
|
||
return QRLoginClient()
|
||
|
||
|
||
def _reset_caches() -> None:
|
||
"""Cookie 变更后清 VIP 探测缓存与全局 MUSIC_U 缓存（与 Flask 同口径）。"""
|
||
VIP_CACHE['data'] = None
|
||
VIP_CACHE['ts'] = 0.0
|
||
GLOBAL_MUSIC_U_CACHE['v'] = None
|
||
GLOBAL_MUSIC_U_CACHE['ts'] = 0.0
|
||
|
||
|
||
async def _render_qr_svg(unikey: str) -> str:
|
||
def _build() -> str:
|
||
import qrcode
|
||
from qrcode.image.svg import SvgPathImage
|
||
qr = qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_M) # type: ignore[attr-defined]
|
||
qr.add_data(f'https://music.163.com/login?codekey={unikey}')
|
||
qr.make(fit=True)
|
||
svg = qr.make_image(image_factory=SvgPathImage).to_string()
|
||
return svg.decode('utf-8') if isinstance(svg, bytes) else svg
|
||
|
||
return await asyncio.to_thread(_build)
|
||
|
||
|
||
@router.post('/admin/qr/create')
|
||
async def admin_qr_create(_admin: AdminUser, request: Request):
|
||
"""生成管理员登录二维码（返回 SVG）。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
try:
|
||
client = _admin_client()
|
||
unikey = await client.qr_manager.generate_qr_key()
|
||
except Exception:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|
||
if not unikey:
|
||
return error('生成二维码失败', 500)
|
||
try:
|
||
svg = await _render_qr_svg(unikey)
|
||
except Exception:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|
||
return success({'unikey': unikey, 'qr_svg': svg, 'expires_in': 180}, '二维码生成成功')
|
||
|
||
|
||
@router.post('/admin/qr/check')
|
||
async def admin_qr_check(_admin: AdminUser, request: Request):
|
||
"""轮询二维码登录状态，成功时自动保存 Cookie。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
data = await _request_data(request)
|
||
unikey = data.get('unikey')
|
||
if not unikey:
|
||
return error('缺少 unikey 参数')
|
||
try:
|
||
client = _admin_client()
|
||
code, cookie_dict = await client.qr_manager.check_qr_login(unikey)
|
||
except Exception:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|
||
|
||
if code == 803:
|
||
musu_u = cookie_dict.get('MUSIC_U', '')
|
||
if not musu_u:
|
||
return error('登录成功但未获取到 Cookie', 500)
|
||
cookie = f'MUSIC_U={musu_u};os=pc;appver=8.9.70;'
|
||
if cookie_dict.get('__csrf'):
|
||
cookie += f"__csrf={cookie_dict['__csrf']};"
|
||
saved = await asyncio.to_thread(client.save_cookie, cookie)
|
||
if not saved:
|
||
return error('登录成功但 Cookie 保存失败', 500)
|
||
_reset_caches()
|
||
logger.info('二维码登录成功，Cookie 已自动保存')
|
||
info = await asyncio.to_thread(client._get_cookie_info)
|
||
return success({'status': 'success', 'cookie_info': info}, '登录成功，Cookie 已保存')
|
||
if code == 801:
|
||
return success({'status': 'waiting'}, '等待扫码')
|
||
if code == 802:
|
||
return success({'status': 'scanned'}, '已扫码，请在手机上确认登录')
|
||
if code == 800:
|
||
return success({'status': 'expired'}, '二维码已过期')
|
||
return success({'status': 'error', 'code': code}, f'登录状态异常（{code}）')
|
||
|
||
|
||
@router.post('/admin/cookie/status')
|
||
async def admin_cookie_status(_admin: AdminUser, request: Request):
|
||
"""获取当前 Cookie 状态。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
try:
|
||
client = _admin_client()
|
||
info = await asyncio.to_thread(client._get_cookie_info)
|
||
return success(info, '获取 Cookie 状态成功')
|
||
except Exception:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|
||
|
||
|
||
@router.post('/admin/cookie/delete')
|
||
async def admin_cookie_delete(_admin: AdminUser, request: Request):
|
||
"""删除（清除）Cookie，备份后删除，可重新扫码登录。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
try:
|
||
client = _admin_client()
|
||
ok = await asyncio.to_thread(client.logout)
|
||
if not ok:
|
||
info = await asyncio.to_thread(client._get_cookie_info)
|
||
return error(
|
||
f"Cookie 清除失败（文件仍存在：{info.get('file_exists')}，"
|
||
f"可能被占用或为挂载点，请手动删除）", 500)
|
||
_reset_caches()
|
||
logger.info('管理员已清除 Cookie')
|
||
info = await asyncio.to_thread(client._get_cookie_info)
|
||
return success(info, 'Cookie 已清除')
|
||
except Exception:
|
||
return error('服务暂时不可用，请稍后再试', 500)
|
||
|
||
|
||
@router.post('/admin/logs')
|
||
async def admin_logs(_admin: AdminUser, request: Request):
|
||
"""用户 IP 解析/下载记录（分页，最新在前）。"""
|
||
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
||
data = await _request_data(request)
|
||
try:
|
||
page = max(1, _safe_int(data.get('page'), 1))
|
||
page_size = max(1, min(50, _safe_int(data.get('page_size'), 10)))
|
||
except ValueError:
|
||
return error('参数格式错误')
|
||
logs = list(reversed(REQ_LOG))
|
||
total = len(logs)
|
||
start = (page - 1) * page_size
|
||
items = logs[start:start + page_size]
|
||
return success({
|
||
'total': total,
|
||
'page': page,
|
||
'page_size': page_size,
|
||
'pages': max(1, (total + page_size - 1) // page_size),
|
||
'items': items,
|
||
}, '获取操作记录成功')
|