project/tbsg/steps/executor.py

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