import logging import time import subprocess import traceback import socket import os import requests from appium import webdriver from appium.webdriver.common.appiumby import AppiumBy from appium.webdriver.appium_service import AppiumService from appium.options.android import UiAutomator2Options from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, NoSuchElementException, InvalidSessionIdException, WebDriverException import globals.global_variable as global_variable def init_appium_driver(device_id, app_package="com.bjjw.cjgc", app_activity=".activity.LoginActivity"): """ 初始化Appium驱动的全局函数 参数: device_id: 设备ID app_package: 应用包名，默认为"com.bjjw.cjgc" app_activity: 应用启动Activity，默认为".activity.LoginActivity" 返回: (driver, wait): (WebDriver实例, WebDriverWait实例)，如果初始化失败则抛出异常 """ logging.info(f"设备 {device_id} 开始初始化Appium驱动") # 创建并配置Appium选项 options = UiAutomator2Options() options.platform_name = "Android" options.device_name = device_id options.app_package = app_package options.app_activity = app_activity options.automation_name = "UiAutomator2" options.no_reset = True options.auto_grant_permissions = True options.new_command_timeout = 28800 options.udid = device_id # 增加uiautomator2服务器启动超时时间 options.set_capability('uiautomator2ServerLaunchTimeout', 60000) # 60秒 # 增加连接超时设置 options.set_capability('connection_timeout', 120000) # 120秒 try: # 连接Appium服务器 # driver_url = "http://127.0.0.1:4723/wd/hub" # logging.info(f"设备 {device_id} 正在连接Appium服务器: {driver_url}") # driver = webdriver.Remote(driver_url, options=options) # logging.info(f"设备 {device_id} Appium服务器连接成功") driver_urls = [ "http://127.0.0.1:4723/wd/hub", # 标准路径 "http://127.0.0.1:4723", # 简化路径 "http://localhost:4723/wd/hub", # localhost ] driver = None last_exception = None # 尝试多个URL for driver_url in driver_urls: try: logging.info(f"设备 {device_id} 正在连接Appium服务器: {driver_url}") driver = webdriver.Remote(driver_url, options=options) logging.info(f"设备 {device_id} Appium服务器连接成功: {driver_url}") break # 连接成功，跳出循环 except Exception as e: last_exception = e logging.warning(f"设备 {device_id} 连接失败 {driver_url}: {str(e)[:100]}") continue # 检查是否连接成功 if not driver: logging.error(f"设备 {device_id} 所有Appium服务器地址尝试失败") logging.error(f"最后错误: {str(last_exception)}") raise Exception(f"设备 {device_id} 无法连接到Appium服务器: {str(last_exception)}") # 初始化等待对象 wait = WebDriverWait(driver, 20) logging.info(f"设备 {device_id} WebDriverWait初始化成功") # 等待应用稳定 time.sleep(2) # 设置屏幕永不休眠 try: # 使用ADB命令设置屏幕永不休眠 screen_timeout_cmd = [ "adb", "-s", device_id, "shell", "settings", "put", "system", "screen_off_timeout", "86400000" ] timeout_result = subprocess.run(screen_timeout_cmd, capture_output=True, text=True, timeout=15) if timeout_result.returncode == 0: logging.info(f"设备 {device_id} 已成功设置屏幕永不休眠") else: logging.warning(f"设备 {device_id} 设置屏幕永不休眠失败: {timeout_result.stderr}") except Exception as timeout_error: logging.warning(f"设备 {device_id} 设置屏幕永不休眠时出错: {str(timeout_error)}") logging.info(f"设备 {device_id} Appium驱动初始化完成") return driver, wait except Exception as e: logging.error(f"设备 {device_id} 初始化驱动失败: {str(e)}") logging.error(f"错误类型: {type(e).__name__}") logging.error(f"错误堆栈: {traceback.format_exc()}") # 如果驱动已创建，尝试关闭 if 'driver' in locals() and driver: try: driver.quit() except: pass raise def check_session_valid(driver, device_id=None): """ 检查当前会话是否有效 参数: driver: WebDriver实例 device_id: 设备ID（可选） 返回: bool: 会话有效返回True，否则返回False """ if device_id is None: device_id = global_variable.get_device_id() device_str = f"设备 {device_id} " if device_id else "" if not driver: logging.debug(f"{device_str}驱动实例为空") return False try: # 首先检查driver是否有session_id属性 if not hasattr(driver, 'session_id') or not driver.session_id: logging.debug(f"{device_str}驱动缺少有效的session_id") return False # 尝试获取当前上下文 current_context = driver.current_context logging.debug(f"{device_str}会话检查通过，当前上下文: {current_context}") return True except InvalidSessionIdException: logging.debug(f"{device_str}会话已失效") return False except WebDriverException as e: error_msg = str(e).lower() # 明确的会话失效错误 if any(phrase in error_msg for phrase in [ "session is either terminated or not started", "could not proxy command to the remote server", "socket hang up", "connection refused", "max retries exceeded" ]): logging.debug(f"{device_str}会话连接错误: {error_msg[:100]}") return False else: logging.debug(f"{device_str}WebDriver异常但可能不是会话失效: {error_msg[:100]}") return True except (ConnectionError, ConnectionRefusedError, ConnectionResetError) as e: logging.debug(f"{device_str}网络连接错误: {str(e)}") return False except Exception as e: error_msg = str(e) # 检查是否是连接相关错误 if any(phrase in error_msg.lower() for phrase in [ "10054", "10061", "connection", "connect", "refused", "urllib3" ]): logging.debug(f"{device_str}连接相关异常: {error_msg[:100]}") return False else: logging.debug(f"{device_str}检查会话时出现其他异常: {error_msg[:100]}") return True # 对于真正的未知异常，保守返回True def reconnect_driver(device_id, old_driver=None, app_package="com.bjjw.cjgc", app_activity=".activity.LoginActivity"): """ 重新连接Appium驱动，不重新启动应用 参数: device_id: 设备ID old_driver: 旧的WebDriver实例（可选） app_package: 应用包名 app_activity: 应用启动Activity 返回: (driver, wait): 新的WebDriver和WebDriverWait实例 """ # 使用传入的device_id或从全局变量获取 if not device_id: device_id = global_variable.get_device_id() # 修复device_id参数类型问题并使用全局设备ID作为备用 actual_device_id = device_id # 检查device_id是否为有效的字符串格式 if not actual_device_id or (isinstance(actual_device_id, str) and ("session=" in actual_device_id or len(actual_device_id.strip()) == 0)): # 尝试从old_driver获取设备ID if old_driver and hasattr(old_driver, 'capabilities'): capability_device_id = old_driver.capabilities.get('udid') if capability_device_id: actual_device_id = capability_device_id logging.warning(f"检测到device_id参数无效，已从old_driver中提取设备ID: {actual_device_id}") # 如果仍然没有有效的设备ID，使用全局变量 if not actual_device_id or (isinstance(actual_device_id, str) and ("session=" in actual_device_id or len(actual_device_id.strip()) == 0)): actual_device_id = global_variable.get_device_id() logging.warning(f"无法获取有效设备ID，使用全局变量GLOBAL_DEVICE_ID: {actual_device_id}") device_id = actual_device_id # 使用修正后的设备ID logging.info(f"设备 {device_id} 开始重新连接驱动（不重启应用）") # # 首先安全关闭旧驱动 # if old_driver: # safe_quit_driver(old_driver, device_id) max_reconnect_attempts = 3 reconnect_delay = 5 # 秒 for attempt in range(max_reconnect_attempts): try: logging.info(f"设备 {device_id} 第{attempt + 1}次尝试重新连接") # 确保Appium服务器运行 if not ensure_appium_server_running(): logging.warning(f"设备 {device_id} Appium服务器未运行，尝试启动") time.sleep(reconnect_delay) continue # 创建并配置Appium选项 - 重点：设置 autoLaunch=False 不自动启动应用 options = UiAutomator2Options() options.platform_name = "Android" options.device_name = device_id options.app_package = app_package options.app_activity = app_activity options.automation_name = "UiAutomator2" options.no_reset = True options.auto_grant_permissions = True options.new_command_timeout = 3600 options.udid = device_id # 关键设置：不自动启动应用 options.set_capability('autoLaunch', False) options.set_capability('skipUnlock', True) options.set_capability('skipServerInstallation', True) options.set_capability('skipDeviceInitialization', True) # 增加uiautomator2服务器启动超时时间 options.set_capability('uiautomator2ServerLaunchTimeout', 60000) # 60秒 # 增加连接超时设置 options.set_capability('connection_timeout', 120000) # 120秒 # 连接Appium服务器 logging.info(f"设备 {device_id} 正在连接Appium服务器（不启动应用）") driver = webdriver.Remote("http://localhost:4723", options=options) logging.info(f"设备 {device_id} Appium服务器连接成功") # 初始化等待对象 wait = WebDriverWait(driver, 20) logging.info(f"设备 {device_id} WebDriverWait初始化成功") # 不启动应用，直接附加到当前运行的应用 try: # 获取当前运行的应用 current_package = driver.current_package logging.info(f"设备 {device_id} 当前运行的应用: {current_package}") # 如果当前运行的不是目标应用，尝试切换到目标应用 if current_package != app_package: logging.info(f"设备 {device_id} 当前应用不是目标应用，尝试启动目标应用") launch_app_manually(driver, app_package, app_activity) else: logging.info(f"设备 {device_id} 已成功连接到运行中的目标应用") except Exception as attach_error: logging.warning(f"设备 {device_id} 获取当前应用信息失败: {str(attach_error)}") # 即使获取当前应用失败，也继续使用连接 # 验证新会话是否有效 if check_session_valid(driver, device_id): logging.info(f"设备 {device_id} 重新连接成功") return driver, wait else: logging.warning(f"设备 {device_id} 新创建的会话无效，将重试") safe_quit_driver(driver, device_id) except Exception as e: logging.error(f"设备 {device_id} 第{attempt + 1}次重新连接失败: {str(e)}") if attempt < max_reconnect_attempts - 1: wait_time = reconnect_delay * (attempt + 1) logging.info(f"设备 {device_id} 将在{wait_time}秒后重试重新连接") time.sleep(wait_time) else: logging.error(f"设备 {device_id} 所有重新连接尝试均失败") # 所有尝试都失败后才关闭旧驱动 if old_driver: safe_quit_driver(old_driver, device_id) raise # 所有尝试都失败 raise Exception(f"设备 {device_id} 重新连接失败，已尝试{max_reconnect_attempts}次") # def ensure_appium_server_running(port=4723): # """使用完整的环境变量启动Appium，解决路径缺失问题""" # try: # # 获取当前用户的环境变量 # env = os.environ.copy() # # ========== 仅修改这部分：Windows下Node.js/Appium常见路径 ========== # additional_paths = [ # # Windows默认Node.js安装路径（64位） # os.path.join("C:\\", "Program Files\\nodejs"), # # Windows32位Node.js路径 # os.path.join("C:\\", "Program Files (x86)\\nodejs"), # # npm全局安装路径（Windows核心，appium一般装在这里） # os.path.expanduser("~\\AppData\\Roaming\\npm"), # # 系统默认路径（防止基础命令缺失） # os.path.join("C:\\", "Windows\\System32"), # os.path.join("C:\\", "Windows\\System"), # # 自定义Node.js/npm路径（可选，根据你的实际安装路径加） # # "D:\\Program Files\\nodejs", # 若你装在D盘，解开注释并修改 # ] # # ========== Windows路径修改结束 ========== # # 更新PATH环境变量（跨系统通用，os.pathsep自动适配Windows的;和macOS的:） # current_path = env.get('PATH', '') # new_path = current_path + os.pathsep + os.pathsep.join(additional_paths) # env['PATH'] = new_path # # 构建启动命令（和原函数一致，无修改） # appium_cmd = f"appium -p {port} --log-level error" # # 使用完整环境启动（跨系统通用，Windows下正常执行） # process = subprocess.Popen( # appium_cmd, # shell=True, # Windows下字符串命令必须开启，和macOS一致 # env=env, # 传入补全后的Windows环境变量（核心） # stdout=subprocess.PIPE, # 捕获输出，控制台不刷屏 # stderr=subprocess.PIPE, # text=True # 输出为字符串，无需手动解码（跨系统通用） # ) # logging.info(f"Appium启动进程已创建，PID: {process.pid}") # # 等待并校验服务启动成功（需确保wait_for_appium_start已定义，timeout有值） # return wait_for_appium_start(port) # except Exception as e: # logging.error(f"Windows下启动Appium时出错: {str(e)}") # return False def is_port_in_use(port): """检查端口是否被占用""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: return s.connect_ex(('127.0.0.1', port)) == 0 def kill_system_process(process_name): """杀掉系统进程""" try: # Windows subprocess.run(f"taskkill /F /IM {process_name}", shell=True, stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL) except Exception: pass def start_appium_server(port=4723): """启动 Appium 服务，强制指定路径兼容性""" # 1. 先尝试清理可能占用的 node 进程 if is_port_in_use(port): logging.warning(f"端口 {port} 被占用，尝试清理 node.exe...") kill_system_process("node.exe") time.sleep(2) # 2. 构造启动命令 # 注意：这里增加了 --base-path /wd/hub 解决 404 问题 # --allow-cors 允许跨域，有时候能解决连接问题 appium_cmd = f"appium -p {port} --base-path /wd/hub --allow-cors" logging.info(f"正在启动 Appium: {appium_cmd}") try: # 使用 shell=True 在 Windows 上更稳定 # creationflags=subprocess.CREATE_NEW_CONSOLE 可以让它在后台运行不弹出窗口 subprocess.Popen(appium_cmd, shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) logging.info("Appium 启动命令已发送，等待服务就绪...") except Exception as e: logging.error(f"启动 Appium 进程失败: {e}") def check_server_status(port): """检测服务器状态，兼容 Appium 1.x 和 2.x 路径""" base_url = f"http://127.0.0.1:{port}" check_paths = ["/wd/hub/status", "/status"] # 优先检查 /wd/hub for path in check_paths: try: url = f"{base_url}{path}" response = requests.get(url, timeout=1) if response.status_code == 200: return True except: pass return False def ensure_appium_server_running(port=4723): """确保 Appium 服务器正在运行，如果没运行则启动它""" # 1. 第一次快速检测 if check_server_status(port): logging.info(f"Appium 服务已在端口 {port} 运行") return True # 2. 如果没运行，启动它 logging.warning(f"Appium 未在端口 {port} 运行，准备启动...") start_appium_server(port) # 3. 循环等待启动成功（最多等待 20 秒） max_retries = 20 for i in range(max_retries): if check_server_status(port): logging.info("Appium 服务启动成功并已就绪！") return True time.sleep(1) if i % 5 == 0: logging.info(f"等待 Appium 启动中... ({i}/{max_retries})") logging.error("Appium 服务启动超时！请检查 appium 命令是否在命令行可直接运行。") return False def wait_for_appium_start(port, timeout=10): """ 检测指定端口的Appium服务是否真正启动并可用 :param port: Appium服务端口 :param timeout: 最大等待时间（秒） :return: 服务就绪返回True，超时/失败返回False """ # Appium官方状态查询接口 check_url = f"http://localhost:{port}/wd/hub/status" # 检测开始时间 start_check_time = time.time() logging.info(f"开始检测Appium服务是否就绪，端口：{port}，最大等待{timeout}秒") while time.time() - start_check_time < timeout: try: # 发送HTTP请求，超时1秒（避免单次检测卡太久） response = requests.get( url=check_url, timeout=1, verify=False # 本地接口，禁用SSL验证 ) # 接口返回200（HTTP成功状态码），且JSON中status=0（Appium服务正常） if response.status_code == 200 and response.json().get("status") == 0: logging.info(f"Appium服务检测成功，端口{port}已就绪") return True except Exception as e: # 捕获所有异常（连接拒绝、超时、JSON解析失败等），说明服务未就绪 logging.debug(f"本次检测Appium服务未就绪：{str(e)}") # 调试日志，不刷屏 # 检测失败，休眠1秒后重试 time.sleep(1) # 循环结束→超时 logging.error(f"检测超时！{timeout}秒内Appium服务端口{port}仍未就绪") return False def safe_quit_driver(driver, device_id=None): """ 安全关闭驱动的全局函数 参数: driver: WebDriver实例 device_id: 设备ID（可选） """ if device_id is None: device_id = global_variable.get_device_id() device_str = f"设备 {device_id} " if device_id else "" logging.info(f"{device_str}开始关闭驱动") if not driver: logging.info(f"{device_str}没有可关闭的驱动实例") return # 检查driver是否为WebDriver实例或是否有quit方法 if not hasattr(driver, 'quit'): logging.warning(f"{device_str}驱动对象类型无效，不具有quit方法: {type(driver).__name__}") return max_quit_attempts = 3 for attempt in range(max_quit_attempts): try: logging.info(f"{device_str}尝试关闭驱动 (尝试 {attempt + 1}/{max_quit_attempts})") driver.quit() logging.info(f"{device_str}驱动已成功关闭") return except Exception as e: logging.error(f"{device_str}关闭驱动时出错 (尝试 {attempt + 1}/{max_quit_attempts}): {str(e)}") if attempt < max_quit_attempts - 1: # 等待一段时间后重试 wait_time = 2 logging.info(f"{device_str}将在 {wait_time} 秒后重试") time.sleep(wait_time) else: logging.critical(f"{device_str}尝试多次关闭驱动失败，可能导致资源泄漏") def is_app_launched(driver, package_name="com.bjjw.cjgc"): """ 检查应用是否已启动 参数: driver: WebDriver实例 package_name: 应用包名，默认为"com.bjjw.cjgc" 返回: bool: 如果应用已启动则返回True，否则返回False """ try: # 通过检查当前活动的包名来确认应用是否已启动 current_package = driver.current_package return current_package == package_name except Exception as e: logging.error(f"检查应用启动状态时出错: {str(e)}") return False def launch_app_manually(driver, device_id, package_name="com.bjjw.cjgc", activity=".activity.LoginActivity"): """ 手动启动应用 参数: driver: WebDriver实例 package_name: 应用包名，默认为"com.bjjw.cjgc" activity: 应用启动Activity，默认为".activity.LoginActivity" """ try: if not device_id: device_id = global_variable.get_device_id() # 尝试从driver获取设备ID if driver and hasattr(driver, 'capabilities'): device_id = driver.capabilities.get('udid') device_str = f"设备 {device_id} " if device_id else "" else: device_str = "" logging.info(f"{device_str}尝试手动启动应用: {package_name}/{activity}") # 首先使用ADB命令退出应用 if device_id: try: # 使用ADB命令强制停止应用 stop_cmd = [ "adb", "-s", device_id, "shell", "am", "force-stop", package_name ] stop_result = subprocess.run(stop_cmd, capture_output=True, text=True, timeout=15) if stop_result.returncode == 0: logging.info(f"{device_str}已使用ADB命令成功退出应用") else: logging.warning(f"{device_str}ADB退出应用失败: {stop_result.stderr}") except Exception as stop_error: logging.warning(f"{device_str}退出应用时出错: {str(stop_error)}") # 首先尝试使用driver的execute_script方法启动应用 try: if driver: driver.execute_script("mobile: startActivity", { "intent": f"{package_name}/{activity}" }) logging.info(f"{device_str}已使用Appium startActivity命令启动应用") except Exception as inner_e: logging.warning(f"{device_str}使用Appium startActivity命令失败: {str(inner_e)}，尝试使用ADB命令") # 如果device_id可用，使用ADB命令启动应用 if device_id: cmd = [ "adb", "-s", device_id, "shell", "am", "start", "-n", f"{package_name}/{activity}" ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=15) if result.returncode == 0: logging.info(f"{device_str}已使用ADB命令成功启动应用") else: logging.error(f"{device_str}ADB启动应用失败: {result.stderr}") else: logging.warning("无法获取设备ID，无法使用ADB命令启动应用") # 设置屏幕永不休眠 if device_id: try: # 使用ADB命令设置屏幕永不休眠 screen_timeout_cmd = [ "adb", "-s", device_id, "shell", "settings", "put", "system", "screen_off_timeout", "0" ] timeout_result = subprocess.run(screen_timeout_cmd, capture_output=True, text=True, timeout=15) if timeout_result.returncode == 0: logging.info(f"{device_str}已成功设置屏幕永不休眠") else: logging.warning(f"{device_str}设置屏幕永不休眠失败: {timeout_result.stderr}") except Exception as timeout_error: logging.warning(f"{device_str}设置屏幕永不休眠时出错: {str(timeout_error)}") # 等待应用启动 time.sleep(5) except Exception as e: logging.error(f"手动启动应用时出错: {str(e)}") logging.error(f"错误堆栈: {traceback.format_exc()}") def go_main_click_tabber_button(driver, device_id, tabber_button_text, max_retries=3): """ 跳转到主页面并点击对应的导航菜单按钮（带重试机制） 参数: driver: WebDriver实例 device_id: 设备ID tabber_button_text: 导航菜单按钮的文本 max_retries: 最大重试次数 返回: bool: 成功返回True，失败返回False """ retry_count = 0 while retry_count < max_retries: try: logging.info(f"设备 {device_id} 第 {retry_count + 1} 次尝试执行导航操作") if not check_session_valid(driver, device_id): logging.warning(f"设备 {device_id} 会话无效，尝试重新连接驱动...") try: # 重新连接，获取新的driver new_driver, _ = reconnect_driver(device_id, driver) driver = new_driver # 更新driver引用 logging.info(f"设备 {device_id} 驱动重连成功") except Exception as e: logging.error(f"设备 {device_id} 驱动重连失败: {str(e)}") retry_count += 1 if retry_count < max_retries: time.sleep(2) continue else: return False # 检查当前是否已经在主页面 current_activity = driver.current_activity logging.info(f"设备 {device_id} 当前Activity: {current_activity}") if ".activity.MainActivity" in current_activity: logging.info(f"设备 {device_id} 已在主页面") else: logging.info(f"设备 {device_id} 当前不在主页面，循环点击返回按钮，直到在主页面") max_back_presses = 10 # 最大返回键次数 back_press_count = 0 while ".activity.MainActivity" not in current_activity and back_press_count < max_back_presses: try: if not check_session_valid(driver, device_id): logging.warning(f"设备 {device_id} 会话无效，尝试重新连接驱动...") if not reconnect_driver(device_id, driver): logging.error(f"设备 {device_id} 驱动重连失败") # 点击返回按钮 driver.back() back_press_count += 1 time.sleep(1) # 更新当前Activity current_activity = driver.current_activity logging.info(f"设备 {device_id} 点击返回按钮 {back_press_count} 次后，当前Activity: {current_activity}") except Exception as inner_e: logging.warning(f"设备 {device_id} 点击返回按钮时出错: {str(inner_e)}") break # 检查是否成功回到主页面 if ".activity.MainActivity" not in current_activity: logging.warning(f"设备 {device_id} 无法回到主页面，当前Activity: {current_activity}") # 不立即返回，继续重试逻辑 retry_count += 1 if retry_count < max_retries: logging.info(f"设备 {device_id} 等待2秒后重试...") time.sleep(2) continue else: logging.error(f"设备 {device_id} 达到最大重试次数，无法回到主页面") return False # 现在已经在主页面，点击指定的导航菜单按钮 # logging.info(f"设备 {device_id} 已在主页面，尝试点击导航菜单按钮: {tabber_button_text}") try: tabber_button = driver.find_element(AppiumBy.ID, tabber_button_text) # 点击按钮 tabber_button.click() tabber_button.click() logging.info(f"设备 {device_id} 已成功点击导航菜单按钮: {tabber_button_text}") # 等待页面加载 time.sleep(2) # 验证操作是否成功 # 可以添加一些验证逻辑，比如检查是否跳转到目标页面 new_activity = driver.current_activity logging.info(f"设备 {device_id} 点击后当前Activity: {new_activity}") return True except TimeoutException: logging.error(f"设备 {device_id} 等待导航菜单按钮 '{tabber_button_text}' 超时") except Exception as e: logging.error(f"设备 {device_id} 点击导航菜单按钮 '{tabber_button_text}' 时出错: {str(e)}") # 检查会话有效性并尝试重连 if not check_session_valid(driver, device_id): logging.warning(f"设备 {device_id} 会话无效，尝试重新连接驱动...") if reconnect_driver(device_id, driver): logging.info(f"设备 {device_id} 驱动重连成功，继续重试") # 重连成功后继续循环 retry_count += 1 if retry_count < max_retries: time.sleep(2) continue else: logging.error(f"设备 {device_id} 驱动重连失败") return False else: # 会话有效但点击失败，可能是页面元素问题 logging.warning(f"设备 {device_id} 会话有效但点击失败，可能是页面加载问题") # 如果点击按钮失败，增加重试计数 retry_count += 1 if retry_count < max_retries: logging.info(f"设备 {device_id} 点击按钮失败，等待2秒后第 {retry_count + 1} 次重试...") time.sleep(2) else: logging.error(f"设备 {device_id} 达到最大重试次数 {max_retries}，导航失败") return False except Exception as e: logging.error(f"设备 {device_id} 第 {retry_count + 1} 次尝试时出错: {str(e)}") retry_count += 1 if retry_count < max_retries: logging.info(f"设备 {device_id} 等待2秒后重试...") time.sleep(2) else: logging.error(f"设备 {device_id} 达到最大重试次数 {max_retries}，导航失败") return False return False def check_connection_error(exception): """检查是否为连接拒绝错误""" error_str = str(exception) connection_errors = [ '远程主机强迫关闭了一个现有的连接', '由于目标计算机积极拒绝，无法连接', 'ConnectionResetError', 'NewConnectionError', '10054', '10061' ] return any(error in error_str for error in connection_errors) def restart_appium_server(port=4723): """重启Appium服务器""" try: # 杀死可能存在的Appium进程 subprocess.run(['taskkill', '/f', '/im', 'node.exe'], capture_output=True, shell=True) time.sleep(2) # 启动Appium服务器 # appium_command = f'appium -p {port}' appium_command = f"appium -p {port} --base-path /wd/hub" subprocess.Popen(appium_command, shell=True) # 等待Appium启动 time.sleep(10) return True except Exception as e: print(f"重启Appium服务器失败: {str(e)}") return False def is_appium_running(port=4723): """检查Appium服务器是否在运行""" try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: result = s.connect_ex(('127.0.0.1', port)) return result == 0 except: return False