"""OpenSTF REST API 客户端封装。 负责：设备列表查询、占用、远程 ADB 隧道建立、释放。 不负责实际 UI 操作（那是 DeviceWorker 的事）。 健壮性设计： - 所有请求带 timeout，避免 STF 网关 504 时长时间卡死 - remoteConnect 带重试，STF provider 偶尔慢启动 - 友好错误分类：设备离线 / 网络不通 / STF 内部错误，便于上层决策 - 占用冲突自动重试（STF 偶发 400 "device already in use"） """ import time import requests from config import STF_URL, STF_TOKEN from core.logger import get_logger _log = get_logger("core.stf") # 请求超时（秒）。connect 超时 + read 超时 _TIMEOUT = (5, 15) # remoteConnect 重试次数（STF provider 慢启动时需要重试） _RC_RETRIES = 3 # remoteConnect 重试间隔 _RC_DELAY = 2 class STFError(Exception): """STF 相关错误基类。""" def __init__(self, message, code=""): super().__init__(message) self.code = code # 错误码：offline / network / conflict / server / unknown class DeviceOfflineError(STFError): """设备离线/不可用。""" class STFNetworkError(STFError): """STF 服务不可达。""" class DeviceConflictError(STFError): """设备已被占用。""" def _headers(): return {"Authorization": f"Bearer {STF_TOKEN}"} class STFClient: """OpenSTF REST API 封装。""" def __init__(self, base_url=STF_URL, token=STF_TOKEN): self.base_url = base_url self.headers = _headers() # ================== 设备查询 ================== def list_all_devices(self): """返回 STF 上所有设备（含状态）。""" try: resp = requests.get(f"{self.base_url}/api/v1/devices", headers=self.headers, timeout=_TIMEOUT) resp.raise_for_status() return resp.json().get("devices", []) except requests.exceptions.ConnectionError as e: raise STFNetworkError(f"STF 服务不可达: {e}", "network") from e except requests.exceptions.Timeout as e: raise STFNetworkError(f"STF 请求超时: {e}", "network") from e except requests.exceptions.RequestException as e: _log.error(f"获取设备列表失败: {e}") raise STFError(f"获取设备列表失败: {e}", "server") from e def list_free_devices(self): """返回可占用的空闲设备。""" return [d for d in self.list_all_devices() if d.get("present") and d.get("ready") and not d.get("using") and d.get("owner") is None] def list_my_devices(self): """返回当前账户已占用的设备。""" try: resp = requests.get(f"{self.base_url}/api/v1/user/devices", headers=self.headers, timeout=_TIMEOUT) resp.raise_for_status() return resp.json().get("devices", []) except requests.exceptions.RequestException as e: _log.error(f"获取已占用设备失败: {e}") return [] # ================== 占用/释放 ================== def occupy(self, serial, retries=2): """占用设备。返回 (ok, msg)。 冲突（已被占用）时自动重试，偶发 400 "already in use" 可能是脏状态。 """ for attempt in range(1, retries + 1): try: r = requests.post(f"{self.base_url}/api/v1/user/devices/{serial}", headers=self.headers, timeout=_TIMEOUT) if r.status_code in (200, 201): return True, r.text[:200] if r.status_code == 400 and "already" in r.text.lower(): # 已被占用（可能是自己之前占用没释放干净），尝试先释放再占 _log.warning(f"{serial} 占用冲突，尝试清理后重试 ({attempt}/{retries})") try: requests.delete(f"{self.base_url}/api/v1/user/devices/{serial}", headers=self.headers, timeout=_TIMEOUT) except Exception: pass time.sleep(1) continue return False, f"HTTP {r.status_code}: {r.text[:200]}" except requests.exceptions.RequestException as e: _log.warning(f"{serial} 占用请求异常 ({attempt}/{retries}): {e}") if attempt < retries: time.sleep(1) return False, str(e) return False, "占用冲突重试耗尽" def release(self, serial, retries=3): """释放设备占用。返回 (ok, msg)。 STF 释放可能因设备响应超时返回 504（如设备过载/卡死），此时设备实际未释放。 必须检查响应状态并重试，不能静默吞掉——否则占用会一直悬着。 """ self.remote_disconnect(serial) last_status = None last_body = "" for attempt in range(1, retries + 1): try: r = requests.delete(f"{self.base_url}/api/v1/user/devices/{serial}", headers=self.headers, timeout=_TIMEOUT) if r.status_code in (200, 201, 202, 204): return True, "released" last_status = r.status_code last_body = r.text[:150] _log.warning(f"释放 {serial} 失败 HTTP {r.status_code}: {last_body} ({attempt}/{retries})") except requests.exceptions.RequestException as e: _log.warning(f"释放 {serial} 请求异常 ({attempt}/{retries}): {e}") last_status = -1 last_body = str(e) if attempt < retries: time.sleep(2) return False, f"STF 释放失败(HTTP {last_status}): {last_body}" def release_all_mine(self): """释放当前账户占用的所有设备（清理用）。返回 (released, failed)。""" released, failed = [], [] for d in self.list_my_devices(): serial = d["serial"] ok, msg = self.release(serial) if ok: released.append(serial) _log.info(f"已释放 {serial}") else: failed.append(serial) _log.error(f"释放 {serial} 失败: {msg}") return released, failed # ================== 远程 ADB 隧道 ================== def remote_connect(self, serial): """建立远程 ADB 隧道，返回 remoteConnectUrl。 STF provider 慢启动或设备掉线时会 504，这里带重试。 设备真离线时快速失败，不长时间卡住 worker。 """ last_err = None for attempt in range(1, _RC_RETRIES + 1): try: conn = requests.post( f"{self.base_url}/api/v1/user/devices/{serial}/remoteConnect", headers=self.headers, timeout=_TIMEOUT, ) # 504 = STF 网关等 provider 响应超时，通常是设备掉线或 provider 卡死 if conn.status_code == 504: last_err = f"STF 网关超时(504)，设备可能掉线" _log.warning(f"{serial} remoteConnect 504 ({attempt}/{_RC_RETRIES})") if attempt < _RC_RETRIES: time.sleep(_RC_DELAY) continue raise DeviceOfflineError( f"{serial} remoteConnect 超时，设备可能掉线或 provider 卡死", "offline", ) conn.raise_for_status() data = conn.json() if not data.get("success"): raise STFError(f"远程连接失败: {data.get('description')}", "server") url = (data.get("remoteConnectUrl") or data.get("remoteAdbUrl") or data.get("remote_adb_url") or data.get("adbUrl") or data.get("url")) if not url: raise STFError("STF 返回成功但无 remoteConnectUrl", "server") return url except DeviceOfflineError: raise except requests.exceptions.ConnectionError as e: last_err = str(e) _log.warning(f"{serial} remoteConnect 网络异常 ({attempt}/{_RC_RETRIES}): {e}") if attempt < _RC_RETRIES: time.sleep(_RC_DELAY) except requests.exceptions.Timeout as e: last_err = str(e) _log.warning(f"{serial} remoteConnect 超时 ({attempt}/{_RC_RETRIES}): {e}") if attempt < _RC_RETRIES: time.sleep(_RC_DELAY) except requests.exceptions.RequestException as e: last_err = str(e) _log.error(f"{serial} remoteConnect 请求异常: {e}") raise STFError(f"远程连接请求失败: {e}", "server") from e # 重试耗尽 raise DeviceOfflineError( f"{serial} remoteConnect 重试 {_RC_RETRIES} 次失败: {last_err}", "offline", ) def remote_disconnect(self, serial): """断开远程 ADB 隧道。""" try: requests.post( f"{self.base_url}/api/v1/user/devices/{serial}/remoteDisconnect", headers=self.headers, timeout=_TIMEOUT, ) except requests.exceptions.RequestException: pass # 断开失败不影响主流程