準備工作

1. pip install --pre uiautomator2 （安裝u2的庫）
2. pip install -U weditor （桌面可生成一個可執行的py）
3. 根據網上的資料去了解u2的使用方法

安裝App主檔案（install_app.py）

import subprocess
import uiautomator2 as u2
import time
import argparse
from threading import Thread
import pymysql.cursors
import pymysql
import sys


def get_serial_and_apk():
    """
    獲取裝置名和包名
    """
    parser = argparse.ArgumentParser()
    parser.add_argument('--serial', type=str)  # 從jenkins獲取需要安裝的App的手機的uid
    parser.add_argument('--upload_filename', type=str)  # 獲取需要安裝的Apk的包名
    parser.add_argument('--replace_install', type=int)  # 這邊是根據公司需要使用原有Apk構建，還是重新覆蓋安裝新的包
    args = parser.parse_args()
    serial = args.serial
    upload_file = args.upload_filename
    replace_install = args.replace_install
    print(replace_install)
    apk_path = "/Users/data/upload_apk/" + upload_file  # 安裝包的路徑，linux為例
    print(apk_path)
    return serial, apk_path, replace_install, upload_file  # 輸出裝置號，apk的安裝路徑，是否覆蓋安裝，包名


def select_install_pwd():
    """
        執行sql語句，主要是去資料庫根據uid去查當前裝置的安裝密碼
        :param
        sql_words: sql語句
        :return: 查詢資料對應的字典列表
    """
    connect = pymysql.Connect(
        host="192.168.1.1",
        port=3306,
        user="root",
        passwd="123456",
        db="auto_build",
        charset="utf8"
    )
    # 獲取遊標
    cursor = connect.cursor(cursor=pymysql.cursors.DictCursor)
    try:
        sql = f"""SELECT install_passwd FROM device_info WHERE id =\'{serial}\'"""
        cursor.execute(sql)
        results = cursor.fetchall()
        for row in results:
            print(row['install_passwd'])
            return row['install_passwd']
    except Exception as e:
        print(e)
        connect.rollback()
    cursor.close()
    connect.close()
    return None


def time_out(start_time, time_out=120.0):
	"""
	定義一個超時的時間2min
	"""
    end_time = time.time()
    difference = round((end_time - start_time), 2)
    if difference - time_out > 0:
        print(f"設定的時間{time_out}到了退出")
        return False
    else:
        return True


def listen_alert(driver):
    """
    監聽彈框
    """
    time.sleep(1)
    if driver(textContains='請輸入').exists():
        pwd = select_install_pwd()
        driver(textContains='請輸入').set_text(pwd)
        print("準備點選確定")
        time.sleep(2)
        if driver(text='確定').exists:
            driver(text='確定').click()
            print("已點選確定")
    if driver(text="繼續安裝").exists():
        driver(text="繼續安裝").click()
        print("點選繼續安裝")
    if driver(text='無視風險安裝').exists():
        driver(text='無視風險安裝').click()
    if driver(text='安裝').exists():
        driver(text='安裝').click()


def cmd(cmd, remove_shell=False):
	"""
	adb命令頭部封裝
	"""
    head = adb_head
    if remove_shell:
        head = adb_head.replace('shell', '')
    print(head + ' ' + cmd)
    import os
    r = os.popen(head + ' ' + cmd)
    print(r.read())
    return r.read()


def create_dir_if_not_exists(remote_path):
	"""
	判斷下手機裝置中是否有我們上傳的一個路徑
	為什麼需要上傳到手機，而不是直接adb install 安裝呢，主要是需要獲取安裝成功的Success的提示，以手機內部安裝更為合適
	"""
    r = cmd('ls ' + remote_path)
    if 'No such file or directory ' in r:
        cmd(' mkdirs ' + remote_path)


def push_file(file_path, remote_path):
	"""
	推Apk到手機中儲存
	"""
    # create_dir_if_not_exists(remote_path)
    cmd(' push ' + file_path + ' ' + remote_path, remove_shell=True)
    print(' push ' + file_path + ' ' + remote_path)


def install(apk_path, apk_name):
    """
    安裝apk
    """
    push_file(apk_path, '/data/local/tmp')  # 上傳apk到手機中
    remote_path = '/data/local/tmp/' + apk_name  # 手機中的安裝包的路徑
    cmd = adb_head + ' pm install  -r ' + remote_path  # 強制安裝
    # cmd = "adb -s %s install %s" % (serial, apk_path)
    process = subprocess.Popen(cmd, shell=True,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
    process.wait(120)
    for line in process.stdout:
        line_str = line.decode("utf-8")
        print(line_str)
        if 'Success' in str(line_str):
            print("安裝包已成功安裝".center(70, "-"))
            return True
        else:
            if 'Failure' in str(line_str):
            	print(line_str)
                print("請檢查當前安裝的apk版本是否過低".center(50, "*"))
                print("注意此時apk安裝失敗，斷開測試連線，請檢查完畢後重新構建".center(50, "*"))
                sys.exit(1)
            return False


if __name__ == '__main__':
    serial, apk_path, replace_install, upload_file = get_serial_and_apk()
    adb_head = 'adb -s ' + serial + ' shell '
    if replace_install == 3:
        p1 = Thread(target=install, args=(apk_path, upload_file))
        p1.start()
        try:
            driver = u2.connect_usb(serial)
        except RuntimeError as e:
            print("檢測到是否是裝置斷開連線！！！".center(70, "*"))
            sys.exit(1)
        while p1.is_alive():
            listen_alert(driver)
        p1.join()
        print("Ending".center(70, "-"))
        try:
            driver.service("uiautomator").stop()
        except Exception as e:
            print("未能停止uiautomator2的程式")
            print(e)
    else:
        print("使用原有的apk進行構建...")