Python常用工具类之adbtool示例代码_python_

#!/usr/bin/evn python # -*- coding:utf-8 -*- # FileName adbtools.py # Author: HeyNiu # Created Time: 2016/9/19 """ adb 工具类 """ import os import platform import re import time #import utils.timetools class AdbTools(object): def __init__(self, device_id=''): self.__system = platform.system() self.__find = '' self.__command = '' self.__device_id = device_id self.__get_find() self.__check_adb() self.__connection_devices() def __get_find(self): """ 判断系统类型，windows使用findstr，linux使用grep :return: """ # if self.__system is "Windows": #self.__find = "findstr" #else: #self.__find = "grep" def __check_adb(self): """ 检查adb 判断是否设置环境变量ANDROID_HOME :return: """ if "ANDROID_HOME" in os.environ: if self.__system == "Windows": path = os.path.join(os.environ["ANDROID_HOME"], "platform-tools", "adb.exe") if os.path.exists(path): self.__command = path else: raise EnvironmentError( "Adb not found in $ANDROID_HOME path: %s." % os.environ["ANDROID_HOME"]) else: path = os.path.join(os.environ["ANDROID_HOME"], "platform-tools", "adb") if os.path.exists(path): self.__command = path else: raise EnvironmentError( "Adb not found in $ANDROID_HOME path: %s." % os.environ["ANDROID_HOME"]) else: raise EnvironmentError( "Adb not found in $ANDROID_HOME path: %s." % os.environ["ANDROID_HOME"]) def __connection_devices(self): """ 连接指定设备，单个设备可不传device_id :return: """ if self.__device_id == "": return self.__device_id = "-s %s" % self.__device_id def adb(self, args): """ 执行adb命令 :param args:参数 :return: """ cmd = "%s %s %s" % (self.__command, self.__device_id, str(args)) # print(cmd) return os.popen(cmd) def shell(self, args): """ 执行adb shell命令 :param args:参数 :return: """ cmd = "%s %s shell %s" % (self.__command, self.__device_id, str(args)) # print(cmd) return os.popen(cmd) def mkdir(self, path): """ 创建目录 :param path: 路径 :return: """ return self.shell('mkdir %s' % path) def get_devices(self): """ 获取设备列表 :return: """ l = self.adb('devices').readlines() return (i.split()[0] for i in l if 'devices' not in i and len(i) > 5) def get_current_application(self): """ 获取当前运行的应用信息 :return: """ return self.shell('dumpsys window w | %s \/ | %s name=' % (self.__find, self.__find)).read() def get_current_package(self): """ 获取当前运行app包名 :return: """ reg = re.compile(r'name=(.+?)/') return re.findall(reg, self.get_current_application())[0] def get_current_activity(self): """ 获取当前运行activity :return: package/activity """ reg = re.compile(r'name=(.+?)\)') return re.findall(reg, self.get_current_application())[0] '''def __get_process(self, package_name): """ 获取进程信息 :param package_name: :return: """ if self.__system is "Windows": pid_command = self.shell("ps | %s %s$" % (self.__find, package_name)).read() else: pid_command = self.shell("ps | %s -w %s" % (self.__find, package_name)).read() return pid_command''' def process_exists(self, package_name): """ 返回进程是否存在 :param package_name: :return: """ process = self.__get_process(package_name) return package_name in process def get_pid(self, package_name): """ 获取pid :return: """ pid_command = self.__get_process(package_name) if pid_command == '': print("The process doesn't exist.") return pid_command req = re.compile(r"\d+") result = str(pid_command).split() result.remove(result[0]) return req.findall(" ".join(result))[0] def get_uid(self, pid): """ 获取uid :param pid: :return: """ result = self.shell("cat /proc/%s/status" % pid).readlines() for i in result: if 'uid' in i.lower(): return i.split()[1] def get_flow_data_tcp(self, uid): """ 获取应用tcp流量 :return:(接收, 发送) """ tcp_rcv = self.shell("cat proc/uid_stat/%s/tcp_rcv" % uid).read().split()[0] tcp_snd = self.shell("cat proc/uid_stat/%s/tcp_snd" % uid).read().split()[0] return tcp_rcv, tcp_snd def get_flow_data_all(self, uid): """ 获取应用流量全部数据 包含该应用多个进程的所有数据 tcp udp等 (rx_bytes, tx_bytes) >> (接收, 发送) :param uid: :return:list(dict) """ all_data = [] d = {} data = self.shell("cat /proc/net/xt_qtaguid/stats | %s %s" % (self.__find, uid)).readlines() for i in data: if not i.startswith('\n'): item = i.strip().split() d['idx'] = item[0] d['iface'] = item[1] d['acct_tag_hex'] = item[2] d['uid_tag_int'] = item[3] d['cnt_set'] = item[4] d['rx_bytes'] = item[5] d['rx_packets'] = item[6] d['tx_bytes'] = item[7] d['tx_packets'] = item[8] d['rx_tcp_bytes'] = item[9] d['rx_tcp_packets'] = item[10] d['rx_udp_bytes'] = item[11] d['rx_udp_packets'] = item[12] d['rx_other_bytes'] = item[13] d['rx_other_packets'] = item[14] d['tx_tcp_bytes'] = item[15] d['tx_tcp_packets'] = item[16] d['tx_udp_bytes'] = item[17] d['tx_udp_packets'] = item[18] d['tx_other_bytes'] = item[19] d['tx_other_packets'] = item[20] all_data.append(d) d = {} return all_data @staticmethod def dump_apk(path): """ dump apk文件 :param path: apk路径 :return: """ # 检查build-tools是否添加到环境变量中 # 需要用到里面的aapt命令 l = os.environ['PATH'].split(';') build_tools = False for i in l: if 'build-tools' in i: build_tools = True if not build_tools: raise EnvironmentError("ANDROID_HOME BUILD-TOOLS COMMAND NOT FOUND.\nPlease set the environment variable.") return os.popen('aapt dump badging %s' % (path,)) @staticmethod def dump_xml(path, filename): """ dump apk xml文件 :return: """ return os.popen('aapt dump xmlstrings %s %s' % (path, filename)) def uiautomator_dump(self): """ 获取屏幕uiautomator xml文件 :return: """ return self.shell('uiautomator dump').read().split()[-1] def pull(self, source, target): """ 从手机端拉取文件到电脑端 :return: """ self.adb('pull %s %s' % (source, target)) def push(self, source, target): """ 从电脑端推送文件到手机端 :param source: :param target: :return: """ self.adb('push %s %s' % (source, target)) def remove(self, path): """ 从手机端删除文件 :return: """ self.shell('rm %s' % (path,)) def clear_app_data(self, package): """ 清理应用数据 :return: """ self.shell('pm clear %s' % (package,)) def install(self, path): """ 安装apk文件 :return: """ # adb install 安装错误常见列表 errors = {'INSTALL_FAILED_ALREADY_EXISTS': '程序已经存在', 'INSTALL_DEVICES_NOT_FOUND': '找不到设备', 'INSTALL_FAILED_DEVICE_OFFLINE': '设备离线', 'INSTALL_FAILED_INVALID_APK': '无效的APK', 'INSTALL_FAILED_INVALID_URI': '无效的链接', 'INSTALL_FAILED_INSUFFICIENT_STORAGE': '没有足够的存储空间', 'INSTALL_FAILED_DUPLICATE_PACKAGE': '已存在同名程序', 'INSTALL_FAILED_NO_SHARED_USER': '要求的共享用户不存在', 'INSTALL_FAILED_UPDATE_INCOMPATIBLE': '版本不能共存', 'INSTALL_FAILED_SHARED_USER_INCOMPATIBLE': '需求的共享用户签名错误', 'INSTALL_FAILED_MISSING_SHARED_LIBRARY': '需求的共享库已丢失', 'INSTALL_FAILED_REPLACE_COULDNT_DELETE': '需求的共享库无效', 'INSTALL_FAILED_DEXOPT': 'dex优化验证失败', 'INSTALL_FAILED_DEVICE_NOSPACE': '手机存储空间不足导致apk拷贝失败', 'INSTALL_FAILED_DEVICE_COPY_FAILED': '文件拷贝失败', 'INSTALL_FAILED_OLDER_SDK': '系统版本过旧', 'INSTALL_FAILED_CONFLICTING_PROVIDER': '存在同名的内容提供者', 'INSTALL_FAILED_NEWER_SDK': '系统版本过新', 'INSTALL_FAILED_TEST_ONLY': '调用者不被允许测试的测试程序', 'INSTALL_FAILED_CPU_ABI_INCOMPATIBLE': '包含的本机代码不兼容', 'CPU_ABIINSTALL_FAILED_MISSING_FEATURE': '使用了一个无效的特性', 'INSTALL_FAILED_CONTAINER_ERROR': 'SD卡访问失败', 'INSTALL_FAILED_INVALID_INSTALL_LOCATION': '无效的安装路径', 'INSTALL_FAILED_MEDIA_UNAVAILABLE': 'SD卡不存在', 'INSTALL_FAILED_INTERNAL_ERROR': '系统问题导致安装失败', 'INSTALL_PARSE_FAILED_NO_CERTIFICATES': '文件未通过认证 >> 设置开启未知来源', 'INSTALL_PARSE_FAILED_INCONSISTENT_CERTIFICATES': '文件认证不一致 >> 先卸载原来的再安装', 'INSTALL_FAILED_INVALID_ZIP_FILE': '非法的zip文件 >> 先卸载原来的再安装', 'INSTALL_CANCELED_BY_USER': '需要用户确认才可进行安装', 'INSTALL_FAILED_VERIFICATION_FAILURE': '验证失败 >> 尝试重启手机', 'DEFAULT': '未知错误' } print('Installing...') l = self.adb('install -r %s' % (path,)).read() if 'Success' in l: print('Install Success') if 'Failure' in l: reg = re.compile('\\[(.+?)\\]') key = re.findall(reg, l)[0] try: print('Install Failure >> %s' % errors[key]) except KeyError: print('Install Failure >> %s' % key) return l def uninstall(self, package): """ 卸载apk :param package: 包名 :return: """ print('Uninstalling...') l = self.adb('uninstall %s' % (package,)).read() print(l) def screenshot(self, target_path=''): """ 手机截图 :param target_path: 目标路径 :return: """ format_time = utils.timetools.timestamp('%Y%m%d%H%M%S') self.shell('screencap -p /sdcard/%s.png' % (format_time,)) time.sleep(1) if target_path == '': self.pull('/sdcard/%s.png' % (format_time,), os.path.expanduser('~')) else: self.pull('/sdcard/%s.png' % (format_time,), target_path) self.remove('/sdcard/%s.png' % (format_time,)) def get_cache_logcat(self): """ 导出缓存日志 :return: """ return self.adb('logcat -v time -d') def get_crash_logcat(self): """ 导出崩溃日志 :return: """ return self.adb('logcat -v time -d | %s AndroidRuntime' % (self.__find,)) def clear_cache_logcat(self): """ 清理缓存区日志 :return: """ self.adb('logcat -c') def get_device_time(self): """ 获取设备时间 :return: """ return self.shell('date').read().strip() def ls(self, command): """ shell ls命令 :return: """ return self.shell('ls %s' % (command,)).readlines() def file_exists(self, target): """ 判断文件在目标路径是否存在 :return: """ l = self.ls(target) for i in l: if i.strip() == target: return True return False def is_install(self, target_app): """ 判断目标app在设备上是否已安装 :param target_app: 目标app包名 :return: bool """ return target_app in self.shell('pm list packages %s' % (target_app,)).read() def get_device_model(self): """ 获取设备型号 :return: """ return self.shell('getprop ro.product.model').read().strip() def get_device_id(self): """ 获取设备id :return: """ return self.adb('get-serialno').read().strip() def get_device_android_version(self): """ 获取设备Android版本 :return: """ return self.shell('getprop ro.build.version.release').read().strip() def get_device_sdk_version(self): """ 获取设备SDK版本 :return: """ return self.shell('getprop ro.build.version.sdk').read().strip() def get_device_mac_address(self): """ 获取设备MAC地址 :r
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：Python 修改CSV文件实例详解_python_
                                                                
下一篇：python常用数据结构字典梳理_python_