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。
84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""路由公共工具：请求取参、ID 归一化、Cookie、安全转换。
|
|
|
|
这些函数原先在 11 个 router 模块里各复制一份（约 200 行重复），
|
|
现集中于此，行为与各副本完全一致（原逻辑逐字保留）。
|
|
"""
|
|
import re
|
|
from typing import Any
|
|
|
|
from fastapi import Request
|
|
|
|
# music.163.com 链接 / 纯数字 → ID（与 Flask MusicAPIService._extract_music_id 的
|
|
# 查询参数口径一致；短链展开由调用方按需先做）
|
|
_ID_QUERY_RE = re.compile(r'[?&#]id=(\d+)')
|
|
|
|
|
|
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:
|
|
"""用户输入的安全整数解析：垃圾值回退默认值（而非 500）。"""
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def extract_id(raw: Any) -> str:
|
|
"""URL / 分享链接 / 纯数字 → 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 = _ID_QUERY_RE.search(s)
|
|
if m:
|
|
return m.group(1)
|
|
return s
|
|
|
|
|
|
def cookies() -> dict[str, str]:
|
|
"""全局账号 Cookie（管理员 cookie.txt）。"""
|
|
import music_api
|
|
return music_api.load_cookies()
|
|
|
|
|
|
def dedup_ids(raw_ids: Any, transform=extract_id) -> list[str]:
|
|
"""ID 列表去重（保持首次出现顺序）：与 Flask 侧「提取→去重」计费口径一致。"""
|
|
if isinstance(raw_ids, str):
|
|
raw_list = raw_ids.split(',')
|
|
else:
|
|
raw_list = [str(x) for x in (raw_ids or [])]
|
|
out: list[str] = []
|
|
seen: set[str] = set()
|
|
for x in raw_list:
|
|
s = str(transform(x) or '').strip()
|
|
if s and s not in seen:
|
|
seen.add(s)
|
|
out.append(s)
|
|
return out
|