# -*- 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)