mobile-rpa/utils/utils.py

import json

import uiautomator2 as u2


class UiElement(object):

    def __init__(self, d: u2.Device, xpath: str = None, description: str = None, debug: bool = True, point: dict = None) -> None:
        x, y = 0, 0
        self.info = {}
        self.d = d
        self.debug = debug

        if point is not None:
            self.info['point'] = point
            self.info['xpath'] = xpath
            self.info['description'] = description
            return

        if xpath is not None:
            self.info['xpath'] = xpath
            el = d.xpath(xpath)
            x, y = el.center()
        elif description is not None:
            self.info['description'] = description
            el = d(description=description)
            x, y = el.center()

        self.info['point'] = {'x': x, 'y': y}

    def log(self, *args, sep=' ', end='\n', file=None):
        if self.debug:
            print(args, sep=sep, end=end, file=file)

    def click(self):
        """
        点击
        """
        point = self.info['point']
        self.log("点击元素：", self.info)
        self.d.shell(f"input tap {point['x']} {point['y']}")

    def text(self, text: str = None):
        """
        设置文本框内容
        text 文本内容
        """
        point = self.info['point']
        self.log("点击元素：", self.info)
        self.d.shell(f"input tap {point['x']} {point['y']}")
        self.log("设置文本：", text)
        self.d.shell(f"input text '{text}'")  # 输入文本内容

    def clear(self, length: int = 10):
        """
        清空文本框
        """
        point = self.info['point']
        self.log("点击元素：", self.info)
        self.d.shell(f"input tap {point['x'] + length} {point['y']}")
        self.log("清空文本框：")
        # 模拟删除键输入
        # self.d.shell(f"for i in {{1..{20}}}; do input keyevent KEYCODE_CLEAR; done")  # 20 次删除键事件
        for _ in range(length):
            self.d.shell("input keyevent 67")


class UiFactory(object):
    """
    这个类对 u2.connect 的几个方法做了封装，主要解决耗时问题。
    并且做缓存。不重复定位。

    可以使用 load_point 方法 加载 xxx_point.json中，已经配置好的坐标。
    """

    _cache = {}

    def __init__(self, serial: str):
        """
        :param serial: 设备序列号。 例如 127.0.0.1:6555
            可以通过 `adb devices`  获取
        """
        self.d = u2.connect(serial)
        pass

    def xpath(self, alisa: str, xpath: str = None):
        """
        通过 xpath获取元素
        alisa 别名
        xpath  如果别名没有获取缓存， 会通过xpath 重新定位，并且缓存
        """
        if alisa in self._cache:
            return self._cache[alisa]

        el = UiElement(self.d, xpath=xpath)

        self._cache[alisa] = el

        return el

    def desc(self, alisa: str, desc: str = None):
        """
        通过文本获取元素
        alisa 别名
        desc 如果别名没有获取缓存， 会通过description 重新定位，并且缓存
        """
        if alisa in self._cache:
            return self._cache[alisa]

        el = UiElement(self.d, description=desc)

        self._cache[alisa] = el

        return el

    def load_point(self, point_json: str):
        __tmp__ = {}
        with open(point_json, "r", encoding="utf-8") as f:
            __tmp__ = json.load(f)

        for k, v in __tmp__.items():
            if 'xpath' in v:
                self._cache[k] = UiElement(self.d, xpath=v['xpath'], point=v['point'])
            elif 'description' in v:
                self._cache[k] = UiElement(self.d, description=v['description'], point=v['point'])

    def save_point(self, point_json: str):
        """
        保存坐标
        """
        __tmp__ = {}
        for k, v in self._cache.items():
            __tmp__[k] = v.info

        with open(point_json, "w", encoding="utf-8") as f:
            json.dump(__tmp__, f, ensure_ascii=False, indent=4)

    def check_app_running(self, appid: str):
        # 获取所有运行的APP
        running_apps = self.d.app_list_running()
        print("安装列表：", running_apps)
        # 如果app没有启动， 启动APP
        if appid not in running_apps:
            print(f"{appid} 重新启动")
            self.d.app_start(appid)
            return True
        return False