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, jitter: int = 15):
        """
        点击坐标，带重试；落点在目标 ±15px 内随机偏移（拟人，避免每次点同一像素）
        失败抛出异常
        """

        try:
            jx = max(0, x + random.randint(-jitter, jitter))
            jy = max(0, y + random.randint(-jitter, jitter))
            self.driver.click(jx, jy)
            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