hospital-front-agent/agent/main_app_client.py

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

    # ================= 基础 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)