146 lines
5.8 KiB
Python
146 lines
5.8 KiB
Python
"""平台（auto_control）HTTP 客户端：登录会话 + CSRF + API 封装。
|
||
|
||
会话失效自动重登；CSRF token 每次登录后获取，POST 必带。
|
||
只读接口（GET）与写接口（POST）分离封装，写操作由 MCP 门控层决定是否调用。
|
||
"""
|
||
import httpx
|
||
import logging
|
||
|
||
_log = logging.getLogger("mcp.platform")
|
||
|
||
_CSRF_HEADER = "X-CSRF-Token"
|
||
|
||
|
||
class PlatformError(Exception):
|
||
def __init__(self, code, message):
|
||
super().__init__(message)
|
||
self.code = code
|
||
self.message = message
|
||
|
||
|
||
class PlatformClient:
|
||
def __init__(self, url, user, password, timeout=30.0):
|
||
self.url = url.rstrip("/")
|
||
self.user = user
|
||
self.password = password
|
||
self._client = httpx.Client(base_url=self.url, timeout=timeout,
|
||
follow_redirects=True)
|
||
self._csrf = ""
|
||
self._login()
|
||
|
||
# ---------- 会话 ----------
|
||
def _login(self):
|
||
"""登录平台，拿会话 cookie + CSRF token。"""
|
||
r = self._client.post("/login", data={
|
||
"username": self.user, "password": self.password})
|
||
if r.status_code != 302 and r.status_code != 200:
|
||
raise PlatformError("platform_unavailable",
|
||
f"平台登录失败 HTTP {r.status_code}: {r.text[:120]}")
|
||
csrf = self._client.get("/api/csrf")
|
||
if csrf.status_code == 200:
|
||
self._csrf = (csrf.json() or {}).get("token", "")
|
||
_log.info("平台登录成功: %s", self.user)
|
||
|
||
def _ensure_session(self):
|
||
"""会话过期（401/403/302 到登录页）时重登。"""
|
||
r = self._client.get("/api/status")
|
||
if r.status_code in (401, 403) or r.url.path.endswith("/login"):
|
||
_log.warning("会话失效，重新登录")
|
||
self._login()
|
||
return True
|
||
return False
|
||
|
||
# ---------- 基础请求 ----------
|
||
def _get(self, path, **params):
|
||
self._ensure_session()
|
||
return self._client.get(path, params=params)
|
||
|
||
def _post(self, path, json=None):
|
||
self._ensure_session()
|
||
headers = {_CSRF_HEADER: self._csrf} if self._csrf else {}
|
||
return self._client.post(path, json=json or {}, headers=headers)
|
||
|
||
# ---------- 平台 API 封装（M0） ----------
|
||
def list_devices(self):
|
||
"""设备列表与状态（GET /api/status）。"""
|
||
r = self._get("/api/status")
|
||
if r.status_code != 200:
|
||
raise PlatformError("platform_unavailable",
|
||
f"/api/status HTTP {r.status_code}")
|
||
j = r.json() or {}
|
||
return j.get("devices", [])
|
||
|
||
def screenshot(self, serial):
|
||
"""截图（GET /api/screen/thumb），返回 (jpeg_bytes, screen_state)。"""
|
||
r = self._get("/api/screen/thumb", serial=serial)
|
||
if r.status_code == 503:
|
||
raise PlatformError("device_offline", r.text[:120])
|
||
if r.status_code != 200:
|
||
raise PlatformError("platform_unavailable",
|
||
f"截图 HTTP {r.status_code}")
|
||
return r.content, r.headers.get("X-Screen-State", "unknown")
|
||
|
||
def screen_size(self, serial):
|
||
"""屏幕原生分辨率（GET /api/screen/size），返回 (w, h)。"""
|
||
r = self._get("/api/screen/size", serial=serial)
|
||
if r.status_code == 503:
|
||
raise PlatformError("device_offline", r.text[:120])
|
||
if r.status_code != 200:
|
||
raise PlatformError("platform_unavailable",
|
||
f"分辨率 HTTP {r.status_code}")
|
||
j = r.json() or {}
|
||
if not j.get("ok"):
|
||
raise PlatformError("device_offline", str(j.get("error", "取分辨率失败"))[:120])
|
||
return int(j["width"]), int(j["height"])
|
||
|
||
def tap(self, serial, x, y):
|
||
"""点击（POST /api/screen/tap）。"""
|
||
r = self._post("/api/screen/tap", json={"serial": serial,
|
||
"x": int(x), "y": int(y)})
|
||
return self._check_op(r, "tap")
|
||
|
||
def swipe(self, serial, x1, y1, x2, y2, duration=0.2):
|
||
"""滑动（POST /api/screen/swipe）。"""
|
||
r = self._post("/api/screen/swipe", json={
|
||
"serial": serial, "x1": int(x1), "y1": int(y1),
|
||
"x2": int(x2), "y2": int(y2),
|
||
"duration": float(duration)})
|
||
return self._check_op(r, "swipe")
|
||
|
||
def ui_elements(self, serial):
|
||
"""UI 元素树（GET /api/uiauto/elements，uiautodev 服务）。"""
|
||
r = self._get("/api/uiauto/elements", serial=serial)
|
||
if r.status_code == 503:
|
||
raise PlatformError("device_offline", r.text[:120])
|
||
if r.status_code != 200:
|
||
raise PlatformError("platform_unavailable",
|
||
f"元素树 HTTP {r.status_code}")
|
||
j = r.json() or {}
|
||
if not j.get("ok"):
|
||
raise PlatformError("device_offline", str(j.get("error", "取元素失败"))[:120])
|
||
return j.get("elements", [])
|
||
|
||
def wake(self, serial):
|
||
"""亮屏并解锁（POST /api/device/screen_all mode=on）。"""
|
||
r = self._post("/api/device/screen_all",
|
||
json={"mode": "on", "serials": [serial]})
|
||
return self._check_op(r, "wake")
|
||
|
||
def press_key(self, serial, key):
|
||
"""按键（POST /api/screen/key）。"""
|
||
r = self._post("/api/screen/key",
|
||
json={"serial": serial, "key": key})
|
||
return self._check_op(r, "key")
|
||
|
||
@staticmethod
|
||
def _check_op(r, name):
|
||
if r.status_code == 503:
|
||
raise PlatformError("device_offline", r.text[:120])
|
||
if r.status_code != 200:
|
||
raise PlatformError("platform_unavailable",
|
||
f"{name} HTTP {r.status_code}: {r.text[:120]}")
|
||
j = r.json() or {}
|
||
if not j.get("ok"):
|
||
raise PlatformError("device_offline", str(j.get("error", "操作失败"))[:120])
|
||
return j
|