JOJO
9786eec86e
拆包时 resources.py 独立 import 头漏了 get_current_user_role：host 模式 + 非显式身份 + 存在 host 用户记录时触发 NameError，被 except 吞掉后工具分类限制与禁用模型等管理员策略静默不应用（默认无 host 记录不触发）。 - resources.py/usage.py 顺带补 TYPE_CHECKING 导入净化字符串类型注解，server/context/ 与 server/runtime/ pyflakes F821 清零 - 新增回归测试：host 模式 + HTTP 请求上下文 + host 记录存在，断言策略真正应用到 terminal；红绿验证通过（回退导入测试即红） 测试：25/25 全绿（24 + 新增 1）
41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
"""配额追踪器（UsageTracker）的获取与广播。"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional, TYPE_CHECKING
|
||
|
||
if TYPE_CHECKING:
|
||
import modules.user_manager
|
||
|
||
from modules.usage_tracker import UsageTracker
|
||
from server import state
|
||
|
||
|
||
def get_or_create_usage_tracker(username: Optional[str], workspace: Optional['modules.user_manager.UserWorkspace'] = None) -> Optional[UsageTracker]:
|
||
if not username:
|
||
return None
|
||
tracker = state.usage_trackers.get(username)
|
||
if tracker:
|
||
return tracker
|
||
from modules.user_manager import UserWorkspace # noqa: F401 # 保持类型引用兼容
|
||
if workspace is None:
|
||
workspace = state.user_manager.ensure_user_workspace(username)
|
||
record = state.user_manager.get_user(username)
|
||
role = getattr(record, "role", "user") if record else "user"
|
||
tracker = UsageTracker(str(workspace.data_dir), role=role or "user")
|
||
state.usage_trackers[username] = tracker
|
||
return tracker
|
||
|
||
|
||
def emit_user_quota_update(username: Optional[str]):
|
||
from server.extensions import socketio
|
||
if not username:
|
||
return
|
||
tracker = get_or_create_usage_tracker(username)
|
||
if not tracker:
|
||
return
|
||
try:
|
||
snapshot = tracker.get_quota_snapshot()
|
||
socketio.emit('quota_update', {'quotas': snapshot}, room=f"user_{username}")
|
||
except Exception:
|
||
pass
|