import uiautomator2 as u2 import time import subprocess import random from typing import Optional, Tuple, Union def find_template_position(small_img_path, big_img_path, threshold=0.8): """ 在小图中查找大图的位置，返回中心坐标 Args: small_img_path: 小块碎片路径（要查找的图片） big_img_path: 整张图片路径（被查找的图片） threshold: 匹配阈值 0-1，默认0.8 Returns: (x, y): 中心坐标，如果没找到返回 None """ # 1. 读取图片 small = cv2.imread(small_img_path) big = cv2.imread(big_img_path) if small is None or big is None: print("❌ 图片读取失败") return None # 2. 模板匹配 result = cv2.matchTemplate(big, small, cv2.TM_CCOEFF_NORMED) min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result) print(f"匹配度: {max_val:.3f}") # 3. 判断是否匹配成功 if max_val < threshold: print(f"❌ 未找到匹配 (阈值: {threshold})") return None # 4. 计算中心坐标 h, w = small.shape[:2] center_x = max_loc[0] + w // 2 center_y = max_loc[1] + h // 2 return (center_x, center_y) class SafeExecutor: """ 安全执行器：为 uiautomator2 提供重试、降级、随机延迟等能力 所有方法失败时直接抛出异常，不返回结果对象 """ def __init__(self,device_id, default_retry: int = 2, human_delay: bool = True): """ 参数: driver: uiautomator2 连接对象 default_retry: 默认重试次数 human_delay: 是否添加随机延迟（模拟人类操作） """ self.code = 5 self.device_id = device_id self.driver = u2.connect_usb(self.device_id) self.default_retry = default_retry self.human_delay = human_delay w, h = self.driver.window_size() self.h = h # ========== 私有工具方法 ========== def _random_sleep(self, min_sec: float = 0.05, max_sec: float = 0.2): """随机睡眠，模拟人类操作间隔""" if self.human_delay: time.sleep(random.uniform(min_sec, max_sec)) def _wait_before_retry(self, attempt: int): """重试前等待（指数退避）""" wait_time = min(0.5 * (2 ** attempt), 3) time.sleep(wait_time) # ========== 基础操作 ========== def safe_click(self,xpath): try: if self.No_slider(): if type(xpath)!=str: elem = xpath else: elem = self.driver.xpath(xpath) if not elem.exists: return False elem.click() time.sleep(2) if self.No_slider(): return True else: return False except Exception as exc: print(f"点击失败，已跳过: {xpath}, error: {exc}") return False def tap(self, x: int, y: int, retry: Optional[int] = None): """ 点击坐标，带重试 失败抛出异常 """ try: self.driver.click(x, y) self._random_sleep(0.1, 0.3) return # 成功，直接返回 except Exception as e: raise Exception(f"点击失败 {x},{y},{e}") def swipe(self,distance,times=2): """ 向上滑动指定距离，分多次完成 Args: distance: 总滑动距离（如 1200） times: 分成几次（默认2次） """ w, h = self.driver.window_size() x = w // 2 + random.randint(-30, 30) # X轴随机偏移 start_y = int(h * 0.7) + random.randint(-20, 20) # 起始位置随机偏移 per_distance = distance // times current_y = start_y # print(f"📏 向上滑动 {distance}px，分 {times} 次") for i in range(times): # 每次滑动距离随机波动 if i == times - 1: end_y = current_y - (distance - (per_distance * (times - 1))) else: # 每段距离 ±20px 波动 seg_distance = per_distance + random.randint(-15, 15) end_y = current_y - seg_distance end_y = max(50, end_y) # 滑动时长随机 duration = random.uniform(0.06, 0.12) # print(f" 第{i+1}次: {current_y} → {end_y}") self.driver.drag(x, current_y, x, end_y, duration=duration) if i < times - 1: # 暂停时间随机 pause_time = random.uniform(0.2, 0.5) #print(f" ⏸️ 暂停 {pause_time:.2f}s") time.sleep(pause_time) current_y = end_y def input_text(self,xpaths,text): """ 输入文本，带重试 失败抛出异常 """ try: search_input = self.driver.xpath(xpaths) if search_input.exists: search_input.set_text('') time.sleep(1) # 模拟打字间隔 search_input.set_text(text) return True return False except Exception as e: raise Exception(f"输入失败 '{text}'，: {e}") def is_connected(self) -> bool: """检查与设备的连接是否正常""" try: # 方式1：检查 atx-agent 是否响应 info = self.driver.info # 会请求 atx-agent return True except Exception: self._recover_service() return False def _recover_service(self): #重新连接 # 停止 atx-agent subprocess.run(["adb", "-s", self.device_id, "shell", "/data/local/tmp/atx-agent", "server", "-d", "--stop"], capture_output=True, timeout=5) time.sleep(1) # 启动 atx-agent subprocess.run(["adb", "-s", self.device_id, "shell", "/data/local/tmp/atx-agent", "server", "-d"], capture_output=True, timeout=5) time.sleep(2) # 重新连接 uiautomator2 self.driver = u2.connect_usb(self.device_id) def No_slider(self) -> bool: """检测当前是否有滑块（默认实现，可覆盖）""" # 这里用你实际的滑块检测逻辑 # 示例：检测特定图片或UI元素 for i in range(4): if '检测成功' in '检测成功': self._solve_slider() else: break return True return False def _solve_slider(self): """解决滑块验证（默认实现，可覆盖""" num = 0 while(self.driver.xpath('//*[@text="检测到您的账号存在风险，为保障账号安全，请进行安全验证"]').exists or self.driver.xpath('//*[@text="身份安全验证"]').exists): self.driver.press('back') time.sleep(0.3) num+=1 if(num>=6): break time.sleep(random.uniform(2, 3)) if self.driver.xpath('//*[@text="检测到您的账号存在风险，为保障账号安全，请进行安全验证"]').exists or self.driver.xpath('//*[@text="身份安全验证"]').exists: self._solve_slider() if(self.driver.xpath('//*[@text="验证码"]').exists): print('其他验证码,无xpath') self.driver.screenshot(f'test{code}.jpg') code+=1 # self.driver.screenshot('test.jpg') #time.sleep(1) # x= find_template_position('t.jpg','test.jpg') # self.driver.click(x[0],x[1]) # 这里写你的滑块处理逻辑 # 例如：找到滑块位置，模拟滑动轨迹 pass def safe_get_all(self,xpath): try: return self.driver.xpath(xpath).all() except Exception as exc: print(f"获取元素列表失败，已跳过: {xpath}, error: {exc}") return [] def safe_exists(self,xpath): try: return self.driver.xpath(xpath).exists except Exception as exc: print(f"xpath不存在或检查失败，已跳过: {xpath}, error: {exc}") return False def ele(self,xpath): try: return self.driver.xpath(xpath) except Exception as exc: print(f"xpath不存在或检查失败，已跳过: {xpath}, error: {exc}") return False def safe_get_text(self,xpath): try: elem = self.driver.xpath(xpath) if not elem.exists: return "" node = elem.get() return (node.text or "").strip() if node else "" except Exception as exc: print(f"获取文本失败，已跳过: {xpath}, error: {exc}") return "" def safe_press_back(self): try: self.driver.press('back') return True except Exception as exc: print(f"safe_press_back 返回失败: {exc}") return False