From 0e3fffef441a6cd5472472dc85a37eff0fa51bd6 Mon Sep 17 00:00:00 2001 From: dekun Date: Wed, 7 Oct 2026 14:24:26 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=AD=E6=8E=A7=E6=80=BB=E8=A7=88=E6=94=B9?= =?UTF-8?q?=E4=B8=BA=E5=90=8E=E5=8F=B0=E5=B8=B8=E9=A9=BB=E5=B9=B6=E8=A1=8C?= =?UTF-8?q?=E6=8B=89=E5=8F=96=E7=83=AD=E7=BC=93=E5=AD=98=EF=BC=8C=E6=B5=8F?= =?UTF-8?q?=E8=A7=88=E5=99=A8=E5=8F=AA=E8=AF=BB=E7=BC=93=E5=AD=98=E5=B9=B6?= =?UTF-8?q?=E7=BB=8F=20SSE=20=E6=9B=B4=E6=96=B0=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- .env.example | 2 + app.py | 62 +++++++----- lib/overview_cache_lib.py | 197 ++++++++++++++++++++++++++++++++++++++ static/hub.js | 100 ++++++++++++++++--- templates/dashboard.html | 2 +- 5 files changed, 327 insertions(+), 36 deletions(-) create mode 100644 lib/overview_cache_lib.py diff --git a/.env.example b/.env.example index 040c886..788712a 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/app.py b/app.py index 444ebd7..72bd326 100644 --- a/app.py +++ b/app.py @@ -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( diff --git a/lib/overview_cache_lib.py b/lib/overview_cache_lib.py new file mode 100644 index 0000000..39bedc3 --- /dev/null +++ b/lib/overview_cache_lib.py @@ -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() diff --git a/static/hub.js b/static/hub.js index d37c1df..5c20fe3 100644 --- a/static/hub.js +++ b/static/hub.js @@ -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 = - '

暂无启用实例,请先到「实例」添加.

'; + d.pulling || !d.updated_at + ? '

后台首次拉取中…

' + : '

暂无启用实例,请先到「实例」添加.

'; } else if (isHubPhone()) { grid.innerHTML = '
' + list.map(renderPhoneCard).join("") + "
"; @@ -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); diff --git a/templates/dashboard.html b/templates/dashboard.html index 7bf4dbc..c231020 100644 --- a/templates/dashboard.html +++ b/templates/dashboard.html @@ -15,6 +15,6 @@
{% endblock %} {% block scripts %} - + {% endblock %}