Archived
snowy-yang
cd0f1f78a3
补齐最后缺口： - POST /api/v1/auth/admin_login/email、/auth/admin_login/verify（管理员 2FA Step2/3）， 复用 auth_routes 的票据表与尝试上限（跨栈共享），签发会话 Cookie 重构（消除约 200 行重复，11 个 router 受益）： - web/request_utils.py：request_data / safe_int / extract_id / cookies / dedup_ids 原在各 router 各复制一份，现集中 - web/serializers.py：format_credits / user_public / record_req_log - 各 router 改为导入别名，行为逐字不变 覆盖率：Flask 90 条业务路由全部有 FastAPI 等价端点（含 5 个向后兼容别名）。 验证：pytest 69/69；ruff All checks passed；pyright 0 errors；奇偶 36/36；冒烟 10/10。
37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""响应序列化与记录辅助（消除 router 间重复定义）。"""
|
|
import time
|
|
from typing import Any
|
|
|
|
from core.state import REQ_LOG
|
|
|
|
# 站点用户公开字段（与 Flask auth_system._user_public 的非 full 口径一致）
|
|
_PUBLIC_FIELDS = ('id', 'username', 'email', 'role', 'duration_seconds',
|
|
'created_at', 'last_login', 'ann_seen')
|
|
|
|
|
|
def format_credits(count: int) -> str:
|
|
"""次数文案（与 Flask auth_system.format_credits 逐字一致）。"""
|
|
if count < 0:
|
|
return '不限次数'
|
|
if count <= 0:
|
|
return '已用完'
|
|
return f'{count} 次'
|
|
|
|
|
|
def user_public(row: dict[str, Any]) -> dict[str, Any]:
|
|
"""用户公开字段（未登录/缺失字段安全）。"""
|
|
return {k: row.get(k) for k in _PUBLIC_FIELDS}
|
|
|
|
|
|
def record_req_log(action: str, ip: str, song: str) -> None:
|
|
"""用户解析/下载操作记录（写入共享 REQ_LOG，保留最近 2000 条）。"""
|
|
REQ_LOG.append({
|
|
'action': action,
|
|
'ip': ip,
|
|
'song': str(song or '')[:100],
|
|
'time': time.strftime('%Y-%m-%d %H:%M:%S'),
|
|
})
|
|
if len(REQ_LOG) > 2000:
|
|
del REQ_LOG[:len(REQ_LOG) - 2000]
|