211 lines
7.9 KiB
Python
211 lines
7.9 KiB
Python
# coding=utf-8
|
||
import asyncio
|
||
import logging
|
||
import os
|
||
import time
|
||
import uuid
|
||
|
||
import uiautomator2 as u2
|
||
|
||
from Apps.XinDianTu.Kit import take_screenshot
|
||
from Apps.XinDianTu.ReadImageKit import ReadImageKit
|
||
from Config.Config import TEMP_IMAGE_DIR
|
||
|
||
# pip install adbutils
|
||
# 配置日志输出，方便调试和监控
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger("OpenXinDianTu")
|
||
|
||
# 获取当前脚本所在目录
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
async def check_and_close_ad(d):
|
||
"""
|
||
检测并关闭广告弹窗（仅使用 VL 视觉模型）
|
||
"""
|
||
logger.info("开始检测广告弹窗 (VL方案)...")
|
||
|
||
# [已移除] 策略：先向上滑动一点，触发悬浮广告（如兔子）收起/躲避
|
||
# 用户反馈：一滚动，兔子就藏起来了，导致无法检测到广告，因此取消此逻辑。
|
||
# try:
|
||
# w, h = d.window_size()
|
||
# logger.info(f"执行微小滑动 (Swipe Up)，尝试触发悬浮广告收起...")
|
||
# # 从 70% 处滑到 50% 处，模拟手指上滑，页面内容上移
|
||
# d.swipe(w * 0.5, h * 0.7, w * 0.5, h * 0.5, duration=0.3)
|
||
# # 等待滑动动画结束和广告收起动画
|
||
# await asyncio.sleep(0.5)
|
||
# except Exception as e:
|
||
# logger.warning(f"滑动操作异常: {e}")
|
||
|
||
# 1. 拍摄截图
|
||
t1 = time.time()
|
||
image_uuid = str(uuid.uuid4())
|
||
# 使用相对路径: 基于当前脚本目录下的 Images 文件夹
|
||
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")
|
||
|
||
logger.info(f"通过视觉大模型检测到广告关闭按钮: ({x}, {y}) [Type: {ad_type}]")
|
||
|
||
# 针对“新电兔AI”兔子广告的强力关闭策略
|
||
if ad_type == "rabbit":
|
||
logger.info(">>> 检测到顽固的兔子广告，执行强力双击策略...")
|
||
d.double_click(x, y, duration=0.05)
|
||
|
||
# 处理“误触背景卡片”的逻辑
|
||
logger.info(">>> 等待 3s 以确认是否进入了二级页面...")
|
||
await asyncio.sleep(3.0)
|
||
|
||
# 检查是否还在列表页
|
||
is_list_page = d(text="全城").exists or d(text="综合排序").exists or d(text="搜索场站").exists
|
||
if not is_list_page:
|
||
logger.info(">>> 检测到当前不在列表页，尝试执行返回操作...")
|
||
d.press("back")
|
||
await asyncio.sleep(1.5)
|
||
else:
|
||
logger.info(">>> 当前仍在列表页，无需返回。")
|
||
else:
|
||
d.click(x, y)
|
||
|
||
if os.path.exists(screenshot_path):
|
||
os.remove(screenshot_path)
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"视觉大模型广告检测异常: {e}")
|
||
|
||
logger.info(f"VL 检测完成，未发现已知广告。")
|
||
|
||
# 如果未检测到广告，保留截图以便调试（保存为 debug_ad_failure.jpg）
|
||
debug_path = os.path.join(os.path.dirname(screenshot_path), "debug_ad_failure.jpg")
|
||
if os.path.exists(debug_path):
|
||
os.remove(debug_path)
|
||
if os.path.exists(screenshot_path):
|
||
os.rename(screenshot_path, debug_path)
|
||
logger.info(f"未检测到广告，截图已保存至: {debug_path} (请检查该图片是否真的包含广告)")
|
||
|
||
return False
|
||
|
||
|
||
|
||
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("准备输入搜索内容: 新电途")
|
||
|
||
# 启用 u2 快速输入法 (使用更新的方法名)
|
||
# d.set_input_ime(True)
|
||
|
||
# 尝试点击搜索框位置以确保获取焦点 (通常在顶部居中偏左)
|
||
# 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) # 稍微多等一会儿让输入法准备好
|
||
|
||
# 直接发送按键
|
||
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}")
|
||
# 最后的尝试：切换回默认输入法（有时 FastInputIME 会卡住）
|
||
# d.set_fastinput_ime(False)
|
||
# d.send_keys("新电途")
|
||
|
||
await asyncio.sleep(3)
|
||
|
||
# # 兜底点击：点击搜索结果的第一项位置 (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}")
|