utils.py

import time
import random
import io
import re
import cv2
import numpy as np
import ddddocr
import subprocess
import ssl
import urllib.request
import torch
print("PyTorch 版本:", torch.version)
print("CUDA 是否可用:", torch.cuda.is_available())
if hasattr(torch.backends, "mps"):
    print("MPS 是否可用:", torch.backends.mps.is_available())
else:
    print("MPS 不受支持")

# 正确的SSL禁用方式：赋值为「调用后的上下文对象」，而非函数本身
original_context = ssl._create_default_https_context
ssl._create_default_https_context = ssl._create_unverified_context()  # 关键：加()调用函数

# 额外配置urllib的opener，双重确保跳过SSL验证
opener = urllib.request.build_opener(
    urllib.request.HTTPSHandler(context=ssl._create_unverified_context())
)
urllib.request.install_opener(opener)

# from paddleocr import PaddleOCR
from PIL import Image
import easyocr
easyocr_reader = easyocr.Reader(['ch_sim', 'en'], gpu=True)


# 关闭 ppocr 的所有日志（推荐）
# logging.getLogger('ppocr').setLevel(logging.WARNING)  # 或 logging.ERROR
# 或者更粗暴地全局关闭 DEBUG 以下级别（会影响其他库）
# logging.basicConfig(level=logging.WARNING)

TB_APP = "com.taobao.taobao"
ALIPAY_APP = "com.eg.android.AlipayGphone"
FISH_APP = "com.taobao.idlefish"
TMALL_APP = "com.tmall.wireless"

# 应用启动配置，键为包名，值为activity
APP_START_CONFIG = {
    TB_APP: "com.taobao.tao.welcome.Welcome",
    FISH_APP: "com.taobao.fleamarket.home.activity.InitActivity",
    TMALL_APP: "com.tmall.wireless.maintab.module.TMMainTabActivity",
    ALIPAY_APP: "com.eg.android.AlipayGphone.AlipayLogin"  # 默认配置，不指定activity
}


def check_chars_exist(text, chars=None):
    if chars is None:
        chars = ["拉好友", "抢红包", "搜索兴趣商品下单", "买精选商品", "全场3元3件", "固定入口", "农场小游戏", "砸蛋","大众点评", "蚂蚁新村", "消消乐", "3元抢3件包邮到家", "拍一拍", "1元抢爆款好货", "拉1人助力","玩消消乐", "下单即得", "添加签到神器", "下单得肥料", "88VIP", "邀请好友", "好货限时直降", "连连消","下单即得", "拍立淘", "玩任意游戏", "首页回访", "百亿外卖", "玩趣味游戏得大额体力", "天猫积分换体力", "头条刷热点", "一淘签到", "每拉", "闪购拿大额补贴", "开心消消乐过1关", "通关", "购买商品", "去闪购领红包点外卖", "冒险大作战", "斗地主", "买限时折扣好物", "趣头条", "(1000/3500)", "任意下单", "农场对对碰匹配", "任意充值"]
    for char in chars:
        if char in text:
            return True
    return False


def get_current_app(d):
    info = d.shell("dumpsys window | grep mCurrentFocus").output
    match = re.search(r'mCurrentFocus=Window\{.*? u0 (.*?)/(.*?)\}', info)
    if match:
        package_name = match.group(1)
        activity_name = match.group(2)
        return package_name, activity_name
    return None, None


other_app = ["蚂蚁森林", "农场", "百度", "支付宝", "芝麻信用", "蚂蚁庄园", "闲鱼", "神奇海洋", "淘宝特价版", "点淘", "饿了么", "微博", "直播", "领肥料礼包", "福气提现金", "看小说", "菜鸟", "斗地主", "领肥料礼包"]


def fish_not_click(text, chars=None):
    if chars is None:
        chars = ["发布一件新宝贝", "买到或卖出", "中国移动", "视频", "下单", "点淘", "一淘", "收藏", "购买"]
    for char in chars:
        if char in text:
            return True
    return False


def find_button(image, btn_path, region=None):
    template = cv2.imread(btn_path)
    # 如果指定了区域，裁剪图像
    if region is not None:
        x, y, w_region, h_region = region
        image = image[y:y + h_region, x:x + w_region]
    # 转换为灰度图像
    screenshot_gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    template_gray = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY)
    # 获取模板图像的宽度和高度
    w, h = template_gray.shape[::-1]
    # 使用模板匹配
    res = cv2.matchTemplate(screenshot_gray, template_gray, cv2.TM_CCOEFF_NORMED)
    threshold = 0.7
    loc = np.where(res >= threshold)
    for pt in zip(*loc[::-1]):
        return pt
    return None


# 基于特征查找图片 threshold:匹配阈值（越高质量要求越高） scales:搜索的尺度范围 60%~140%
def find_button_multiscale(screen_shot, template_path, scales=np.linspace(0.6, 1.4, 20), threshold=0.78, method=cv2.TM_CCOEFF_NORMED):
    # 读取图片（建议都转成RGB或灰度，视情况）
    template = cv2.imread(template_path)
    if screen_shot is None or template is None:
        return None, None, None
    if isinstance(screen_shot, Image.Image):
        screen_shot = cv2.cvtColor(np.array(screen_shot), cv2.COLOR_RGB2BGR)
    elif isinstance(screen_shot, bytes):
        img = Image.open(io.BytesIO(screen_shot))
        screen_shot = cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)
    # 建议都转灰度（速度快很多，且很多按钮是单色/对比度强的）
    large_gray = cv2.cvtColor(screen_shot, cv2.COLOR_BGR2GRAY)
    tpl_gray = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY)
    h, w = tpl_gray.shape[:2]
    best_val = -1
    best_loc = None
    best_scale = 1.0
    best_rect = None
    for scale in scales:
        # 缩放模板（注意：也可以反过来缩放大图，但通常缩放小图更快）
        resize_w = int(w * scale)
        resize_h = int(h * scale)
        if resize_w < 5 or resize_h < 5:
            continue
        resized_tpl = cv2.resize(tpl_gray, (resize_w, resize_h), interpolation=cv2.INTER_AREA if scale < 1 else cv2.INTER_CUBIC)
        # 检查是否还能匹配（防止模板比大图还大）
        if resized_tpl.shape[0] > large_gray.shape[0] or resized_tpl.shape[1] > large_gray.shape[1]:
            continue
        # 模板匹配
        result = cv2.matchTemplate(large_gray, resized_tpl, method)
        min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result)
        if method in [cv2.TM_SQDIFF, cv2.TM_SQDIFF_NORMED]:
            val = min_val
            loc = min_loc
        else:
            val = max_val
            loc = max_loc
        if val > best_val:  # 对于相关系数类方法，越大越好
            best_val = val
            best_loc = loc
            best_scale = scale
            best_rect = (loc[0], loc[1], resize_w, resize_h)
    if best_val >= threshold:
        x, y, bw, bh = best_rect
        center_x = x + bw // 2
        center_y = y + bh // 2
        print(f"找到匹配！置信度: {best_val:.3f}")
        print(f"左上角: ({x}, {y})")
        print(f"中心点: ({center_x}, {center_y})")
        print(f"按钮大小: {bw}×{bh}  (缩放比例 {best_scale:.2f})")
        # 可视化（可选）
        cv2.rectangle(screen_shot, (x, y), (x + bw, y + bh), (0, 255, 0), 2)
        cv2.circle(screen_shot, (center_x, center_y), 5, (0, 0, 255), -1)
        cv2.imwrite("result.jpg", screen_shot)
        return (center_x, center_y), best_val, best_scale
    else:
        print(f"未找到足够匹配，最高置信度仅: {best_val:.3f}")
        return None, best_val, None


def find_text_position(image, text):
    ocr = ddddocr.DdddOcr(show_ad=False)
    ocr_result = ocr.classification(image)
    # 将 OCR 结果按行解析
    lines = ocr_result.split('\n')
    # 遍历每一行，查找目标文本的位置
    for line in lines:
        if text in line:
            # 获取文本的位置
            start_index = line.find(text)
            end_index = start_index + len(text)
            return start_index, end_index
    return None


def find_text_by_easyocr(screenshot, target_text):
    results = easyocr_reader.readtext(screenshot, detail=1)  # detail=1返回详细信息
    for bbox, text, confidence in results:
        if target_text in text:
            print(f"找到文字: '{text}' (置信度: {confidence:.2f})")
            # 获取边界框坐标
            # bbox格式: [[x1, y1], [x2, y2], [x3, y3], [x4, y4]]
            # 转换为(x, y, width, height)
            x_coord = [point[0] for point in bbox]
            y_coord = [point[1] for point in bbox]
            x_min = int(min(x_coord))
            x_max = int(max(x_coord))
            y_min = int(min(y_coord))
            y_max = int(max(y_coord))
            width = x_max - x_min
            height = y_max - y_min
            x_center = x_min + width // 2
            y_center = y_min + height // 2
            print(f"位置: x={x_min}, y={y_min}, width={width}, height={height}")
            print(f"中心点: ({x_center}, {y_center})")
            return (x_center, y_center)
    return None


def check_can_open(d):
    open_btn = d(className="android.widget.Button", textMatches=r"打开|允许|始终允许")
    if open_btn.exists:
        open_btn.click()
        time.sleep(2)


# ocr = PaddleOCR(use_angle_cls=True, lang='ch', show_log=True,  # 显示详细日志，看卡在哪一步
#     use_space_char=False,  # 减少不必要的计算
#     det_db_thresh=0.3,  # 降低检测阈值，加快速度
#     det_db_box_thresh=0.5)
#
#
# def paddle_ocr(image):
#     if isinstance(image, Image.Image):
#         image = np.array(image)
#     result = ocr.ocr(image)
#     texts = []
#     for line in result[0]:  # result 是列表，result[0] 是当前图片的行信息
#         text = line[1][0]  # line[1][0] 是识别的文字，line[1][1] 是置信度
#         texts.append(text)
#     # 拼接方式：可以直接连在一起，或者加空格/换行，根据你的图片实际情况调整
#     full_sentence = ''.join(texts)  # 无空格直接拼接（适合连续文字）
#     print(f"提取的完整文字：{full_sentence}")
#     return full_sentence

 # ch_sim: 简体中文


def easy_ocr(image):
    if isinstance(image, Image.Image):
        image = np.array(image)
    result = easyocr_reader.readtext(image)
    text = ' '.join([res[1] for res in result])  # 直接拼接文字
    return text


# 判断一个字符是否为中文字符
def is_chinese(char):
    return '\u4e00' <= char <= '\u9fff'


def majority_chinese(text):
    if not text:
        return False
    chinese_count = sum(1 for char in text if is_chinese(char))
    return chinese_count > len(text) / 2


search_keys = ["华硕a豆air", "机械革命星耀14", "ipadmini7", "iphone16", "红米note13", "macbookairm4", "华硕灵耀14", "微星星影15"]


def task_loop(d, back_func, origin_app=TB_APP, is_fish=False, duration=22):
    check_can_open(d)
    history_lst = d.xpath(
        '(//android.widget.TextView[@text="历史搜索"]/following-sibling::android.widget.ListView)/android.view.View[1]')
    if history_lst.exists:
        print("查找到搜索关键字", history_lst)
        history_lst.click()
        time.sleep(2)
    else:
        search_view = d(className="android.view.View", text="搜索有福利")
        if search_view.exists:
            search_edit = d.xpath("//android.widget.EditText")
            if search_edit.exists:
                search_edit.set_text(random.choice(search_keys))
                search_btn = d(className="android.widget.Button", text="搜索")
                if search_btn.exists:
                    search_btn.click()
                    time.sleep(2)
    screen_width, screen_height = d.window_size()
    package_name, _ = get_current_app(d)
    # check_count = 3
    # while check_count >= 0:
    #     if not func():
    #         break
    #     print(f"检查次数：{check_count}当前在任务页面，没有执行任务。。。")
    #     check_count -= 1
    #     if check_count <= 0:
    #         return
    #     time.sleep(2)
    start_time = time.time()
    print("开始做任务。。。")
    while True:
        try:
            bt_open = d(resourceId="android:id/button1", text="浏览器打开")
            if bt_open.exists:
                bt_close = d(resourceId="android:id/button2", text="取消")
                if bt_close.exists:
                    bt_close.click()
                    time.sleep(2)
                    break
            if time.time() - start_time > duration:
                break
            if is_fish:
                print("开始查找闲鱼商品")
                time.sleep(4)
                commodity_view1 = d.xpath("//android.widget.ListView/android.view.View[1]")
                if commodity_view1.exists:
                    print(f"存在commodity_view1，点击{commodity_view1.center()}")
                    commodity_view1.click()
                    time.sleep(18)
                    break
                commodity_view2 = d(className="android.view.View", resourceId="feedsContainer")
                if commodity_view2.exists:
                    print(f"存在commodity_view2，点击{(100, commodity_view2.center()[1])}")
                    d.click(300, commodity_view2.center()[1])
                    time.sleep(18)
                    break
            if package_name == origin_app or package_name == TMALL_APP:
                if package_name == ALIPAY_APP:
                    screen_image = d.screenshot(format='opencv')
                    pt1, _, _ = find_button_multiscale(screen_image, "./img/alipay_get.png")
                    if pt1:
                        print("检测到立即领取的弹框，点击立即领取")
                        d.click(int(pt1[0]) + 50, int(pt1[1]) + 20)
                        time.sleep(1)
                start_x = random.randint(screen_width // 6, screen_width // 2)
                start_y = random.randint(screen_height // 2, screen_height - screen_height // 4)
                end_x = random.randint(start_x - 100, start_x)
                end_y = random.randint(200, start_y - 300)
                swipe_time = random.uniform(0.4, 1) if end_y - start_y > 500 else random.uniform(0.2, 0.5)
                print("模拟滑动", start_x, start_y, end_x, end_y, swipe_time)
                d.swipe(start_x, start_y, end_x, end_y, swipe_time)
                time.sleep(random.uniform(0.8, 2))
            else:
                time.sleep(5)
        except Exception as e:
            time.sleep(5)
    back_func()


def back_to_video(d):
    while True:
        temp_package, temp_activity = get_current_app(d)
        if temp_package is None or temp_activity is None or "Ext2ContainerActivity" in temp_activity:
            continue
        if FISH_APP not in temp_package:
            start_app(d, FISH_APP)
            continue




def video_task(d, back_func):
    video_type = None
    screen_width, screen_height = d.window_size()
    while True:
        count_view1 = d(className="android.widget.TextView", textMatches=r"再逛\d+秒后可领奖")
        if count_view1.exists:
            print("属于右上角看广告类型")
            video_type = 1
        else:
            if video_type == 1:
                print("任务完成，返回任务页面")
                back_func()
                break
        if video_type == 1:
            d.swipe(301, screen_height - 500, 322, 500, 0.3)
            time.sleep(2)
        elif video_type == 2:
            speed_btn = d(className="android.widget.TextView", text="我要加速")
            if speed_btn.exists:
                print("点击我要加速")
                speed_btn.click()
                time.sleep(18)
            has_btn = d(className="android.widget.TextView", text="奖励已领取")
            if has_btn.exists:
                print("点击奖励已领取，退出循环")
                has_btn.click()
                time.sleep(2)
                break
        screen_shot = d.screenshot(format='opencv')
        find_button_multiscale(screen_shot, "./img/alipay_get.png")





def close_xy_dialog(d):
    dialog_view1 = d.xpath(
        '//android.webkit.WebView[@text="闲鱼币首页"]/android.view.View/android.view.View[2]//android.widget.Image[1]')
    if dialog_view1.exists:
        dialog_view1.click()
        time.sleep(2)


def get_connected_devices():
    """通过ADB获取所有连接的安卓设备序列号"""
    try:
        # 执行adb命令获取设备列表
        result = subprocess.run(
            ["adb", "devices"],
            capture_output=True,
            text=True,
            check=True
        )

        # 解析输出，提取设备序列号
        output = result.stdout
        devices = []
        for line in output.splitlines():
            # 跳过标题行和空行
            if line.strip() == "" or line.startswith("List of devices attached"):
                continue
            match = re.match(r"^([^\s]+)\s+device$", line)
            if match:
                devices.append(match.group(1))
        return devices
    except subprocess.CalledProcessError:
        print("执行ADB命令失败，请确保ADB已正确安装并添加到环境变量")
        return []
    except FileNotFoundError:
        print("未找到ADB命令，请确保ADB已正确安装并添加到环境变量")
        return []


# 从已连接的设备中，返回用户选中的设备序列号
def select_device():
    # 获取所有连接的设备
    devices = get_connected_devices()

    if not devices:
        raise Exception("未检测到任何连接的安卓设备")

        # 根据设备数量进行处理
    if len(devices) == 1:
        # 只有一个设备，直接返回
        return devices[0]
    else:
        # 多个设备，让用户选择
        print("当前连接多个设备，请输入要执行的设备序号：")
        for i, device in enumerate(devices, 1):
            print(f"  {i}: {device}")

        # 获取用户输入并验证
        while True:
            try:
                choice = input("请输入设备序号：")
                index = int(choice) - 1  # 转换为列表索引

                if 0 <= index < len(devices):
                    # 选中的设备
                    return devices[index]
                else:
                    print(f"输入错误，请重新输入序号（1-{len(devices)}）")
            except ValueError:
                print(f"输入错误，请重新输入序号（1-{len(devices)}）")


def start_app(d, package_name, init=False):
    """根据包名启动应用，支持特定应用的activity配置
    init参数控制启动模式：
    - True: 初始化启动，使用stop=True, use_monkey=True
    - False: 普通启动，使用stop=False, use_monkey=False
    默认不使用activity启动，如果失败再尝试使用activity
    activity启动时不使用use_monkey参数
    启动后验证是否成功"""
    # 根据init参数设置stop和use_monkey
    stop = init
    use_monkey = init
    
    # 获取配置的activity
    activity = APP_START_CONFIG.get(package_name)
    try_count = 3
    try:
        # 优先不使用activity启动
        while try_count > 0:
            print(f"启动应用: {package_name}, stop: {stop}, use_monkey: {use_monkey}, 不使用activity")
            d.app_start(package_name, stop=stop, use_monkey=use_monkey)
            time.sleep(5 if stop else 2)
            # 验证应用是否启动成功
            current_package, _ = get_current_app(d)
            if current_package == package_name:
                print(f"应用 {package_name} 启动成功")
                return
            else:
                print(f"应用 {package_name} 未成功启动，当前应用: {current_package}，尝试后退")
                d.press("back")
                time.sleep(1)
            try_count -= 1
    except Exception as e:
        print(f"不使用activity启动失败: {e}")
        
    # 如果失败且有配置activity，则尝试使用activity启动
    if activity:
        try:
            print(f"使用activity启动应用: {package_name}, activity: {activity}, stop: {stop}")
            d.app_start(package_name=package_name, activity=activity, stop=stop)
            time.sleep(2)
            # 验证应用是否启动成功
            current_package, _ = get_current_app(d)
            if current_package == package_name:
                print(f"应用 {package_name} 启动成功")
                return
            else:
                print(f"应用 {package_name} 未成功启动，当前应用: {current_package}")
        except Exception as e:
            print(f"使用activity启动也失败: {e}")


def check_verify(d):
    verify_view = d(className="android.webkit.WebView", text="验证码拦截")
    if verify_view.exists:
        while True:
            print("存在验证码的情况")
            d.shell("input swipe 150 1700 1180 1700 500")
            time.sleep(3)
            verify_view = d(className="android.webkit.WebView", text="验证码拦截")
            if verify_view.exists:
                d.click(500, 1700)
                time.sleep(3)
            else:
                print("验证码滑动成功")
                break


# find_button2(cv2.imread("screenshot.png"), "./img/alipay_get.png")