Airtest框架详解：游戏/Android App 图像+UI树双引擎自动化#

# 必须的三个依赖
# airtest：图像识别、设备交互核心
# opencv-contrib-python：图像识别的SIFT/SURF稳定引擎（airtest自带简化版，偶尔不稳定）
# pocoui：跨平台全量UI树驱动，含Android/iOS/Unity/Cocos等
pip install airtest opencv-contrib-python pocoui -i https://pypi.tuna.tsinghua.edu.cn/simple

from airtest.core.api import *
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
import time

class AirtestController:
    """Airtest+Poco 双引擎基础控制器"""
    
    def __init__(self, device_id=None):
        self.device_id = device_id
        self.poco = None
        self._setup_device()
    
    def _setup_device(self):
        """连接Android设备 + 初始化Poco（关闭每步截图提速80%+）"""
        try:
            # 单设备：直接连adb扫描到的第一台
            # 多设备：指定device_id（通过adb devices获取）
            device_uri = f"Android:///{self.device_id}" if self.device_id else "Android:///"
            connect_device(device_uri)
            
            # 初始化Poco
            self.poco = AndroidUiautomationPoco(
                use_airtest_input=True,  # 用Airtest输入适配混合/游戏
                screenshot_each_action=False  # 必须关！大幅提升运行速度
            )
            
            print(f"✅ 设备连接成功：{device().get_serial()}")
            
        except Exception as e:
            print(f"❌ 设备连接失败：{str(e)}")
            raise
    
    # ------------ 图像识别封装 ------------
    def find_and_click(self, img_path, threshold=0.8, timeout=10):
        """等待指定阈值的图像出现并点击"""
        try:
            pos = wait(Template(img_path, threshold=threshold), timeout=timeout)
            touch(pos)
            print(f"✅ 点击图像：{img_path}")
            return True
        except Exception as e:
            print(f"❌ 图像未找到/点击失败：{img_path}")
            return False
    
    # ------------ UI树操作封装 ------------
    def wait_for_text(self, text, timeout=10):
        """等待指定文本的UI元素出现"""
        try:
            self.poco(text=text).wait_for_appearance(timeout=timeout)
            print(f"✅ 文本出现：{text}")
            return True
        except:
            print(f"❌ 等待文本超时：{text}")
            return False
    
    # ------------ 通用设备操作 ------------
    def swipe_up(self, duration=0.5, offset=100):
        """从屏幕下往上滑动（适配主流信息流App/游戏）"""
        w, h = device().get_current_resolution()
        start = (w//2, h*4//5 + random.randint(-offset, offset))
        end = (w//2, h//5 + random.randint(-offset, offset))
        swipe(start, end, duration=duration)
        print("📱 向上随机滑动一次")

from airtest.core.api import *
import time
import random

class DouyinAuto(AirtestController):
    """抖音自动化类，继承基础双引擎控制器"""
    
    # 抖音不同渠道包名可能不同：通过adb shell pm list packages -3查找
    DOUYIN_PKG = "com.ss.android.ugc.aweme"
    
    def launch(self):
        """启动抖音并等待首页「推荐」标签加载完成"""
        try:
            start_app(self.DOUYIN_PKG)
            # 首页加载等待时间稍长，适配网络波动
            self.wait_for_text("推荐", timeout=20)
            print("📱 抖音首页加载完成")
            return True
        except Exception as e:
            print(f"❌ 抖音启动失败：{str(e)}")
            return False
    
    def random_like(self):
        """70%概率点赞：优先图像识别，通用坐标兜底"""
        if random.random() > 0.7:
            return
        
        try:
            # 优先识别未点赞的爱心图标
            if exists(Template("like_btn.png", threshold=0.7)):
                touch(Template("like_btn.png"))
            else:
                # 兜底：主流手机抖音右侧爱心的通用坐标
                w, h = device().get_current_resolution()
                touch((w*4//5, h//2 + 100 + random.randint(-20, 20)))
            print("❤️  随机点赞成功")
            time.sleep(random.uniform(0.3, 0.7))
        except:
            pass
    
    def random_comment(self, comments=None):
        """20%概率评论：从预设/用户自定义列表随机选内容"""
        if random.random() > 0.2:
            return
        
        default_comments = ["不错👍", "学到了", "666", "收藏了", "好看！", "太真实了"]
        comment = random.choice(comments or default_comments)
        w, h = device().get_current_resolution()
        
        try:
            # 点击右侧评论图标（通用坐标带随机偏移）
            touch((w*4//5, h//2 + 200 + random.randint(-20, 20)))
            time.sleep(random.uniform(1.5, 2.5))
            # 点击底部输入框
            touch((w//2 + random.randint(-50, 50), h*5//6))
            time.sleep(random.uniform(0.8, 1.2))
            # 输入评论
            text(comment)
            time.sleep(random.uniform(0.3, 0.7))
            # 点击右侧发送按钮
            touch((w*9//10 + random.randint(-20, 20), h*5//6))
            print(f"💬 随机评论成功：{comment}")
            time.sleep(random.uniform(1.5, 2.5))
            # 返回视频页
            keyevent("BACK")
            time.sleep(random.uniform(0.8, 1.2))
        except:
            # 出错直接返回，避免脚本卡住
            keyevent("BACK")
            time.sleep(random.uniform(0.5, 1))
    
    def run(self, cycles=5):
        """运行指定次数的交互循环"""
        if not self.launch():
            return
        
        print(f"\n🤖 开始执行 {cycles} 次抖音交互循环...\n")
        
        for i in range(cycles):
            print(f"🔄 第 {i+1} 次循环：")
            # 随机点赞
            self.random_like()
            # 随机评论
            self.random_comment()
            # 刷下一个视频
            self.swipe_up()
            # 随机等待3-6秒，模拟真人浏览速度（避风控核心！）
            wait_time = random.uniform(3, 6)
            print(f"⏳ 等待 {wait_time:.1f} 秒\n")
            time.sleep(wait_time)
        
        print("🏁 所有循环执行完毕！")
        stop_app(self.DOUYIN_PKG)

# 生成可视化HTML测试报告
from airtest.report.report import simple_report
simple_report(__file__, logpath=True, output="douyin_auto_report.html")

# 测试脚本入口
if __name__ == "__main__":
    douyin = DouyinAuto()
    douyin.run(cycles=3)