# -*- 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 log_download(self, file: str) -> str | None: """日志全量下载（文本，自带 X-Admin-Token 鉴权）。""" return self._get_text(self._url(f"/api/admin/log/download?file={quote(file)}")) def restart(self, delay_ms: int) -> tuple[bool, str]: """触发主程序优雅重启。返回 (是否成功, 失败原因)。""" body, reason = self._post_text_with_reason(self._url(f"/api/admin/restart?delayMs={delay_ms}"), "") # 主程序管理接口鉴权失败时返回 code=3001 的 JSON（HTTP 层可能为 200） if body is not None and "3001" in body: return False, "管理接口鉴权失败（main.admin.token 与主程序不一致）" return (body is not None), reason 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 def get_apis(self) -> str | None: """获取主程序 /order 接口清单（原始 JSON 文本）。""" return self._get_text(self._url("/api/admin/apis")) def http_proxy(self, method: str, path: str, query: dict | None = None, body: str | None = None, extra_headers: dict | None = None) -> dict: """ 代理调用前置机任意接口（本机环回，穿透白名单）。 纯透传：鉴权头由调用方（B 端测试页面）通过 headers 传入， Agent 不自动加认证，便于测试不同账号。 :return {"ok":bool, "statusCode":int, "body":str, "headers":dict, "error":str} """ base = self.config.main_health_url.split("/api/")[0] url = base + path # 拼 query if query: qs = "&".join(f"{quote(k)}={quote(str(v))}" for k, v in query.items() if v is not None) if qs: url += ("&" if "?" in url else "?") + qs # 纯透传 headers（鉴权由 B 端填写，不自动加） headers = {} if extra_headers: headers.update(extra_headers) if extra_headers: headers.update(extra_headers) try: method_upper = method.upper() if method_upper == "GET": resp = self._session.get(url, headers=headers, timeout=(5, 60)) elif method_upper == "POST": ct = headers.get("Content-Type", "application/json") headers.setdefault("Content-Type", ct) resp = self._session.post(url, headers=headers, data=body.encode("utf-8") if body else None, timeout=(5, 60)) elif method_upper in ("PUT", "DELETE"): resp = self._session.request(method_upper, url, headers=headers, data=body.encode("utf-8") if body else None, timeout=(5, 60)) else: return {"ok": False, "error": f"不支持的 method：{method}"} return { "ok": True, "statusCode": resp.status_code, "body": resp.text, "headers": dict(resp.headers), } except Exception as e: log.warning("HTTP_PROXY %s %s 失败：%s", method, path, e) return {"ok": False, "error": str(e)} # ================= 基础 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), ) if not resp.ok: log.warning("GET %s 失败：HTTP %s", url, resp.status_code) return None return resp except Exception as e: log.warning("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。""" body, _ = self._post_text_with_reason(url, form_body) return body def _post_text_with_reason(self, url: str, form_body: str) -> tuple[str | None, str]: """POST 表单并返回 (文本, 失败原因)；成功时原因为空串。""" 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), ) if not resp.ok: reason = f"HTTP {resp.status_code}" log.warning("POST %s 失败：%s", url, reason) return None, reason return resp.text, "" except Exception as e: log.warning("POST %s 失败：%s", url, e) return None, str(e)