Appium自动化常用adb操作封装

import os
import platform
import re
import subprocess
from common import path
from common.exception import AndroidSDKUninstalledError, AndroidDevicesNotFoundError
from common.logger import logger
class ADBRunner:
    def __init__(self):
        self._check_adb_is_installed()
    @staticmethod
    def run_adb(command):
        try:
            result = subprocess.run(command, capture_output=True, text=True, check=True, shell=True)
            logger.debug(f"Execute adb command: {command}")
            return result.stdout.strip()
        except subprocess.CalledProcessError as e:
            logger.error(f"Execute adb command failure: {e}")
            return None
    def _check_adb_is_installed(self):
        result = self.run_adb("adb --version")
        if not result:
            raise AndroidSDKUninstalledError('Android SDK is not installed or configured.')
        return result
    def adb_connect_device(self, device):
        return self.run_adb(f'adb connect {device}')
    def get_connected_device_udids(self):
        """
        获取所有连接设备的序列号udid
        """
        res = self.run_adb('adb devices')
        pattern = r'\b((?!of\b)\S+)\s+device'
        devices = re.findall(pattern, res)
        if not devices:
            raise AndroidDevicesNotFoundError('No connected mobile devices found.')
        logger.info(f'Devices found: {devices}')
        return devices
    @property
    def _grep(self):
        if platform.system() == 'Windows':
            return 'findstr'
        else:
            return 'grep'
    def get_activities(self, udid=None):
        """
        获取当前设备的所有top activities, 输出结果示例：
        ['com.android.settings/.Settings c4e2e17 pid=3637',
        'com.mumu.launcher/.Launcher 9392da7 pid=1434',
        'com.android.browser/com.android.settings de68b73 pid=3722']
        """
        if not udid:
            udid = self.get_connected_device_udids()[0]
        command = f'adb -s {udid} shell dumpsys activity top | {self._grep} ACTIVITY'
        tops = self.run_adb(command).split('ACTIVITY')
        tops = [top.strip(' ').strip('\n') for top in tops if top]
        logger.debug(f'The top activities on device {udid} are: {tops}')
        return tops
    def get_app_package_and_activity(self, udid=None):
        """
        从 com.android.settings/.Settings c4e2e17 pid=3637，获取包名和活动页面名称
        输出：['com.android.settings', '.Settings']
        用途：
            capabilities = {
                "platformName": "Android",
                "automationName": "uiautomator2",
                "deviceName": "9YS0220306003185",
                "appPackage": "com.tencent.mm",  # 包名
                "appActivity": ".ui.LauncherUI",  # 活动页面名称
            }
        """
        last_activity_info = self.get_activities(udid)[-1]
        pattern = r'(\S+) (\S+) pid=(\d+)'
        match = re.match(pattern, last_activity_info)
        if match:
            package, activity = match.group(1).split('/')
            return [package, activity]
        else:
            raise
    def get_apk_path(self, package_name, udid=None):
        """
        从设备中使用包名，获取应用程序的APK路径。
        """
        if not udid:
            udid = self.get_connected_device_udids()[0]
        # 使用pm path命令获取应用程序的APK路径
        command = f'adb -s {udid} shell pm path {package_name}'
        result = self.run_adb(command)
        if result and result.startswith('package:'):
            apk_path = result.replace('package:', '').strip()
            return apk_path
        else:
            logger.error(f"APK path for package '{package_name}' not found on the device {udid}.")
            return None
    def get_apk_version(self, package_name, udid=None):
        """
        获取应用程序的版本号。
        """
        if not udid:
            udid = self.get_connected_device_udids()[0]
        # 使用dumpsys package命令获取应用程序的版本号
        command = f'adb -s {udid} shell dumpsys package {package_name} | {self._grep} versionCode'
        result = self.run_adb(command)
        match = re.search(r'versionCode=(\d+)', result)
        if match:
            version_code = match.group(1)
            return version_code
        else:
            logger.error(f"Version code not found for package '{package_name}'.")
            return None
    def pull_apk_from_device(self, package_name, output_dir=None, apk_name=None, udid=None):
        """
        根据包名从当前设备中将应用程序的APK文件复制至本地。
        用于：
            capabilities = {
                "platformName": "Android",
                "automationName": "uiautomator2",
                "deviceName": "9YS0220306003185",
                "app": apk_path,  # 用在这里
                # "appPackage": "com.tencent.mm",
                # "appActivity": ".ui.LauncherUI",
            }
        """
        if not udid:
            udid = self.get_connected_device_udids()[0]
        if not output_dir:
            output_dir = path.get_apk_resources_dir()
        # 获取应用程序的APK路径
        apk_path = self.get_apk_path(package_name)
        apk_name = apk_name if apk_name else package_name
        apk_version = self.get_apk_version(package_name)
        if apk_path:
            # 构建本地输出路径
            local_output_path = os.path.join(output_dir, f"{apk_name}_{apk_version}.apk")
            # 使用adb pull命令将APK复制到本地
            command = f'adb -s {udid} pull {apk_path} {local_output_path}'
            result = self.run_adb(command)
            if result:
                logger.info(f"APK successfully pulled to: {local_output_path}")
                return local_output_path
            else:
                logger.error("Failed to pull APK from device.")
if __name__ == '__main__':
    adb = ADBRunner()
    print(adb.get_connected_device_udids())
    print(adb.get_app_package_and_activity())
    print(adb.get_apk_path('cn.com.open.mooc'))
    print(adb.get_apk_version('cn.com.open.mooc'))
    print(adb.pull_apk_from_device('cn.com.open.mooc'))