终端 header 显示当前在线人数

- src/utils/online_tracker.py:5 分钟滑动窗口,thread-safe 内存追踪
- web/core.py:认证成功时 record_activity(user_id)
- web/routers/ops.py:GET /api/ops/online-users 返回 {online: N}
- 前端 header 每 60s 轮询,Users 图标 + 数字
This commit is contained in:
2569718930@qq.com
2026-05-26 08:40:17 +08:00
parent 23203f6ebb
commit c88d313f60
4 changed files with 73 additions and 0 deletions
+43
View File
@@ -0,0 +1,43 @@
"""Lightweight online-user tracker. Records last-seen timestamps in memory
and exposes a count of users active within a sliding window. No external
dependency — just a dict + lock + periodic cleanup thread."""
from __future__ import annotations
import threading
import time
from typing import Dict
_online_users: Dict[str, float] = {}
_lock = threading.Lock()
_cleanup_interval_sec = 120
_window_sec = 300 # 5 minutes
def _cleanup_stale() -> None:
cutoff = time.time() - _window_sec
with _lock:
stale = [uid for uid, ts in _online_users.items() if ts < cutoff]
for uid in stale:
del _online_users[uid]
def _start_cleanup() -> None:
_cleanup_stale()
threading.Timer(_cleanup_interval_sec, _start_cleanup).start()
_start_cleanup()
def record_activity(user_id: str) -> None:
if not user_id:
return
with _lock:
_online_users[user_id] = time.time()
def online_count() -> int:
_cleanup_stale()
with _lock:
return len(_online_users)