# -*- coding: utf-8 -*- """Agent 配置加载。 从 conf/agent.properties 读取(key=value),支持环境变量覆盖(AGENT_ 前缀大写)。 配置项: - gateway.url B 端网关基础地址(如 http://10.10.10.228:19901)。 配置后:WebSocket 地址 = 网关地址(scheme转ws) + socket.path + 设备号; 升级下载地址 = 网关地址 + download.prefix + B端下发的文件相对路径。 - socket.path WebSocket 路由相对路径(默认 /medical/agentSocket/,配合 gateway.url 用) - download.prefix 网关文件下载路由前缀(默认 /file/down,如 http://网关:端口/file/down/hf-upgrade/20260902/xxx.jar) - server.url B 端 WebSocket 完整地址(未配置 gateway.url 时的回退方式, 以 /agentSocket/ 结尾,连接时自动拼接设备号) - 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.home 主程序主目录(默认 ".",即与 Agent 同目录;分目录部署时如 ../main) - main.jar.path 主程序 jar 相对 main.home 的路径(默认 bin/hospital-front.jar) - main.pid.file 主程序 PID 文件相对 main.home 的路径(默认 pid/hospital-front.pid) - nssm.path NSSM 可执行文件路径(配置 main.service.name 时用于拉起服务) - main.service.name 主程序注册的 Windows 服务名(配置后看门狗用 nssm restart 拉起,优先于 main.start.cmd) - heartbeat.sec 心跳间隔(默认 15 秒,须小于网关空闲超时) - guard.enable 是否守护主程序(默认 true) - log.dir 日志目录(默认 logs,相对部署根目录) """ import os import re from dataclasses import dataclass @dataclass class AgentConfig: """Agent 全量配置(不可变)。""" gateway_url: str socket_path: str download_prefix: str server_url: str device_id: str device_token: str main_health_url: str main_admin_token: str main_start_cmd: str main_home: str main_jar_path: str main_pid_file: str @property def main_jar(self) -> str: """主程序 jar 完整路径(main.home + main.jar.path)。""" return os.path.join(self.main_home, self.main_jar_path) @property def main_pid(self) -> str: """主程序 PID 文件完整路径(main.home + main.pid.file)。""" return os.path.join(self.main_home, self.main_pid_file) nssm_path: str main_service_name: str heartbeat_sec: int guard_enabled: bool log_dir: str order_basic: str @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()] = _unescape(value.strip()) except Exception: # 读取失败时按空配置处理,交由各配置项默认值兜底 pass return cls( gateway_url=_get(props, "gateway.url", ""), socket_path=_get(props, "socket.path", "/medical/agentSocket/"), download_prefix=_get(props, "download.prefix", "/file/down"), 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"), main_home=_get(props, "main.home", "."), main_jar_path=_get(props, "main.jar.path", "bin/hospital-front.jar"), main_pid_file=_get(props, "main.pid.file", "pid/hospital-front.pid"), nssm_path=_get(props, "nssm.path", ""), main_service_name=_get(props, "main.service.name", ""), heartbeat_sec=int(_get(props, "heartbeat.sec", "15")), guard_enabled=_get(props, "guard.enable", "true").lower() == "true", log_dir=_get(props, "log.dir", "logs"), order_basic=_get(props, "order.basic", ""), ) def _unescape(value: str) -> str: """还原 Java properties 的 \\uXXXX 中文转义(兼容前置的双反斜杠)。 IDEA 等编辑器保存 .properties 时会把非 ASCII 字符(如中文路径)自动转为 \\uXXXX(有时前置双反斜杠);不还原的话路径按字面反斜杠序列处理,找不到目录。 仅替换「反斜杠+u+4位十六进制」序列,Windows 路径分隔符(\\Users 等)不受影响。 """ # 「N 个反斜杠 + u + 4 位十六进制」:解码为字符并保留 N-1 个反斜杠 # (双反斜杠时前一个是 Windows 路径分隔符,不能吃掉) return re.sub( r"(\\+)u([0-9a-fA-F]{4})", lambda m: "\\" * (len(m.group(1)) - 1) + chr(int(m.group(2), 16)), value, ) 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)