feat: Python 版前置机 Agent 完整实现

- WebSocket 连接 B 端（注册/心跳/指数退避重连） - 命令分发（10 种远程运维命令） - 看门狗守护、日志实时流、在线升级含自动回滚 - 修复 main.py 日志格式 style 与占位符不匹配的问题 - 补充 README、配置示例、依赖清单 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-27 11:58:31 +08:00
parent cc35a41ea4
commit 68751c2354
13 changed files with 1188 additions and 3 deletions
@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+"""日志实时推送管理器。
+
+B 端通过 LOG_TAIL_CTRL 命令开启/停止某设备某日志文件的实时流：
+- 开启后每 2 秒从主程序 /api/admin/log/tail 拉取增量行
+- 新行打包为 LOG_BATCH 上报 B 端（B 端转发给浏览器展示）
+- 每个文件独立游标（已发送行数），支持多文件并行
+
+增量算法：tail(max(offset+FETCH, 400)) 行，取 offset 之后的部分；
+日志滚动（文件变小/重置）时自动重置游标。
+"""
+
+import json
+import logging
+import threading
+import time
+
+log = logging.getLogger(__name__)
+
+# 轮询间隔（秒）
+POLL_SEC = 2
+# 单次拉取上限（行）
+FETCH_MAX = 400
+
+
+class LogStreamManager:
+    """日志实时流：单线程轮询全部活跃文件，增量上报 LOG_BATCH。"""
+
+    def __init__(self, config, main_app_client, client):
+        self.config = config
+        self.main_app_client = main_app_client
+        self.client = client
+
+        self._cursors: dict[str, int] = {}   # 文件名 → 已发送行数
+        self._active: dict[str, bool] = {}   # 文件名 → 是否激活
+        self._lock = threading.Lock()
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+
+    # ================= 控制入口 =================
+
+    def control(self, action: str, file: str) -> dict:
+        """
+        控制命令入口。
+
+        :param action: start / stop
+        :param file:   日志文件名（info.log / error.log）
+        """
+        if action == "start":
+            with self._lock:
+                if self._active.get(file):
+                    return {"ok": True, "msg": f"流已存在：{file}"}
+                self._active[file] = True
+                self._cursors.pop(file, None)
+            self._ensure_thread()
+            log.info("日志流开启：%s", file)
+            return {"ok": True, "msg": f"日志流已开启：{file}"}
+
+        if action == "stop":
+            with self._lock:
+                self._active.pop(file, None)
+                self._cursors.pop(file, None)
+            log.info("日志流停止：%s", file)
+            return {"ok": True, "msg": f"日志流已停止：{file}"}
+
+        return {"ok": False, "msg": f"未知 action：{action}（支持 start/stop）"}
+
+    def stop_all(self):
+        """停止全部流（Agent 停机时调用）。"""
+        self._stop.set()
+        with self._lock:
+            self._active.clear()
+            self._cursors.clear()
+
+    def _ensure_thread(self):
+        """确保轮询线程在运行（首个流开启时启动）。"""
+        if self._thread is not None and self._thread.is_alive():
+            return
+        self._thread = threading.Thread(target=self._poll_loop, name="log-stream", daemon=True)
+        self._thread.start()
+
+    # ================= 轮询 =================
+
+    def _poll_loop(self):
+        """轮询主循环：每 POLL_SEC 秒拉一轮全部活跃文件。"""
+        while not self._stop.wait(POLL_SEC):
+            with self._lock:
+                files = list(self._active.keys())
+            for file in files:
+                if self._stop.is_set():
+                    return
+                try:
+                    self._poll_one(file)
+                except Exception as e:
+                    log.warning("日志流轮询失败（%s）：%s", file, e)
+
+    def _poll_one(self, file: str):
+        """单轮增量拉取：取尾部行，比对游标，新行打包上报。"""
+        with self._lock:
+            if not self._active.get(file):
+                return
+            offset = self._cursors.get(file, 0)
+
+        # 多拉一段，保证能覆盖到 offset 之前的已有内容
+        fetch = max(offset + 100, FETCH_MAX)
+        raw = self.main_app_client.log_tail(file, fetch)
+        log.debug("日志流轮询：file=%s, offset=%s, fetch=%s, rawLen=%s",
+                  file, offset, fetch, len(raw) if raw else -1)
+        if raw is None:
+            return
+        try:
+            node = json.loads(raw)
+        except Exception:
+            return
+        if node.get("code") != 0:
+            return
+        content = node.get("data")
+        if not content:
+            return
+
+        all_lines = content.split("\n")
+
+        # 日志滚动检测：行数比游标还少很多 → 文件已重置
+        if len(all_lines) < offset:
+            with self._lock:
+                self._cursors[file] = 0
+            return
+        if len(all_lines) <= offset:
+            return  # 无新行
+
+        fresh = [line for line in all_lines[offset:] if line and line.strip()]
+        with self._lock:
+            self._cursors[file] = len(all_lines)
+        if not fresh:
+            return
+
+        self.client.send("LOG_BATCH", {
+            "file": file,
+            "lines": fresh,
+            "ts": int(time.time()),
+        })