refactor: 优化 Appium 服务管理逻辑与进程清理

- 弃用 npm run 改为直接调用 appium.cmd，消除 Windows 进程残留。
- 统一 _cleanup_process_tree 逻辑，确保异常退出时无孤儿进程。
- 重构 start_appium_service 为单一循环状态机，增强启动探测的健壮性。
- 新增 with_appium 装饰器，确保无论测试结果如何均能干净退出

main.py

@@ -3,41 +3,41 @@ import time
 from appium import webdriver
 from appium.options.android import UiAutomator2Options
 
-from run_appium import start_appium_service, stop_appium_service
+from run_appium import start_appium_service, stop_appium_service, with_appium
 
-# 在自动化套件启动前执行
-proc = start_appium_service()
 
-# 配置Android设备参数
+@with_appium
-capabilities = dict(
+def main(service):
+    print(f"正在测试，服务模式: {service.role}")
+    # 简单操作示例
+    driver = None
+    try:
+        # 在自动化套件启动前执行
+        # proc = start_appium_service()
+
+        # 配置Android设备参数
+        capabilities = dict(
             platformName='Android',
             automationName='uiautomator2',
             deviceName='Android',
             appPackage='com.android.settings',
             appActivity='.Settings'
-)
+        )
 
-# 转换capabilities为Appium Options
+        # 转换capabilities为Appium Options
-options = UiAutomator2Options().load_capabilities(capabilities)
+        options = UiAutomator2Options().load_capabilities(capabilities)
 
-# 连接Appium服务器
+        # 连接Appium服务器
-# driver = webdriver.Remote('http://localhost:4723', options=options)
+        # driver = webdriver.Remote('http://localhost:4723', options=options)
-driver = webdriver.Remote('http://127.0.0.1:4723', options=options)
+        driver = webdriver.Remote('http://127.0.0.1:4723', options=options)
-
-
-def main():
-    # 简单操作示例
-    try:
         time.sleep(1)
         print("当前Activity:", driver.current_activity)
     finally:
         driver.quit()
         # 在自动化套件结束后执行
-        stop_appium_service(proc)
+        # stop_appium_service(proc)
     print("Hello from AppAutoTest!")
 
 
-
-
 if __name__ == "__main__":
     main()

run_appium.py

@@ -9,6 +9,8 @@
 @date: 2026/1/12 10:21
 @desc: 
 """
+import functools
+import signal
 import subprocess
 import time
 import os
@@ -16,8 +18,9 @@ import sys
 import http.client
 import socket
 import json
+from collections import namedtuple
 from pathlib import Path
-from enum import Enum, auto
+from enum import Enum
 
 # --- 核心配置 ---
 # 使用 pathlib 获取当前脚本所在的绝对路径
@@ -25,8 +28,10 @@ BASE_DIR = Path(__file__).resolve().parent
 
 APPIUM_HOST = "127.0.0.1"
 APPIUM_PORT = 4723
+
+
 # 使用 npm run 确保 APPIUM_HOME=. 变量和本地版本生效
-NODE_CMD = f"npm run appium -- -p {APPIUM_PORT}"
+# NODE_CMD = f"npm run appium -- -p {APPIUM_PORT}"
 
 
 class AppiumStatus(Enum):
@@ -39,6 +44,80 @@ class AppiumStatus(Enum):
     UNKNOWN = "未知状态"
 
 
+class ServiceRole(Enum):
+    """服务角色枚举：明确权责"""
+    MANAGED = "由脚本启动 (受控模式)"  # 需要负责清理
+    EXTERNAL = "复用外部服务 (共享模式)"  # 不负责清理
+    NULL = "无效服务 (空模式)"  # 占位或失败状态
+
+
+def get_appium_command():
+    """精确定位 Appium 执行文件，避免 npm 包装层"""
+    bin_name = "appium.cmd" if sys.platform == "win32" else "appium"
+    appium_bin = BASE_DIR / "node_modules" / ".bin" / bin_name
+
+    if not appium_bin.exists():
+        # 报错提示：找不到本地 Appium，引导用户安装
+        print(f"\n❌ 错误: 在路径 {appium_bin} 未找到 Appium 执行文件。")
+        print("💡 请确保已在项目目录下执行过: npm install appium")
+        sys.exit(1)
+    # 返回执行列表（用于 shell=False）
+    return [str(appium_bin), "-p", str(APPIUM_PORT)]
+
+
+# 在全局或 start_appium_service 中获取命令
+APP_CMD_LIST = get_appium_command()
+
+
+def _cleanup_process_tree(process: subprocess.Popen = None):
+    """核心清理逻辑：针对方案一的跨平台递归关闭"""
+    if not process or process.poll() is not None:
+        return
+    print(f"\n🛑 正在关闭 Appium 进程树 (PID: {process.pid})...")
+    try:
+        if sys.platform == "win32":
+            # Windows 下使用 taskkill 强制关闭进程树 /T
+            subprocess.run(['taskkill', '/F', '/T', '/PID', str(process.pid)],
+                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        else:
+            # Unix/Mac: 获取进程组 ID 并发送终止信号
+            pgid = os.getpgid(process.pid)
+            os.killpg(pgid, signal.SIGTERM)
+
+        process.wait(timeout=5)
+        print("✅ 所有相关进程已安全退出。")
+    except Exception as e:
+        print(f"⚠️ 关闭进程时遇到意外: {e}")
+        try:
+            process.kill()
+        except Exception as e:
+            print(f"强制: {e}")
+            # 这里通常保持安静，因为我们已经尝试过清理了
+    print("✅ 服务已完全清理。")
+
+
+class AppiumService:
+    """Appium 服务上下文容器"""
+
+    def __init__(self, role: ServiceRole, process: subprocess.Popen = None):
+        self.role = role
+        self.process = process
+
+    def stop(self):
+        """统一停止接口：根据角色决定是否关闭进程"""
+        match self.role:
+            case ServiceRole.EXTERNAL:
+                print(f"--> [角色: {self.role.value}] 脚本退出，保留原服务运行。")
+                return
+            case ServiceRole.MANAGED:
+                _cleanup_process_tree(self.process)
+            case ServiceRole.NULL:
+                print(f"--> [角色: {self.role.value}] 无需清理。")
+
+    def __repr__(self):
+        return f"<AppiumService 角色='{self.role.value}' 端口={APPIUM_PORT}>"
+
+
 def get_appium_status() -> AppiumStatus:
     """深度探测 Appium 状态"""
     conn = None
@@ -58,7 +137,6 @@ def get_appium_status() -> AppiumStatus:
         is_ready = data.get("value", {}).get("ready")
         return AppiumStatus.READY if is_ready else AppiumStatus.INITIALIZING
 
-
     except (socket.error, ConnectionRefusedError):
         # 3. 如果通信拒绝，检查端口是否真的空闲
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
@@ -69,7 +147,7 @@ def get_appium_status() -> AppiumStatus:
             except OSError:
                 return AppiumStatus.CONFLICT  # 端口被占用但没响应 HTTP
     except (http.client.HTTPException, json.JSONDecodeError):
-        return AppiumStatus.CONFLICT
+        return AppiumStatus.UNKNOWN
     except Exception as e:
         print(e)
         return AppiumStatus.ERROR
@@ -77,18 +155,26 @@ def get_appium_status() -> AppiumStatus:
         if conn: conn.close()
 
 
-def start_appium_service():
+def start_appium_service() -> AppiumService:
     """管理 Appium 服务的生命周期"""
-    # if check_before_start():
-    #     return None
     process = None  # 1. 预先初始化变量，防止作用域错误
+    managed = False
+    # 轮询等待真正就绪
+    max_retries = 40
+
+    for i in range(max_retries):
         status = get_appium_status()
         match status:  # Python 3.10+ 的模式匹配
             case AppiumStatus.READY:
+                if managed:
+                    # 安全打印 PID
+                    pid_str = f"PID: {process.pid}" if process else "EXTERNAL"
+                    print(f"✨ Appium 已经完全就绪! ({pid_str})")
+                    return AppiumService(ServiceRole.MANAGED, process)
+                else:
                     print(f"--> [复用] 有效的 Appium 服务已在运行 (Port: {APPIUM_PORT})")
-            return None
+                    print("      [注意] 脚本退出时将保留该服务，不会将其关闭。")
-        case AppiumStatus.INITIALIZING:
+                    return AppiumService(ServiceRole.EXTERNAL, process)
-            print("⏳ Appium 正在初始化，等待...")
             case AppiumStatus.CONFLICT:
                 print(f"\n[!] 错误: 端口 {APPIUM_PORT} 被非 Appium 程序占用。")
                 print("=" * 60)
@@ -101,6 +187,8 @@ def start_appium_service():
                 print("=" * 60)
                 sys.exit(1)
             case AppiumStatus.OFFLINE:
+
+                if not managed:
                     print("🔌 Appium 未启动")
                     print(f"🚀 正在准备启动本地 Appium 服务 (Port: {APPIUM_PORT})...")
 
@@ -110,63 +198,60 @@ def start_appium_service():
 
                     try:
                         process = subprocess.Popen(
-                    NODE_CMD,
+                            APP_CMD_LIST,
-                    shell=True,
                             stdout=subprocess.DEVNULL,
                             stderr=subprocess.DEVNULL,
                             env=env_vars,
-                    cwd=BASE_DIR
+                            cwd=BASE_DIR,
+                            # Windows 和 Linux/Mac 的处理方式不同
+                            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0,
+                            preexec_fn=os.setsid if sys.platform != "win32" else None
                         )
-
+                        managed = True
                     except Exception as e:
                         print(f"❌ 启动过程发生异常: {e}")
                         sys.exit(1)
-        case _:
+                else:
-            print("Appium 启动异常")
-            sys.exit(1)
-    # 轮询等待真正就绪
-    max_retries = 40
-    for i in range(max_retries):
-        status = get_appium_status()
-
-        if status == AppiumStatus.READY:
-            # 安全打印 PID
-            pid_str = f"PID: {process.pid}" if process else "EXTERNAL"
-            print(f"✨ Appium 已经完全就绪! ({pid_str})")
-            return process
-
-        if status == AppiumStatus.ERROR:
-            print("❌ 探测接口发生内部错误（可能是解析失败或严重网络异常），脚本终止。")
-            if process: process.terminate()
-            sys.exit(1)
-
-        if status == AppiumStatus.INITIALIZING:
-            if i % 4 == 0:
-                print("...Appium 正在加载驱动/插件，请稍候...")
-
-        if status == AppiumStatus.OFFLINE:
-            # 仅当进程是我们启动的（process 不为 None）才检查崩溃(None: 程序正常运行，非None: 程序异常)
                     if process and process.poll() is not None:
                         print("❌ Appium 进程启动后异常退出。")
                         sys.exit(1)
+            case AppiumStatus.INITIALIZING:
+                if managed and process and process.poll() is not None:
+                    print("❌ Appium 驱动加载期间进程崩溃。")
+                    _cleanup_process_tree(process)
+                    sys.exit(1)
+                if i % 4 == 0:  # 每 2 秒提醒一次，避免刷屏
+                    print("⏳ Appium 正在加载驱动/插件，请稍候...")
+            case AppiumStatus.ERROR:
+                print("❌ 探测接口发生内部错误（可能是解析失败或严重网络异常），脚本终止。")
+                if managed and process:
+                    _cleanup_process_tree(process)
+                sys.exit(1)
+            case _:
+                print("Appium 启动异常")
+                sys.exit(1)
 
         time.sleep(0.5)
 
     print("❌ 启动超时：Appium 在规定时间内未完成初始化。")
-    if process: process.terminate()
+    _cleanup_process_tree(process)
     sys.exit(1)
 
 
-def stop_appium_service(process):
+def stop_appium_service(server: AppiumService):
-    """安全关闭服务"""
+    # """安全关闭服务"""
-    if process and process.poll() is None:
+    server.stop()
-        print(f"\n🛑 正在关闭 Appium 服务 (PID: {process.pid})...")
+
-        process.terminate()
+# --- 装饰器实现 ---
+def with_appium(func):
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        service = start_appium_service()
         try:
-            process.wait(timeout=5)
+            return func(service, *args, **kwargs)
-        except subprocess.TimeoutExpired:
+        finally:
-            process.kill()
+            service.stop()
-        print("✅ 服务已安全退出。")
+    return wrapper
 
 
 if __name__ == "__main__":