aiData/Apps/XinDianTu/Opener.py

# coding=utf-8
import asyncio
import os
import time
import uuid

import uiautomator2 as u2

from Apps.XinDianTu.Kit import take_screenshot, setup_logger
from Apps.XinDianTu.ReadImageKit import ReadImageKit
from Config.Config import TEMP_IMAGE_DIR

# pip install adbutils
logger = setup_logger("OpenXinDianTu")

# 获取当前脚本所在目录
BASE_DIR = os.path.dirname(os.path.abspath(__file__))

async def check_and_close_ad(d):
    """
    检测并关闭广告弹窗（循环检测直至无广告）
    """
    max_loops = 3  # 最大检测 3 轮，防止死循环
    for loop_idx in range(max_loops):
        logger.info(f"开始第 {loop_idx + 1} 轮广告检测...")
        
        # 1. 拍摄截图
        t1 = time.time()
        image_uuid = str(uuid.uuid4())
        save_dir = TEMP_IMAGE_DIR
        screenshot_path = take_screenshot(d, image_uuid, save_dir=save_dir)
        logger.info(f"Step [广告检测截图] 耗时: {time.time() - t1:.4f}s")

        # 2. 视觉大模型检测方案 (VL Model)
        try:
            window_size = d.window_size()
            device_info = {
                "width": window_size[0],
                "height": window_size[1],
                "productName": d.info.get('productName', 'unknown')
            }
            
            # 使用最新的 detect_ad_popup 方法
            ad_result = await ReadImageKit.detect_ad_popup(screenshot_path, device_info=device_info)
            if ad_result:
                x, y = ad_result["x"], ad_result["y"]
                ad_type = ad_result.get("ad_type")

                # 根据新的策略：完全静默忽略 rabbit 类型广告
                if ad_type == "rabbit":
                    if os.path.exists(screenshot_path):
                        os.remove(screenshot_path)
                    return True
                else:
                    logger.info(f"检测到广告关闭按钮: ({x}, {y}) [Type: {ad_type}]")
                    logger.info(f">>> 正在点击关闭非 rabbit 广告 ({ad_type})...")
                    d.click(x, y)
                    await asyncio.sleep(2.0) # 等待关闭动画
                    
                    if os.path.exists(screenshot_path):
                        os.remove(screenshot_path)
                    # 继续下一轮循环，检查是否还有其他广告
                    continue
            else:
                logger.info("本轮未检测到广告。")
                if os.path.exists(screenshot_path):
                    os.remove(screenshot_path)
                return True
        except Exception as e:
            logger.error(f"视觉大模型广告检测异常: {e}")
            if os.path.exists(screenshot_path): os.remove(screenshot_path)
            break

    return True



async def open_mini_program():
    """
    异步形式的进入微信小程序
    """
    # 连接设备
    d = u2.connect()

    logger.info("执行进入小程序.")

    # 启动微信 (强制停止后再启动，确保回到主页)
    logger.info("启动微信...")
    d.app_start("com.tencent.mm", stop=True)
    await asyncio.sleep(5)  # 给微信充足的启动时间

    # 0. 确保在“微信”消息列表标签页
    tab_chat = d(text="微信", resourceId="com.tencent.mm:id/f2s")  # 常见底部标签 ID
    if not tab_chat.exists:
        tab_chat = d(text="微信")

    if tab_chat.exists:
        # 如果没选中（通常选中的文本颜色不同或有其他特征，这里直接点一下确保切换）
        logger.info("点击底部‘微信’标签，确保在消息列表页.")
        tab_chat.click()
        await asyncio.sleep(1)

    # 1. 直接使用坐标点击搜索按钮（放大镜图标）
    logger.info("直接使用百分比坐标点击 '搜索按钮' (84%, 8%)...")
    window_size = d.window_size()
    w, h = window_size[0], window_size[1]
    click_x = int(w * 0.84)
    click_y = int(h * 0.08)
    d.click(click_x, click_y)
    logger.info(f"使用精确坐标点击搜索按钮: ({click_x}, {click_y})")

    # 点击后给予一定的加载时间
    await asyncio.sleep(2)

    # 2. 确保输入法和焦点正常，然后输入搜索内容
    logger.info("准备输入搜索内容: 新电途")

    # 尝试启用 FastInputIME，提升 send_keys 的成功率
    try:
        logger.info("尝试启用 FastInputIME 输入法...")
        d.set_input_ime(True)
        logger.info("FastInputIME 已启用.")
    except Exception as e:
        logger.warning(f"启用 FastInputIME 失败: {e}")

    # 再点击一次搜索框，确保真正拿到输入焦点
    try:
        search_bar_x, search_bar_y = int(w * 0.4), int(h * 0.08)
        logger.info(f"再次点击搜索框以获取焦点: ({search_bar_x}, {search_bar_y})")
        d.click(search_bar_x, search_bar_y)
        await asyncio.sleep(1.5)
    except Exception as e:
        logger.warning(f"点击搜索框以获取焦点失败: {e}")

    # 直接发送按键
    try:
        # 尝试先不使用 clear=True，因为它在某些设备上会导致 ADB_KEYBOARD_CLEAR_TEXT 错误
        # 如果需要清除文本，可以尝试全选+删除，或者假设搜索框是空的
        logger.info("尝试输入文字: 新电途")
        d.send_keys("新电途")
        logger.info("文字输入指令已发送.")
    except Exception as e:
        logger.warning(f"直接 send_keys 失败: {e}")
        
        try:
            # 备选方案：尝试获取焦点元素并设置文本
            logger.info("尝试使用 set_text 设置文本...")
            d(focused=True).set_text("新电途")
            logger.info("set_text 指令已发送.")
        except Exception as e2:
            logger.error(f"文字输入彻底失败: {e2}")

    await asyncio.sleep(3)

    # 尝试恢复系统默认输入法，避免后续人工操作受影响
    try:
        logger.info("尝试恢复系统默认输入法...")
        d.set_input_ime(False)
        logger.info("已恢复系统默认输入法.")
    except Exception as e:
        logger.warning(f"恢复系统默认输入法失败: {e}")

    # # 兜底点击：点击搜索结果的第一项位置 (x=50%, y=18%)
    # res_x, res_y = int(w * 0.5), int(h * 0.18)
    # logger.info(f"尝试坐标点击第一项: ({res_x}, {res_y})")
    # d.click(res_x, res_y)

    # 3. 直接使用坐标点击搜索结果第一项 (50%, 18%)
    logger.info("直接使用百分比坐标点击 '搜索结果第一项' (50%, 18%)...")
    res_x, res_y = int(w * 0.5), int(h * 0.18)
    d.click(res_x, res_y)
    logger.info(f"坐标点击搜索结果第一项: ({res_x}, {res_y})")

    await asyncio.sleep(10)
    # 进入小程序后，检查并处理广告
    await check_and_close_ad(d)
    return True


async def main():
    success = await open_mini_program()
    if success:
        logger.info("任务执行成功.")
    else:
        logger.error("任务执行失败.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("程序被用户中断.")
    except Exception as e:
        logger.exception(f"运行过程中出现异常: {e}")