From 68751c23545bca04e1a297a2561765fdb3ecfeca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=BF=9E=E9=BE=99=E5=88=9A?= Date: Thu, 27 Aug 2026 11:58:31 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Python=20=E7=89=88=E5=89=8D=E7=BD=AE?= =?UTF-8?q?=E6=9C=BA=20Agent=20=E5=AE=8C=E6=95=B4=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - WebSocket 连接 B 端（注册/心跳/指数退避重连） - 命令分发（10 种远程运维命令） - 看门狗守护、日志实时流、在线升级含自动回滚 - 修复 main.py 日志格式 style 与占位符不匹配的问题 - 补充 README、配置示例、依赖清单 Co-Authored-By: Claude --- .gitignore | 2 +- README.md | 69 +++++++++- agent/__init__.py | 2 + agent/config.py | 71 +++++++++++ agent/dispatcher.py | 141 +++++++++++++++++++++ agent/log_stream.py | 141 +++++++++++++++++++++ agent/main.py | 95 ++++++++++++++ agent/main_app_client.py | 117 +++++++++++++++++ agent/process_guard.py | 110 ++++++++++++++++ agent/upgrade.py | 267 +++++++++++++++++++++++++++++++++++++++ agent/ws_client.py | 161 +++++++++++++++++++++++ conf/agent.properties | 13 ++ requirements.txt | 2 + 13 files changed, 1188 insertions(+), 3 deletions(-) create mode 100644 agent/__init__.py create mode 100644 agent/config.py create mode 100644 agent/dispatcher.py create mode 100644 agent/log_stream.py create mode 100644 agent/main.py create mode 100644 agent/main_app_client.py create mode 100644 agent/process_guard.py create mode 100644 agent/upgrade.py create mode 100644 agent/ws_client.py create mode 100644 conf/agent.properties create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore index 36b13f1..575c1ad 100644 --- a/.gitignore +++ b/.gitignore @@ -166,7 +166,7 @@ cython_debug/ # be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ +.idea/ # Ruff stuff: .ruff_cache/ diff --git a/README.md b/README.md index 1daa3a3..0214a95 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,68 @@ -# hospital-front-agent +# hospital-front-agent（Python 版） -前置机代理 \ No newline at end of file +医院前置机被控端 Agent，从 [hospital-front](../hospital-front)（Kotlin/Java）拆出的独立 Python 实现。 + +## 职责 + +1. **WebSocket 主动连接 B 端管理服务器**（注册 / 心跳 / 接收命令） +2. **守护主程序**（健康探测 + 宕机自动拉起 + 告警） +3. **远程运维命令**：健康查询、版本查询、日志查看（尾部 + 实时流）、配置读写、优雅重启、**在线升级（含自动回滚）** + +与 Java 版的**消息协议完全一致**，B 端无需任何修改。 + +## 目录结构 + +``` +hospital-front-agent/ +├── agent/ +│ ├── main.py # 入口：组装、PID 文件、信号处理 +│ ├── config.py # conf/agent.properties + AGENT_ 环境变量 +│ ├── ws_client.py # WS 客户端：注册/心跳/指数退避重连 +│ ├── dispatcher.py # 命令分发（10 种命令） +│ ├── main_app_client.py # 环回 HTTP 调主程序 /api/admin +│ ├── process_guard.py # 看门狗：3 次失败拉起、告警 +│ ├── log_stream.py # 日志实时流：2s 增量轮询、滚动检测 +│ └── upgrade.py # 升级 SOP：下载→校验→备份→停→替换→拉起→健康检查→回滚 +├── conf/agent.properties # 配置示例 +└── requirements.txt +``` + +## 运行 + +依赖 Python 3.10+。 + +```bash +pip install -r requirements.txt +python -m agent.main +``` + +工作目录须与主程序部署根目录一致（`bin/`、`pid/`、`tmp/`、`backup/` 所在目录）， +配置默认从 `conf/agent.properties` 读取。 + +## 配置 + +见 `conf/agent.properties`。优先级：环境变量（`AGENT_SERVER_URL` 等） > properties > 默认值。 + +## 打包（PyInstaller 单 exe） + +```bash +pip install pyinstaller +pyinstaller --onefile --name hospital-agent --paths . agent/main.py +``` + +产物 `dist/hospital-agent.exe`，拷贝到部署根目录 `bin/` 下即可。 + +## Windows 服务注册（NSSM 示例） + +```bat +nssm install HospitalAgent "C:\hospital\bin\hospital-agent.exe" +nssm set HospitalAgent AppDirectory "C:\hospital" +nssm set HospitalAgent AppStdout "C:\hospital\logs\agent-out.log" +nssm set HospitalAgent AppStderr "C:\hospital\logs\agent-err.log" +``` + +## 与 Java 版的差异 + +- Agent 为独立 Python 进程，**不再占用 `bin/hospital-front.jar`**， + 升级替换 jar 时无需维护 Agent jar 副本（Java 版遗留的 `hospital-agent.jar` 复制逻辑已移除）。 +- 其余命令协议、升级 SOP、看门狗策略、日志流算法均与 Java 版等价。 diff --git a/agent/__init__.py b/agent/__init__.py new file mode 100644 index 0000000..3038200 --- /dev/null +++ b/agent/__init__.py @@ -0,0 +1,2 @@ +# -*- coding: utf-8 -*- +"""hospital-front-agent 包。""" diff --git a/agent/config.py b/agent/config.py new file mode 100644 index 0000000..6d30f55 --- /dev/null +++ b/agent/config.py @@ -0,0 +1,71 @@ +# -*- coding: utf-8 -*- +"""Agent 配置加载。 + +从 conf/agent.properties 读取（key=value），支持环境变量覆盖（AGENT_ 前缀大写）。 + +配置项： +- server.url B 端 WebSocket 地址（经网关转发，须带 /medical 服务前缀） +- device.id 设备编号（医院唯一，如 jinghe-01） +- device.token 设备鉴权 Token +- main.health.url 主程序健康检查地址（http://127.0.0.1:8082/api/admin/health） +- main.admin.token 主程序管理接口 Token（与主程序 admin.token 一致） +- main.start.cmd 主程序启动命令（Windows: bin\\start.bat；Linux: bin/start.sh） +- heartbeat.sec 心跳间隔（默认 15 秒，须小于网关空闲超时） +- guard.enable 是否守护主程序（默认 true） +""" + +import os +from dataclasses import dataclass + + +@dataclass +class AgentConfig: + """Agent 全量配置（不可变）。""" + + server_url: str + device_id: str + device_token: str + main_health_url: str + main_admin_token: str + main_start_cmd: str + heartbeat_sec: int + guard_enabled: bool + + @classmethod + def load(cls, conf_path: str = "conf/agent.properties") -> "AgentConfig": + """从指定路径加载配置，环境变量优先。""" + props = {} + if os.path.exists(conf_path): + try: + with open(conf_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + # 跳过空行与注释 + if not line or line.startswith("#"): + continue + if "=" in line: + key, _, value = line.partition("=") + props[key.strip()] = value.strip() + except Exception: + # 读取失败时按空配置处理，交由各配置项默认值兜底 + pass + + return cls( + server_url=_get(props, "server.url", "ws://127.0.0.1:19901/medical/agentSocket/unknown-device"), + device_id=_get(props, "device.id", "unknown-device"), + device_token=_get(props, "device.token", ""), + main_health_url=_get(props, "main.health.url", "http://127.0.0.1:8082/api/admin/health"), + main_admin_token=_get(props, "main.admin.token", ""), + main_start_cmd=_get(props, "main.start.cmd", "bin/start.bat"), + heartbeat_sec=int(_get(props, "heartbeat.sec", "15")), + guard_enabled=_get(props, "guard.enable", "true").lower() == "true", + ) + + +def _get(props: dict, key: str, default: str) -> str: + """读取单个配置项：环境变量 > properties > 默认值。""" + env_key = "AGENT_" + key.replace(".", "_").upper() + env = os.environ.get(env_key) + if env is not None: + return env + return props.get(key, default) diff --git a/agent/dispatcher.py b/agent/dispatcher.py new file mode 100644 index 0000000..76526c1 --- /dev/null +++ b/agent/dispatcher.py @@ -0,0 +1,141 @@ +# -*- coding: utf-8 -*- +"""下行命令分发器。 + +B 端通过 WebSocket 下发 JSON 命令：{"type":"CMD_TYPE","payload":{...}} + +已支持命令： +- PING 探活（立即回 pong） +- GET_HEALTH 查询主程序健康 +- GET_VERSION 查询主程序版本 +- GET_LOG_FILES 日志文件列表 +- LOG_TAIL 日志尾部（payload: file, lines） +- LOG_TAIL_CTRL 日志实时流开关（payload: action, file） +- RESTART 重启主程序（优雅停机后由 ProcessGuard 拉起） +- GET_CONFIG 读取医院配置文件（payload: hospital） +- SAVE_CONFIG 写回医院配置文件（payload: hospital, content） +- UPGRADE 升级主程序 + +所有命令执行结果通过 AgentClient 上行（CMD_RESULT）。 +""" + +import logging +import threading +import time + +log = logging.getLogger(__name__) + + +class CommandDispatcher: + """命令分发：解析 type → 执行 → 回执 CMD_RESULT（透传 cmdId）。""" + + def __init__(self, config, main_app_client): + self.config = config + self.main_app_client = main_app_client + + self._client = None # 由 main 注入（避免构造环） + self._upgrade_command = None # 升级命令（懒加载） + self._log_stream = None # 日志实时流管理（懒加载） + self._log_stream_lock = threading.Lock() + self._upgrade_lock = threading.Lock() + + def set_client(self, client): + """注入 AgentClient（main 组装阶段调用，避免构造环）。""" + self._client = client + + def _get_upgrade_command(self): + """懒加载升级命令。""" + with self._upgrade_lock: + if self._upgrade_command is None: + from agent.upgrade import UpgradeCommand + self._upgrade_command = UpgradeCommand(self.config, self.main_app_client, self._client) + return self._upgrade_command + + def _get_log_stream(self): + """懒加载日志流管理器。""" + with self._log_stream_lock: + if self._log_stream is None: + from agent.log_stream import LogStreamManager + self._log_stream = LogStreamManager(self.config, self.main_app_client, self._client) + return self._log_stream + + def shutdown_log_stream(self): + """日志实时流（供 main 停机时统一释放）。""" + self._get_log_stream().stop_all() + + # ================= 分发 ================= + + def dispatch(self, msg_type, payload): + """分发并执行命令。payload 为 dict 或 None。""" + log.info("收到命令：%s", msg_type) + payload = payload if isinstance(payload, dict) else None + + if msg_type == "PING": + # 命令协议：B 端下发的 payload 内可带 cmdId，回执时透传以便并发匹配 + result = {"pong": True, "ts": int(time.time())} + + elif msg_type == "GET_HEALTH": + result = self.main_app_client.health() + + elif msg_type == "GET_VERSION": + result = self.main_app_client.version() + + elif msg_type == "GET_LOG_FILES": + result = {"ok": True, "data": self.main_app_client.log_files()} + + elif msg_type == "LOG_TAIL": + file = (payload or {}).get("file") or "info.log" + lines = (payload or {}).get("lines") or 200 + result = {"ok": True, "data": self.main_app_client.log_tail(file, lines)} + + elif msg_type == "LOG_TAIL_CTRL": + action = (payload or {}).get("action") or "start" + file = (payload or {}).get("file") or "info.log" + result = self._get_log_stream().control(action, file) + + elif msg_type == "RESTART": + ok = self.main_app_client.restart(1000) + result = { + "ok": ok, + "msg": "已触发优雅重启，等待 ProcessGuard 拉起" if ok else "重启请求失败", + } + + elif msg_type == "GET_CONFIG": + hospital = (payload or {}).get("hospital") or "" + content = self.main_app_client.config_file(hospital) + result = {"ok": content is not None, "data": content} + + elif msg_type == "SAVE_CONFIG": + hospital = (payload or {}).get("hospital") or "" + content = (payload or {}).get("content") or "" + result = {"ok": self.main_app_client.save_config_file(hospital, content)} + + elif msg_type == "UPGRADE": + # 异步执行（升级耗时长，避免阻塞 WS 消息循环） + version = (payload or {}).get("version") or "unknown" + download_url = (payload or {}).get("downloadUrl") + sha256 = (payload or {}).get("sha256") + if not download_url or not download_url.strip(): + result = {"ok": False, "msg": "缺少 downloadUrl"} + else: + threading.Thread( + target=self._run_upgrade, + args=(version, download_url, sha256), + name="upgrade", + daemon=True, + ).start() + result = {"ok": True, "msg": "升级已开始（异步执行，结果将通过 UPGRADE_RESULT 上报）"} + + else: + result = {"ok": False, "msg": f"未知命令：{msg_type}"} + + envelope = {"cmd": msg_type} + # 透传 B 端下发的 cmdId（并发命令匹配用） + if payload is not None and payload.get("cmdId") is not None: + envelope["cmdId"] = payload.get("cmdId") + envelope["result"] = result + self._client.send("CMD_RESULT", envelope) + + def _run_upgrade(self, version, download_url, sha256): + """升级线程体：执行并上报 UPGRADE_RESULT。""" + r = self._get_upgrade_command().upgrade(version, download_url, sha256) + self._client.send("UPGRADE_RESULT", r) diff --git a/agent/log_stream.py b/agent/log_stream.py new file mode 100644 index 0000000..323e8e0 --- /dev/null +++ b/agent/log_stream.py @@ -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()), + }) diff --git a/agent/main.py b/agent/main.py new file mode 100644 index 0000000..4639a39 --- /dev/null +++ b/agent/main.py @@ -0,0 +1,95 @@ +# -*- coding: utf-8 -*- +"""Agent 启动入口。 + +启动方式： +python -m agent.main （源码运行） +hospital-agent.exe （PyInstaller 打包后） + +职责： +1. WebSocket 主动连接 B 端管理服务器（注册/心跳/命令） +2. 守护主程序（健康探测 + 宕机拉起） +3. 写 PID 文件（pid/hospital-agent.pid） + +由 WinSW/NSSM Windows 服务守护本进程。 +""" + +import logging +import os +import signal +import sys +import threading + +from agent.config import AgentConfig +from agent.main_app_client import MainAppClient + +log = logging.getLogger(__name__) + +# 组装阶段先声明，供信号处理器引用 +_dispatcher = None +_guard = None +_client = None + + +def run(): + """主流程：加载配置 → 写 PID → 组装组件 → 启动 → 主线程保活。""" + global _dispatcher, _guard, _client + + logging.basicConfig( + level=logging.INFO, + format="{asctime} {levelname} [{name}] {message}", + style="{", + ) + # 源码运行时根目录即工作目录，PyInstaller 打包后同理（服务管理器负责设置工作目录） + config = AgentConfig.load() + log.info("Agent 启动：deviceId=%s, server=%s", config.device_id, config.server_url) + + # 写 PID 文件 + _write_pid_file() + + # 组装组件（延迟导入避免环） + from agent.dispatcher import CommandDispatcher + from agent.process_guard import ProcessGuard + from agent.ws_client import AgentClient + + main_app_client = MainAppClient(config) + dispatcher = CommandDispatcher(config, main_app_client) + client = AgentClient(config, dispatcher) + dispatcher.set_client(client) + guard = ProcessGuard(config, main_app_client, client) + _dispatcher, _guard, _client = dispatcher, guard, client + + # 优雅停机信号处理 + signal.signal(signal.SIGINT, _shutdown) + signal.signal(signal.SIGTERM, _shutdown) + + client.start() + guard.start() + + # 主线程保活 + threading.Event().wait() + + +def _shutdown(signum, frame): + """优雅停机：释放日志流 → 停守护 → 断连接 → 退出。""" + log.info("Agent 停机中...（signal=%s）", signum) + if _dispatcher is not None: + _dispatcher.shutdown_log_stream() + if _guard is not None: + _guard.stop() + if _client is not None: + _client.stop() + sys.exit(0) + + +def _write_pid_file(): + """写 PID 文件（pid/hospital-agent.pid）。""" + try: + os.makedirs("pid", exist_ok=True) + with open(os.path.join("pid", "hospital-agent.pid"), "w", encoding="utf-8") as f: + f.write(str(os.getpid())) + except Exception as e: + log.warning("PID 文件写入失败：%s", e) + + +if __name__ == "__main__": + run() diff --git a/agent/main_app_client.py b/agent/main_app_client.py new file mode 100644 index 0000000..21f2f44 --- /dev/null +++ b/agent/main_app_client.py @@ -0,0 +1,117 @@ +# -*- coding: utf-8 -*- +"""主程序 IPC 客户端（环回 HTTP）。 + +Agent 通过本机 HTTP 调用主程序被控端 API（/api/admin）。 +""" + +import logging +from urllib.parse import quote + +import requests + +log = logging.getLogger(__name__) + + +class MainAppClient: + """主程序管理接口客户端。""" + + def __init__(self, config): + self.config = config + self._session = requests.Session() + # 连接失败快速返回（探测场景多，不能长时间阻塞） + self._session.mount("http://", requests.adapters.HTTPAdapter()) + + # ================= 业务接口 ================= + + def health(self) -> dict: + """主程序健康检查：返回解析后的 dict；不可达返回 {"ok":False,"error":"..."}""" + body = self._get_json(self.config.main_health_url) + if body is None: + return {"ok": False, "error": "主程序不可达"} + try: + node = body.json() + return { + "ok": True, + "code": node.get("code"), + "data": node.get("data"), + } + except Exception as e: + return {"ok": False, "error": f"健康响应解析失败：{e}"} + + def version(self) -> dict: + """主程序版本。""" + body = self._get_json(self._url("/api/admin/version")) + if body is None: + return {"ok": False} + try: + node = body.json() + return {"ok": True, "data": node.get("data")} + except Exception as e: + return {"ok": False, "error": str(e)} + + def log_tail(self, file: str, lines: int) -> str | None: + """日志尾部（原始 JSON 文本）。""" + return self._get_text(self._url(f"/api/admin/log/tail?file={quote(file)}&lines={lines}")) + + def log_files(self) -> str | None: + """日志文件列表（原始 JSON 文本）。""" + return self._get_text(self._url("/api/admin/log/files")) + + def restart(self, delay_ms: int) -> bool: + """触发主程序优雅重启。""" + return self._post_text(self._url(f"/api/admin/restart?delayMs={delay_ms}"), "") is not None + + def config_file(self, hospital: str) -> str | None: + """读取医院配置文件（原始 JSON 文本）。""" + return self._get_text(self._url(f"/api/admin/config/file?hospital={quote(hospital)}")) + + def save_config_file(self, hospital: str, content: str) -> bool: + """写回医院配置文件。""" + body = self._post_text( + self._url("/api/admin/config/file"), + f"hospital={quote(hospital)}&content={quote(content)}", + ) + return body is not None + + # ================= 基础 HTTP ================= + + def _url(self, path: str) -> str: + """基于健康地址推导 /api/admin 基地址。""" + idx = self.config.main_health_url.find("/api/") + base = self.config.main_health_url[:idx] if idx >= 0 else self.config.main_health_url + return base + path + + def _get_json(self, url: str): + """GET 并返回 Response 对象；失败返回 None。""" + try: + resp = self._session.get( + url, + headers={"X-Admin-Token": self.config.main_admin_token}, + timeout=(5, 30), + ) + return resp if resp.ok else None + except Exception as e: + log.debug("GET %s 失败：%s", url, e) + return None + + def _get_text(self, url: str) -> str | None: + """GET 并返回文本；失败返回 None。""" + resp = self._get_json(url) + return resp.text if resp is not None else None + + def _post_text(self, url: str, form_body: str) -> str | None: + """POST 表单并返回文本；失败返回 None。""" + try: + resp = self._session.post( + url, + headers={ + "X-Admin-Token": self.config.main_admin_token, + "Content-Type": "application/x-www-form-urlencoded", + }, + data=form_body.encode("utf-8"), + timeout=(5, 30), + ) + return resp.text if resp.ok else None + except Exception as e: + log.debug("POST %s 失败：%s", url, e) + return None diff --git a/agent/process_guard.py b/agent/process_guard.py new file mode 100644 index 0000000..302904d --- /dev/null +++ b/agent/process_guard.py @@ -0,0 +1,110 @@ +# -*- coding: utf-8 -*- +"""主程序进程守护（看门狗）。 + +周期探测主程序健康： +- 连续 3 次失败 → 认为主程序宕机 → 强杀残留 → 调启动命令拉起 +- 拉起后等待健康恢复；连续 3 次拉起失败 → 上报告警（ALERT） +""" + +import logging +import subprocess +import threading + +log = logging.getLogger(__name__) + + +class ProcessGuard: + """看门狗：周期健康探测 + 宕机拉起 + 告警上报。""" + + def __init__(self, config, main_app_client, client): + self.config = config + self.main_app_client = main_app_client + self.client = client + + self._fail_count = 0 # 连续健康探测失败次数 + self._restart_count = 0 # 连续拉起次数 + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + def start(self): + """启动守护（首检延迟 30 秒，之后每 15 秒探测一次）。""" + if not self.config.guard_enabled: + log.info("进程守护未启用（guard.enable=false）") + return + self._thread = threading.Thread(target=self._loop, name="process-guard", daemon=True) + self._thread.start() + log.info("进程守护已启动：每 15 秒探测 %s", self.config.main_health_url) + + def stop(self): + """停止守护。""" + self._stop.set() + + # ================= 探测循环 ================= + + def _loop(self): + """探测主循环：先等 30 秒（给主程序启动时间），再周期探测。""" + if self._stop.wait(30): + return + while not self._stop.wait(15): + try: + self._check() + except Exception as e: + log.warning("守护探测异常：%s", e) + + def _check(self): + """单次探测：健康则清零计数；连续 3 次失败则拉起。""" + health = self.main_app_client.health() + if health.get("ok") is True: + self._fail_count = 0 + return + + self._fail_count += 1 + log.warning("主程序健康探测失败（%d/3）：%s", self._fail_count, health.get("error")) + if self._fail_count < 3: + return + + # 连续 3 次失败 → 拉起 + self._fail_count = 0 + self._restart_count += 1 + attempts = self._restart_count + if attempts > 3: + self.client.send("ALERT", { + "level": "CRITICAL", + "msg": f"主程序连续 {attempts} 次拉起失败，请人工介入", + }) + # 重置计数，之后继续尝试（降频可后续优化） + self._restart_count = 3 + return + + log.error("主程序宕机，执行第 %d 次拉起：%s", attempts, self.config.main_start_cmd) + self.client.send("ALERT", { + "level": "WARN", + "msg": f"主程序宕机，Agent 正在执行第 {attempts} 次拉起", + }) + try: + proc = start_main_app(self.config.main_start_cmd) + log.info("启动命令已执行，pid=%s", proc.pid if proc else None) + except Exception as e: + log.error("启动命令执行失败：%s", e) + + +# ================= 公共工具（upgrade.py 复用） ================= + +def start_main_app(start_cmd: str) -> subprocess.Popen | None: + """ + 拉起主程序。 + + Windows 下 .bat 走 cmd /c（路径转反斜杠）；其余走 sh。 + 返回 Popen 对象（不等待进程结束）。 + """ + cmd = start_cmd.replace("/", "\\") # cmd/c 要求反斜杠路径 + if cmd.lower().endswith(".bat"): + args = ["cmd", "/c", cmd, "start"] + else: + args = ["sh", start_cmd, "start"] + return subprocess.Popen( + args, + stdout=subprocess.DEVNULL, + stderr=subprocess.STDOUT, + # 不设 cwd，继承 Agent 工作目录（与主程序同根目录部署） + ) diff --git a/agent/upgrade.py b/agent/upgrade.py new file mode 100644 index 0000000..1402683 --- /dev/null +++ b/agent/upgrade.py @@ -0,0 +1,267 @@ +# -*- coding: utf-8 -*- +"""主程序升级命令。 + +流程（见 05-部署运维.md SOP）： +1. 从 B 端下载新 jar 到 tmp（SHA-256 校验，防传输损坏） +2. 备份当前 bin/hospital-front.jar → backup/hospital-front-{时间戳}.jar（保留最近 5 个） +3. 优雅停止主程序（调 /api/admin/restart；超时强杀） +4. 替换 jar +5. 调启动命令拉起主程序 +6. 轮询健康检查（最长 90 秒） +7. 成功 → 上报；失败 → 用备份 jar 自动回滚并重启 → 上报 + +期间整个流程串行（upgrade_lock 防并发升级）。 +""" + +import glob +import hashlib +import logging +import os +import shutil +import subprocess +import threading +import time + +import requests + +from agent.process_guard import start_main_app + +log = logging.getLogger(__name__) + +# 备份保留数量 +KEEP_BACKUPS = 5 + + +class UpgradeCommand: + """升级 SOP 执行器。""" + + def __init__(self, config, main_app_client, client): + self.config = config + self.main_app_client = main_app_client + self.client = client + + # 升级互斥锁（同一时刻只允许一个升级在跑） + self._upgrade_lock = threading.Lock() + + def upgrade(self, version: str, download_url: str, sha256: str | None) -> dict: + """ + 执行升级。 + + :param version: 新版本号（仅用于日志与上报） + :param download_url: B 端提供的 jar 下载地址（A 主动出站拉取） + :param sha256: 可选校验值 + """ + with self._upgrade_lock: + log.info("开始升级到版本 %s：%s", version, download_url) + self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "download"}) + + # ===== 1. 下载 ===== + tmp_jar = os.path.join("tmp", f"upgrade-{version}.jar") + os.makedirs(os.path.dirname(tmp_jar), exist_ok=True) + try: + with requests.get(download_url, stream=True, timeout=(10, 120)) as resp: + if resp.status_code != 200: + return self._fail(version, f"下载失败：HTTP {resp.status_code}") + with open(tmp_jar, "wb") as f: + for chunk in resp.iter_content(8192): + f.write(chunk) + except Exception as e: + return self._fail(version, f"下载异常：{e}") + log.info("下载完成：%s（%d 字节）", os.path.basename(tmp_jar), os.path.getsize(tmp_jar)) + + # ===== 2. 校验 ===== + if sha256: + try: + actual = sha256_file(tmp_jar) + except Exception as e: + return self._fail(version, f"SHA-256 计算异常：{e}") + if actual.lower() != sha256.lower(): + os.remove(tmp_jar) + return self._fail(version, f"SHA-256 校验失败：expected={sha256}, actual={actual}") + log.info("SHA-256 校验通过") + + # ===== 3. 备份 ===== + jar_file = os.path.join("bin", "hospital-front.jar") + backup_file = os.path.join("backup", f"hospital-front-{int(time.time() * 1000)}.jar") + os.makedirs(os.path.dirname(backup_file), exist_ok=True) + if os.path.exists(jar_file): + try: + shutil.copyfile(jar_file, backup_file) + except Exception as e: + return self._fail(version, f"备份失败：{e}") + log.info("已备份当前版本 → %s", os.path.basename(backup_file)) + self._clean_old_backups() + + # ===== 4. 停止主程序 ===== + self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "stop"}) + self._stop_main_app() + log.info("主程序已停止") + + # ===== 5. 替换 ===== + # 注意：Agent 为独立 Python 进程，不占用 bin/hospital-front.jar， + # 可安全替换；无需再维护 Agent jar 副本（Java 版遗留逻辑）。 + try: + copy_with_retry(tmp_jar, jar_file, 10) + os.remove(tmp_jar) + except Exception as e: + self._rollback(backup_file) + return self._fail(version, f"替换 jar 失败，已回滚：{e}") + log.info("jar 已替换") + + # ===== 6. 拉起 + 健康检查 ===== + self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "restart"}) + try: + start_main_app(self.config.main_start_cmd) + except Exception as e: + log.error("主程序拉起失败：%s", e) + + self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "healthcheck"}) + if self._wait_healthy(90): + log.info("升级成功：%s", version) + return {"ok": True, "version": version, "msg": "升级成功"} + + # ===== 7. 健康检查失败 → 回滚 ===== + log.error("新版本健康检查失败，自动回滚到 %s", os.path.basename(backup_file)) + self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "rollback"}) + self._stop_main_app() + self._rollback(backup_file) + try: + start_main_app(self.config.main_start_cmd) + except Exception as e: + log.error("回滚后拉起失败：%s", e) + if self._wait_healthy(90): + return self._fail(version, f"新版本启动失败，已自动回滚到 {os.path.basename(backup_file)} 并恢复服务") + + self.client.send("ALERT", { + "level": "CRITICAL", + "msg": f"升级失败且回滚后仍不健康，请人工介入！版本={version}", + }) + return self._fail(version, "升级失败且回滚后仍不健康，请人工介入") + + # ================= 内部步骤 ================= + + def _stop_main_app(self): + """停止主程序：优先优雅（admin API），超时后按 PID 强杀；确保进程完全退出（jar 锁释放）。""" + pid = read_pid_file(os.path.join("pid", "hospital-front.pid")) + if self.main_app_client.restart(100): + # 等待健康探测失败（最长 30 秒） + deadline = time.time() + 30 + while time.time() < deadline: + if self.main_app_client.health().get("ok") is not True: + break + time.sleep(1) + # 兜底：按 PID 文件强杀 + kill_by_pid_file(os.path.join("pid", "hospital-front.pid")) + # 等待进程真正退出（Windows 下进程退出后才释放 jar 文件锁） + if pid is not None: + wait_process_exit(pid, 15) + time.sleep(1) + + def _wait_healthy(self, timeout_sec: int) -> bool: + """轮询等待主程序健康。""" + deadline = time.time() + timeout_sec + while time.time() < deadline: + if self.main_app_client.health().get("ok") is True: + return True + time.sleep(3) + return False + + def _rollback(self, backup_file: str): + """回滚：备份 jar 覆盖回 bin。""" + try: + if os.path.exists(backup_file): + jar_file = os.path.join("bin", "hospital-front.jar") + copy_with_retry(backup_file, jar_file, 10) + log.info("已回滚到 %s", os.path.basename(backup_file)) + except Exception as e: + log.error("回滚失败：%s", e) + + def _clean_old_backups(self): + """清理旧备份（按文件名倒序保留最近 N 个）。""" + backups = sorted(glob.glob(os.path.join("backup", "hospital-front-*")), reverse=True) + for old in backups[KEEP_BACKUPS:]: + try: + os.remove(old) + except Exception: + pass + + def _fail(self, version: str, msg: str) -> dict: + """统一失败上报。""" + log.error("升级失败：%s", msg) + return {"ok": False, "version": version, "msg": msg} + + +# ================= 模块级工具函数 ================= + +def read_pid_file(pid_file: str) -> str | None: + """读 PID 文件，无或为空返回 None。""" + try: + if not os.path.exists(pid_file): + return None + with open(pid_file, "r", encoding="utf-8") as f: + s = f.read().strip() + return s or None + except Exception: + return None + + +def kill_by_pid_file(pid_file: str): + """按 PID 文件强杀进程（Windows taskkill / Linux kill -9）。""" + pid = read_pid_file(pid_file) + if not pid: + return + try: + if os.name == "nt": + subprocess.run(["taskkill", "/PID", pid, "/T", "/F"], + capture_output=True, timeout=15) + else: + subprocess.run(["kill", "-9", pid], + capture_output=True, timeout=15) + except Exception as e: + log.warning("PID 强杀失败：%s", e) + + +def wait_process_exit(pid: str, timeout_sec: int): + """轮询等待进程退出（tasklist 查不到即退出）。""" + deadline = time.time() + timeout_sec + while time.time() < deadline: + alive = False + try: + if os.name == "nt": + out = subprocess.run( + ["tasklist", "/FI", f"PID eq {pid}"], + capture_output=True, timeout=10, + ).stdout.decode("utf-8", errors="ignore") + alive = pid in out + else: + alive = subprocess.run(["kill", "-0", pid], capture_output=True).returncode == 0 + except Exception: + alive = False + if not alive: + return + time.sleep(0.5) + + +def copy_with_retry(src: str, dst: str, retries: int): + """带重试的文件覆盖（Windows 下进程退出有延迟，锁释放需等待）。""" + last_error = None + for _ in range(retries): + try: + if os.path.exists(dst): + os.remove(dst) + if not os.path.exists(dst): + shutil.copyfile(src, dst) + return + except Exception as e: + last_error = e + time.sleep(1) + raise RuntimeError(f"文件替换失败（重试 {retries} 次）：{os.path.basename(dst)}") from last_error + + +def sha256_file(path: str) -> str: + """文件 SHA-256（分块读取，支持大文件）。""" + digest = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(8192), b""): + digest.update(chunk) + return digest.hexdigest() diff --git a/agent/ws_client.py b/agent/ws_client.py new file mode 100644 index 0000000..4ab93ee --- /dev/null +++ b/agent/ws_client.py @@ -0,0 +1,161 @@ +# -*- coding: utf-8 -*- +"""Agent WebSocket 客户端。 + +- 主动出站连接 B 端（穿透白名单，无需 A 端开放入站端口） +- 断线指数退避重连（1s → 2s → 4s → ... 最大 60s） +- 周期心跳：拉取主程序 /api/admin/health 并转发给 B 端 +- 收到下行命令交给 CommandDispatcher 处理 + +上行消息格式：{"type":"HEARTBEAT","deviceId":"...","ts":...,"payload":{...}} +""" + +import json +import logging +import threading +import time + +import websocket # websocket-client + +log = logging.getLogger(__name__) + + +class AgentClient: + """WebSocket 长连接客户端（连接管理 + 心跳 + 重连）。""" + + def __init__(self, config, dispatcher): + self.config = config + self.dispatcher = dispatcher + + self._ws = None # 当前 WebSocket 连接 + self._connected = False # 是否已连接 + self._reconnect_attempts = 0 # 连续重连次数（成功后清零） + self._lock = threading.Lock() # 保护 _ws/_connected 的并发访问 + self._heartbeat_stop = threading.Event() + self._threads: list[threading.Thread] = [] + + # ================= 生命周期 ================= + + def start(self): + """启动：连接 + 心跳线程（首次心跳立即发送，避免网关空闲超时在首个周期内掐断连接）。""" + self._connect() + t = threading.Thread(target=self._heartbeat_loop, name="heartbeat", daemon=True) + t.start() + self._threads.append(t) + + def stop(self): + """停止并释放资源。""" + self._heartbeat_stop.set() + with self._lock: + if self._ws is not None: + try: + self._ws.close() + except Exception: + pass + self._ws = None + self._connected = False + + # ================= 连接管理 ================= + + def _connect(self): + """建立 WebSocket 连接（阻塞直到连接断开，内部自行调度重连）。""" + url = self.config.server_url + log.info("连接管理服务器：%s", url) + + headers = [ + f"X-Device-Id: {self.config.device_id}", + f"X-Device-Token: {self.config.device_token}", + ] + + def on_open(ws): + self._connected = True + self._reconnect_attempts = 0 + log.info("已连接管理服务器") + self.send("REGISTER", { + "deviceId": self.config.device_id, + "version": "1.0.0", + }) + + def on_message(ws, text): + try: + node = json.loads(text) + msg_type = node.get("type") + payload = node.get("payload") + self.dispatcher.dispatch(msg_type, payload) + except Exception as e: + log.warning("命令处理异常：%s", e) + + def on_close(ws, code, reason): + self._connected = False + log.info("连接关闭：code=%s, reason=%s", code, reason) + self._schedule_reconnect() + + def on_error(ws, error): + self._connected = False + log.warning("连接失败：%s，将重连", error) + # websocket-client 出错后必定回调 on_close，重连由 on_close 统一调度 + # 此处不重复调度，避免双重重连 + + ws = websocket.WebSocketApp( + url, + header=headers, + on_open=on_open, + on_message=on_message, + on_close=on_close, + on_error=on_error, + ) + with self._lock: + self._ws = ws + # 阻塞运行（跳过协议层 ping，应用层心跳已足够；关掉默认 ping 避免双重心跳） + ws.run_forever(ping_interval=25, ping_timeout=10) + + def _schedule_reconnect(self): + """指数退避重连：1s 起，翻倍至 60s 封顶。""" + self._reconnect_attempts += 1 + attempt = self._reconnect_attempts + delay = min(1000 * (2 ** min(attempt - 1, 6)), 60_000) / 1000.0 + log.info("%.0fs 后进行第 %d 次重连", delay, attempt) + time.sleep(delay) + if not self._heartbeat_stop.is_set(): + self._connect() + + # ================= 心跳 ================= + + def _heartbeat_loop(self): + """周期心跳：拉主程序健康数据转发。""" + while not self._heartbeat_stop.wait(self.config.heartbeat_sec): + try: + self._heartbeat_once() + except Exception as e: + log.warning("心跳发送异常：%s", e) + + def _heartbeat_once(self): + main_health = self.dispatcher.main_app_client.health() + self.send("HEARTBEAT", { + "deviceId": self.config.device_id, + "mainApp": main_health, + }) + + # ================= 上行消息 ================= + + def send(self, msg_type: str, payload): + """发送上行消息（未连接时丢弃并记日志）。""" + with self._lock: + ws = self._ws + connected = self._connected + if ws is None or not connected: + log.debug("未连接，丢弃上行消息：%s", msg_type) + return + envelope = { + "type": msg_type, + "deviceId": self.config.device_id, + "ts": int(time.time()), + "payload": payload, + } + try: + ws.send(json.dumps(envelope, ensure_ascii=False)) + except Exception as e: + log.warning("上行消息发送失败（%s）：%s", msg_type, e) + + def is_connected(self) -> bool: + """当前是否已连接（供健康状态展示）。""" + return self._connected diff --git a/conf/agent.properties b/conf/agent.properties new file mode 100644 index 0000000..c053027 --- /dev/null +++ b/conf/agent.properties @@ -0,0 +1,13 @@ +# Agent 配置 +# 值优先从环境变量读取（AGENT_ 前缀 + 大写 + 下划线，如 AGENT_SERVER_URL） +# 服务脚本自动加载 conf/.env 设置环境变量；NSSM 用 AppEnvironmentExtra 设置 +# +# 以下是 fallback 默认值（环境变量未设置时使用）： +server.url=ws://10.10.10.228:19901/medical/agentSocket/unknown-device +device.id=unknown-device +device.token= +main.health.url=http://127.0.0.1:8082/api/admin/health +main.admin.token= +main.start.cmd=bin/service.bat main start +heartbeat.sec=15 +guard.enable=true diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..82a1d47 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +websocket-client>=1.6.0 +requests>=2.31.0