137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
# coding=utf-8
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import time
|
|
import uiautomator2 as u2
|
|
import uuid
|
|
from Apps.AiTeJiYiChong.Kit import click_image_template, take_screenshot, setup_logger
|
|
from Apps.AiTeJiYiChong.ReadImageKit import ReadImageKit
|
|
from Config.Config import TEMP_IMAGE_DIR
|
|
|
|
logger = setup_logger("OpenAiTeJiYiChong")
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
async def check_and_close_ad(d):
|
|
"""
|
|
检测并关闭广告弹窗（循环检测直至无广告）
|
|
"""
|
|
max_loops = 3 # 最大检测 3 轮，防止死循环
|
|
device_info = d.info
|
|
w, h = d.window_size()
|
|
device_info['displayWidth'] = w
|
|
device_info['displayHeight'] = h
|
|
|
|
for loop_idx in range(max_loops):
|
|
logger.info(f"开始第 {loop_idx + 1} 轮广告检测...")
|
|
|
|
image_uuid = f"ad_check_{int(time.time())}"
|
|
screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR)
|
|
|
|
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", "unknown")
|
|
logger.info(f"检测到广告 [{ad_type}]，坐标: ({x}, {y})，执行关闭操作...")
|
|
|
|
if ad_type == "rabbit":
|
|
# 针对“兔子”广告的特殊处理
|
|
d.double_click(x, y, duration=0.05)
|
|
else:
|
|
d.click(x, y)
|
|
|
|
await asyncio.sleep(2.0)
|
|
|
|
# 清理截图并继续下一轮检测
|
|
if os.path.exists(screenshot_path): os.remove(screenshot_path)
|
|
else:
|
|
logger.info("未检测到广告弹窗。")
|
|
if os.path.exists(screenshot_path): os.remove(screenshot_path)
|
|
break
|
|
return True
|
|
|
|
async def open_mini_program():
|
|
"""
|
|
异步形式的进入微信小程序: 艾特吉易充
|
|
"""
|
|
d = u2.connect()
|
|
logger.info("执行进入小程序: 艾特吉易充")
|
|
|
|
# 1. 启动微信
|
|
logger.info("启动微信...")
|
|
d.app_start("com.tencent.mm", stop=True)
|
|
await asyncio.sleep(5)
|
|
|
|
# 2. 确保在“微信”消息列表标签页
|
|
tab_chat = d(text="微信", resourceId="com.tencent.mm:id/f2s")
|
|
if not tab_chat.exists:
|
|
tab_chat = d(text="微信")
|
|
|
|
if tab_chat.exists:
|
|
logger.info("点击底部‘微信’标签，确保在消息列表页.")
|
|
tab_chat.click()
|
|
await asyncio.sleep(1)
|
|
|
|
# 3. 点击搜索按钮（放大镜图标）
|
|
logger.info("直接使用百分比坐标点击 '搜索按钮' (84%, 8%)...")
|
|
w, h = d.window_size()
|
|
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)
|
|
|
|
logger.info("输入搜索内容: 艾特吉易充")
|
|
try:
|
|
d.set_input_ime(True)
|
|
except Exception as e:
|
|
logger.warning(f"启用 FastInputIME 失败: {e}")
|
|
|
|
try:
|
|
search_bar_x, search_bar_y = int(w * 0.4), int(h * 0.08)
|
|
d.click(search_bar_x, search_bar_y)
|
|
await asyncio.sleep(1.5)
|
|
except Exception as e:
|
|
logger.warning(f"点击搜索框以获取焦点失败: {e}")
|
|
|
|
try:
|
|
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:
|
|
d.set_input_ime(False)
|
|
except Exception as e:
|
|
logger.warning(f"恢复系统输入法失败: {e}")
|
|
|
|
# 5. 点击小程序
|
|
logger.info("点击搜索结果中的小程序...")
|
|
# 这里由于没有模板，先使用坐标点击作为第一版的测试逻辑 (通常第一个结果在 50%, 18%)
|
|
# 后续有了截图后再补充模板匹配
|
|
w, h = d.window_size()
|
|
d.click(int(w * 0.5), int(h * 0.18))
|
|
logger.info("已点击搜索结果第一项")
|
|
|
|
await asyncio.sleep(8)
|
|
|
|
# 6. 进入后的广告检测 (只在进入时执行一次)
|
|
await check_and_close_ad(d)
|
|
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(open_mini_program())
|