中控总览改为后台常驻并行拉取热缓存,浏览器只读缓存并经 SSE 更新。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-07 14:24:26 +08:00
parent 1735c031ea
commit 0e3fffef44
5 changed files with 327 additions and 36 deletions
+2
View File
@@ -7,6 +7,8 @@ FLASK_SECRET_KEY=change-me-to-long-random
DB_PATH=hub.db
HUB_PULL_INTERVAL_SEC=10
HUB_PULL_TIMEOUT_SEC=5
# 总览 SSE 心跳(秒);浏览器只读缓存,靠 SSE 收版本更新
HUB_SSE_HEARTBEAT_SEC=25
# 中控企业微信(离线/故障告警,与客户端通知分开)
HUB_WECHAT_ENABLED=false
+39 -23
View File
@@ -3,8 +3,6 @@ from __future__ import annotations
import os
import sys
import threading
import time
from functools import wraps
from pathlib import Path
@@ -19,6 +17,7 @@ from flask import (
request,
send_file,
session,
stream_with_context,
url_for,
)
@@ -62,6 +61,7 @@ from lib.instances_lib import (
list_instances,
update_instance,
)
from lib.overview_cache_lib import overview_store
from lib.sso_lib import build_embed_open_url
from lib.stats_agg_lib import build_stats_curve, build_stats_summary, normalize_window
from lib.wechat_lib import (
@@ -486,38 +486,54 @@ def desk():
)
def _on_overview_pulled(rows: list) -> None:
"""常驻 pull 回调:健康巡检 + 余额采样(与浏览器是否打开无关)."""
try:
inspect_overview_rows(rows)
except Exception:
pass
try:
record_cash_samples_from_overview(rows)
except Exception:
pass
@app.route("/api/overview")
@login_required
def api_overview():
rows = pull_all_enabled()
return jsonify({"ok": True, "instances": rows})
"""只读热缓存;不在请求内现场拉实例(后台常驻并行 pull)."""
force = (request.args.get("force") or "").strip().lower() in ("1", "true", "yes")
if force:
overview_store.request_refresh()
return jsonify(overview_store.snapshot_dict())
def _health_pull_interval_sec() -> float:
try:
return max(5.0, float(os.getenv("HUB_PULL_INTERVAL_SEC") or 10))
except (TypeError, ValueError):
return 10.0
@app.route("/api/overview/refresh", methods=["POST"])
@login_required
def api_overview_refresh():
overview_store.request_refresh()
return jsonify({"ok": True, "msg": "已触发后台刷新"})
def _health_watch_loop() -> None:
"""后台巡检:即使无人打开总览也会告警;并采样余额曲线."""
while True:
try:
rows = pull_all_enabled()
inspect_overview_rows(rows)
record_cash_samples_from_overview(rows)
except Exception:
pass
time.sleep(_health_pull_interval_sec())
@app.route("/api/overview/stream")
@login_required
def api_overview_stream():
return Response(
stream_with_context(overview_store.iter_sse()),
mimetype="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
def start_health_watch() -> None:
t = threading.Thread(target=_health_watch_loop, name="hub-health-watch", daemon=True)
t.start()
def start_overview_cache() -> None:
overview_store.start(pull_all_enabled, on_pull=_on_overview_pulled)
start_health_watch()
start_overview_cache()
try:
start_hub_backup_scheduler(
+197
View File
@@ -0,0 +1,197 @@
"""中控总览热缓存:后台常驻并行 pull,与浏览器是否打开无关;SSE 仅扇出版本通知."""
from __future__ import annotations
import json
import os
import queue
import threading
import time
from collections.abc import Callable, Iterator
from typing import Any
OnPullFn = Callable[[list[dict[str, Any]]], None]
def _poll_interval_sec() -> float:
try:
return max(3.0, float(os.getenv("HUB_PULL_INTERVAL_SEC") or 10))
except (TypeError, ValueError):
return 10.0
def _sse_heartbeat_sec() -> float:
try:
return max(10.0, float(os.getenv("HUB_SSE_HEARTBEAT_SEC") or 25))
except (TypeError, ValueError):
return 25.0
class OverviewStore:
def __init__(self) -> None:
self._lock = threading.RLock()
self.version = 0
self.instances: list[dict[str, Any]] = []
self.pulling = True # 启动至首轮完成前视为拉取中
self.last_error: str | None = None
self.updated_at: str = ""
self._ready = False
self._subscribers: list[queue.Queue[str | None]] = []
self._stop = threading.Event()
self._refresh = threading.Event()
self._thread: threading.Thread | None = None
self._pull_fn: Callable[[], list[dict[str, Any]]] | None = None
self._on_pull: OnPullFn | None = None
def start(
self,
pull_fn: Callable[[], list[dict[str, Any]]],
*,
on_pull: OnPullFn | None = None,
) -> None:
self._pull_fn = pull_fn
self._on_pull = on_pull
if self._thread and self._thread.is_alive():
return
self._stop.clear()
with self._lock:
self.pulling = True
self._ready = False
self._thread = threading.Thread(
target=self._loop,
daemon=True,
name="hub-overview-pull",
)
self._thread.start()
def stop(self) -> None:
self._stop.set()
self._refresh.set()
self._broadcast(close=True)
def request_refresh(self) -> None:
self._refresh.set()
def snapshot_dict(self) -> dict[str, Any]:
with self._lock:
rows = list(self.instances)
ver = self.version
pulling = self.pulling
err = self.last_error
updated_at = self.updated_at
return {
"ok": True,
"instances": rows,
"overview_version": ver,
"pulling": pulling,
"updated_at": updated_at,
"error": err,
"poll_interval_sec": _poll_interval_sec(),
"from_cache": True,
}
def event_dict(self) -> dict[str, Any]:
with self._lock:
return {
"overview_version": self.version,
"updated_at": self.updated_at,
"pulling": self.pulling,
"ok": True,
"error": self.last_error,
"poll_interval_sec": _poll_interval_sec(),
}
def _loop(self) -> None:
assert self._pull_fn is not None
while not self._stop.is_set():
self._pull_once()
if self._stop.is_set():
break
self._refresh.clear()
self._refresh.wait(timeout=_poll_interval_sec())
def _pull_once(self) -> None:
assert self._pull_fn is not None
with self._lock:
self.pulling = True
self._broadcast()
try:
rows = self._pull_fn()
if not isinstance(rows, list):
rows = []
if self._on_pull is not None:
try:
self._on_pull(rows)
except Exception:
pass
with self._lock:
self.version += 1
self.instances = rows
self.updated_at = time.strftime("%Y-%m-%d %H:%M:%S")
self.last_error = None
self.pulling = False
self._ready = True
except Exception as e:
with self._lock:
self.version += 1
self.last_error = str(e) or "pull_failed"
self.pulling = False
self._ready = True
# 保留上一轮成功缓存,避免整页空白
self._broadcast()
def _broadcast(self, *, close: bool = False) -> None:
with self._lock:
subs = list(self._subscribers)
event = None if close else json.dumps(self.event_dict(), ensure_ascii=False)
dead: list[queue.Queue[str | None]] = []
for q in subs:
try:
q.put_nowait(None if close else event)
except queue.Full:
try:
q.get_nowait()
except queue.Empty:
pass
try:
q.put_nowait(event)
except queue.Full:
dead.append(q)
except Exception:
dead.append(q)
if dead:
with self._lock:
for q in dead:
if q in self._subscribers:
self._subscribers.remove(q)
def iter_sse(self) -> Iterator[str]:
q: queue.Queue[str | None] = queue.Queue(maxsize=32)
with self._lock:
self._subscribers.append(q)
try:
yield _sse_frame(self.event_dict())
while True:
try:
raw = q.get(timeout=_sse_heartbeat_sec())
except queue.Empty:
yield ": heartbeat\n\n"
continue
if raw is None:
break
try:
data = json.loads(raw)
except Exception:
data = self.event_dict()
yield _sse_frame(data)
finally:
with self._lock:
if q in self._subscribers:
self._subscribers.remove(q)
def _sse_frame(data: dict[str, Any]) -> str:
body = json.dumps(data, ensure_ascii=False)
return f"event: overview\ndata: {body}\n\n"
overview_store = OverviewStore()
+88 -12
View File
@@ -581,7 +581,7 @@
if (dd.submitted_sheets != null) msg += " · " + dd.submitted_sheets + " 张";
if (dd.fully_closed) msg += " · 已全平";
alert(msg);
await refresh();
await requestOverviewRefresh();
} else {
alert((dd && dd.msg) || "平仓失败");
btn.disabled = false;
@@ -759,21 +759,34 @@
return best;
}
async function refresh() {
let overviewVersion = 0;
let overviewLoading = false;
let overviewEventSource = null;
let overviewReconnectTimer = null;
async function refresh(opts) {
opts = opts || {};
const silent = !!opts.silent;
const status = document.getElementById("overview-status");
const grid = document.getElementById("overview-grid");
if (!grid) return;
if (status) status.textContent = "刷新中…";
if (overviewLoading) return;
overviewLoading = true;
if (status && !silent) status.textContent = "读取缓存…";
try {
const r = await fetch("/api/overview");
// 只读中控热缓存;后台常驻并行 pull,不在此请求内打实例
const r = await fetch("/api/overview", { credentials: "same-origin" });
const d = await r.json();
if (!d.ok) throw new Error(d.msg || "失败");
if (!d.ok) throw new Error(d.msg || d.error || "失败");
overviewVersion = Number(d.overview_version) || overviewVersion;
const list = sortInstancesByMode(d.instances || []);
renderOverviewSummary(list);
if (global.HubQuotes && HubQuotes.paint) HubQuotes.paint(list);
if (!list.length) {
grid.innerHTML =
'<p class="muted">暂无启用实例,请先到「实例」添加.</p>';
d.pulling || !d.updated_at
? '<p class="muted">后台首次拉取中…</p>'
: '<p class="muted">暂无启用实例,请先到「实例」添加.</p>';
} else if (isHubPhone()) {
grid.innerHTML =
'<div class="hub-phone-cards">' + list.map(renderPhoneCard).join("") + "</div>";
@@ -790,15 +803,20 @@
}
if (status) {
const s = summarizeOverview(list);
const pulled = latestPulledAt(list);
const pulled = latestPulledAt(list) || d.updated_at || "";
const sec = Number(d.poll_interval_sec) || 10;
const live = overviewEventSource && overviewEventSource.readyState === 1;
status.textContent =
"已更新 · 共 " +
(live ? "SSE 已连接 · " : "") +
"缓存 " +
list.length +
" 台 · 在线 " +
s.all.online +
" · 离线 " +
s.all.offline +
(pulled ? " · 拉取时间 " + pulled : "");
(pulled ? " · " + pulled : "") +
(d.pulling ? " · 后台拉取中" : " · 后台每 " + sec + "s 并行拉") +
(d.error ? " · " + d.error : "");
}
} catch (e) {
if (status) status.textContent = e.message || "加载失败";
@@ -808,9 +826,60 @@
box.innerHTML = "";
}
if (global.HubQuotes && HubQuotes.paint) HubQuotes.paint([]);
} finally {
overviewLoading = false;
}
}
function closeOverviewStream() {
if (overviewEventSource) {
overviewEventSource.close();
overviewEventSource = null;
}
if (overviewReconnectTimer) {
clearTimeout(overviewReconnectTimer);
overviewReconnectTimer = null;
}
}
function connectOverviewStream() {
const grid = document.getElementById("overview-grid");
if (!grid || typeof EventSource === "undefined") return;
closeOverviewStream();
overviewEventSource = new EventSource("/api/overview/stream");
overviewEventSource.addEventListener("overview", function (ev) {
try {
const st = JSON.parse(ev.data || "{}");
const ver = Number(st.overview_version) || 0;
if (ver && ver !== overviewVersion) {
refresh({ silent: true });
} else if (st.pulling) {
const status = document.getElementById("overview-status");
if (status) status.textContent = "后台并行拉取中…";
}
} catch (_) {}
});
overviewEventSource.onerror = function () {
closeOverviewStream();
const status = document.getElementById("overview-status");
if (status) status.textContent = "SSE 断开,8s 后重连…";
overviewReconnectTimer = setTimeout(function () {
connectOverviewStream();
refresh({ silent: true });
}, 8000);
};
}
async function requestOverviewRefresh() {
try {
await fetch("/api/overview/refresh", {
method: "POST",
credentials: "same-origin",
});
} catch (_) {}
refresh({ silent: true });
}
const WEEKDAYS = ["周日", "周一", "周二", "周三", "周四", "周五", "周六"];
const LUNAR_MONTHS = ["正", "二", "三", "四", "五", "六", "七", "八", "九", "十", "冬", "腊"];
const LUNAR_INFO = [
@@ -1001,12 +1070,19 @@
function boot() {
const btn = document.getElementById("btn-refresh");
if (btn) btn.addEventListener("click", refresh);
if (btn) btn.addEventListener("click", function () {
requestOverviewRefresh();
});
tickBeijingClock();
setInterval(tickBeijingClock, 1000);
// 先进缓存秒开,再挂 SSE;不再前端定时现场拉实例
refresh();
setInterval(refresh, 10000);
connectOverviewStream();
}
global.HubOverview = { boot: boot, refresh: refresh };
global.HubOverview = {
boot: boot,
refresh: refresh,
requestRefresh: requestOverviewRefresh,
};
})(window);
+1 -1
View File
@@ -15,6 +15,6 @@
<div id="overview-grid"></div>
{% endblock %}
{% block scripts %}
<script src="{{ url_for('static', filename='hub.js') }}?v=28"></script>
<script src="{{ url_for('static', filename='hub.js') }}?v=29"></script>
<script>HubOverview.boot();</script>
{% endblock %}