Archived
snowy-yang
6c7f8f589e
web/routers/account.py 的音质探测单飞锁原本是 core.state 里的 threading.Lock， 在协程里用 `await asyncio.to_thread(lock.acquire, True, 0.5)` 做有界等待，每次 VIP 请求都白付一次线程池调度。 为什么可以直接换：这把锁**只在协程里用**。原注释写「与 core.state 中同一把锁 共享」，但 grep 全仓只有 _detect_vip 一个消费者（sync 时代另有一个线程侧使用者， 早已不存在）。改为 asyncio.Lock + asyncio.wait_for(acquire(), 0.5)，超时分支用 `except TimeoutError` 走「不等 8 级探测、直接用账号接口降级」的原有语义。 顺带把锁从 core.state 移出：core.state 自己声明「依赖仅限标准库 threading/collections」，而这把锁是 asyncio 的。现在它跟着唯一消费者走。 锁按事件循环隔离（以 id(loop) 为键）：asyncio.Lock 绑定创建它的 loop，应用主 loop 与测试各自的 loop 并存，跨 loop 复用会抛 "bound to a different event loop" （B2 迁移 auth_system 时踩过同一个坑）。 门禁：ruff 全过 / pyright 0-0-0 / pytest 382 项全绿（0 失败）。
104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""进程内运行时状态（框架无关）。
|
||
|
||
自原单体模块迁出：这些容器全部**原地修改**（永不重新绑定），因此
|
||
`from core.state import DOWNLOAD_TASKS` 与各模块的引用共享同一对象，
|
||
搬移不改变任何并发行为。
|
||
|
||
注意：这些状态**仅存于进程内存**，重启即丢（任务记录/限流桶/统计/缓存）；
|
||
持久数据（账本/用户/设置）在 SQLite（auth_system）。集中在此以便
|
||
web/ 与各服务模块共用，并为后续封装成带锁的单例（State 对象）留出唯一入口。
|
||
|
||
依赖仅限标准库 threading/collections——任何下载器/HTTP 相关对象
|
||
（如批量打包信号量）不放在这里，随对应服务模块搬迁。
|
||
"""
|
||
|
||
import collections
|
||
import threading
|
||
from typing import Any, NotRequired, TypedDict
|
||
|
||
# ============ 短链展开结果缓存 ============
|
||
# 用户反复解析同一分享链接不再重复请求 163cn.tv（有界防膨胀）
|
||
SHORTLINK_CACHE: dict[str, str] = {}
|
||
SHORTLINK_CACHE_LOCK = threading.Lock()
|
||
|
||
|
||
# ============ 公网限流（按 IP 滑动窗口，防刷防爆破） ============
|
||
RATE_BUCKETS = {}
|
||
RATE_LOCK = threading.Lock()
|
||
|
||
|
||
# ============ 批量打包排队估算 ============
|
||
# 最近完成批次（成功打包过）的耗时记录，用于估算排队用户的预计等待时间
|
||
BATCH_DURATIONS = collections.deque(maxlen=10)
|
||
BATCH_DURATIONS_LOCK = threading.Lock()
|
||
|
||
|
||
# ============ 下载任务表 ============
|
||
# （批量打包 + 单曲下载统一）：task_id -> {type, total, done, failed, status, start, msg, cancel, pause, device_id, result, params}
|
||
# 前端通过 /api/batch_progress 轮询进度，支持取消；完成后文件按 35 分钟保留期清理
|
||
DOWNLOAD_TASKS: dict[str, dict[str, Any]] = {}
|
||
DOWNLOAD_TASKS_LOCK = threading.Lock()
|
||
|
||
|
||
# ============ 设备在线心跳 ============
|
||
# device_id -> 最近心跳时间。心跳超时（PAUSE_AFTER_SECONDS）未上线的设备，其批量任务
|
||
# 自动暂停让出排队名额；超时超过 CANCEL_AFTER_SECONDS（30 分钟）则自动取消任务并删文件
|
||
DEVICE_LAST_SEEN: dict[str, float] = {}
|
||
DEVICE_LAST_SEEN_LOCK = threading.Lock()
|
||
|
||
|
||
# ============ 接口请求统计 ============
|
||
REQUEST_STATS: dict[str, dict[str, Any]] = {}
|
||
STATS_LOCK = threading.Lock()
|
||
|
||
# 用户解析/下载操作记录（IP）；仅保留最近 2000 条
|
||
REQ_LOG: list[dict[str, Any]] = []
|
||
|
||
|
||
# ============ 上游探测缓存 ============
|
||
# 带时间戳的缓存一律用 TypedDict 显式声明字段类型：无标注的 dict 字面量会被类型
|
||
# 检查器推断为「所有键的值类型取并集」（如 `{'v': None, 'ts': 0.0}` → float | None），
|
||
# 于是每个读写点都退化成联合类型，既丢失精确性也淹没真实缺陷。
|
||
class _TimedValue(TypedDict):
|
||
"""值 + 写入时刻的二元缓存。"""
|
||
v: str | None
|
||
ts: float
|
||
|
||
|
||
class _NickCache(TypedDict):
|
||
"""服务器账号昵称缓存。"""
|
||
nick: str | None
|
||
ts: float
|
||
|
||
|
||
class _VipCache(TypedDict):
|
||
"""全局 VIP 缓存：data 为 None 表示尚未探测（读侧需判空）。"""
|
||
data: dict[str, Any] | None
|
||
ts: float
|
||
ttl: NotRequired[int]
|
||
|
||
|
||
class _VipEntry(TypedDict):
|
||
"""已写入的 VIP 缓存条目——三个字段齐备，data 必定存在。"""
|
||
data: dict[str, Any]
|
||
ts: float
|
||
ttl: int
|
||
|
||
|
||
# 全局账号 MUSIC_U（5 分钟缓存；后台重新扫码后自动跟随新值）
|
||
GLOBAL_MUSIC_U_CACHE: _TimedValue = {'v': None, 'ts': 0.0}
|
||
|
||
# 服务器网易云账号昵称缓存（10 分钟）：非管理员可见响应中遮蔽真实昵称
|
||
SERVER_NICK_CACHE: _NickCache = {'nick': None, 'ts': 0.0}
|
||
SERVER_NICK_LOCK = threading.Lock() # 刷新单飞锁，防并发重复打上游
|
||
# 注：音质探测的单飞锁（原 VIP_PROBE_LOCK）已移出本模块——本模块声明「依赖仅限
|
||
# 标准库 threading/collections」，而那是把 asyncio 锁；且它全仓只有
|
||
# web/routers/account.py 一个消费者（sync 时代遗留的「共享」注释已不成立）。
|
||
|
||
# 全局账号 VIP 探测缓存（5 分钟）：probe 一次要打 8 次网易云接口，
|
||
# 公网下账号栏高频刷新会连累服务器账号触发风控。
|
||
VIP_CACHE: _VipCache = {'data': None, 'ts': 0.0}
|
||
# user 路径按 cookie 哈希的 5 分钟缓存
|
||
VIP_USER_CACHE: dict[str, _VipEntry] = {}
|