adb自动化控制手机工具类

阳光穿透心脏的1/2处 2022-10-12 00:50 213阅读 0赞

使用：继承该工具类，实现business抽象方法，然后调用start方法即可。

import os
    import subprocess
    from tomorrow import threads
    import time
    import abc
    
    
    class PhoneAutoOption(metaclass=abc.ABCMeta):
        def __init__(self):
            self.device_list = self._device_list()
            self.device_caps = self._get_desired_caps()
    
        def _device_list(self):
            devices: list = []
            result = subprocess.getstatusoutput('adb devices')
            if not result[0]:
                # 按换行符切割
                split = result[1].split('\n')[1:-1]
                for d in split:
                    # 去除tab符
                    device = d.split('\t')[0]
                    devices.append(device)
            return devices
    
        # 获取系统名称
        def _platform_name(self):
            return 'Android'
    
        def _automation_name(self):
            return 'Appium'
    
        # 获取系统版本号
        def _platform_version(self, device: str):
            result = subprocess.getstatusoutput('adb -s {} shell getprop ro.build.version.release'.format(device))
            if not result[0]:
                return result[1]
    
        # 获取设备的appPackage和appActivity
        def _app_package_app_activity(self, device: str):
            result = subprocess.getstatusoutput(
                'adb -s {} shell dumpsys window | findstr mCurrentFocus'.format(device))
            if not result[0]:
                split = result[1].split('Window')[1]
                split_split = split[1:-1].split('u0')[1]
                if '/' not in split_split:
                    raise Exception('手机处于锁屏态：', device)
                package_activity = split_split.split('/')
                return {"appPackage": package_activity[0].strip(), "appActivity": package_activity[1].strip()}
    
        # 启动appium服务
        @threads(10)
        def _start_appium_server(self, port: int, udid: str):
            a = os.popen('netstat -ano | findstr "{}" '.format(port))
            time.sleep(5)
            t1 = a.read()
            if "LISTENING" in t1:
                print('***',t1)
                self._kill_process(port)
                print("appium服务已经启动：{}".format(t1))
            # else:
            # 启动appium服务
            os.system("appium -a 127.0.0.1 -p {} -U {} --no-reset --session-override".format(port, udid))
            pass
    
        # 杀掉进程
        def _kill_process(self, port):
            ip = os.popen("ifconfig eth0|grep 'inet addr'|awk -F ':' '{print $2}'|awk '{print $1}'")
            ip = ip.read().strip()
            pid = os.popen("netstat -anp|grep {} |awk '{print $7}'".format(port)).read().split('/')[0]
            print("pid:",pid)
            os.popen('kill -9 {0}'.format(int(pid)))
    
        # 获取设备desired_caps列表
        def _get_desired_caps(self):
            desired_caps: list = []
            device_list = self.device_list
            for device in device_list:
                app_package_app_activity = self._app_package_app_activity(device)
                desired_caps.append({
                    "platformName": self._platform_name(),
                    "platformVersion": self._platform_version(device),
                    "deviceName": device,
                    "appPackage": app_package_app_activity['appPackage'],
                    "appActivity": app_package_app_activity['appActivity'],
                    "automationName": self._automation_name(),
                    "autoAcceptAlerts": True,
                    "noReset": True,
                    "unicodeKeyboard": True,
                    "resetKeyboard": True,
                    "udid": device
                })
            return desired_caps
    
        from appium import webdriver
    
        @threads(2)
        def _async_run_app(self, desired_cap, port):
            self._start_appium_server(port=port, udid=desired_cap['udid'])
            time.sleep(10)
            driver = self.webdriver.Remote('http://127.0.0.1:{}/wd/hub'.format(port), desired_cap)
            time.sleep(10)
            self._business(driver)
    
        # 手机自动化业务的抽象方法，留给使用者进行继承实现。
        @abc.abstractmethod
        def _business(self, driver):
            pass
    
        # 开始键
        def start(self):
            start_id = 4700
            caps = self._get_desired_caps()
            for cap in caps:
                self._async_run_app(cap, start_id)
                start_id = start_id + 1

示例：点击手机主页的微信

from auto_option import PhoneAutoOption
    
    
    class WxAutoOption(PhoneAutoOption):
    
        def _business(self, driver):
            driver.find_element_by_xpath('//android.widget.ImageView[@content-desc="微信"]').click()
            driver.quit()
    
    
    if __name__ == '__main__':
        option = WxAutoOption()
        option.start()
        pass