手机群控爬取实战

我们已经学习的使用 Airtest 爬取 App 数据的流程，仅限爬取一部手机，如果想同时爬取多部手机，该怎么办呢?

本节就来探讨一下如何基于 Airtest 实现手机群控，即同时爬取两部及以上手机的数据。

准备工作

请准备好多部移动设备, 真机或模拟器都可以, 然后将它们与电脑相连(能通过 adb 命令行访问即 可)。这里我配置了三部手机, 运行如下命令可以查看当前的连接状态：

adb devices

运行结果如下:

* daemon not running; starting now at tcp:5037
* daemon started successfully
List of devices attached
R5CNCOF9QEX device
emulator-5554 device
emulator-5556 device

如果结果中的第二列显示的不是 device, 请检查手机的配置, 例如检查 USB 调试有没有打开、 手机和电脑有没有正常连接。如果检查完还是没有显示 device, 可以重启 adb 的服务器:

adb kill-server

然后重新运行 adb devices 命令。按这个步骤依次检查每部设备, 直到电脑可以通过 adb 命令正 常访问它们。另外, 这里还需要额外安装 adbutils 库, 通过 pip3 工具安装即可:

pip3 install adbutils

更详细的安装方式可以参考 https://setup.scrape.center/adbutils 。

群控

群控其实很简单, 说白了就是同时控制, 具体到实现上, 就是新建多个进程, 让它们同时执行同 一个逻辑。第一步, 为了能访问到已经连接的多部手机, 我们使用 adbutils 命令替代 adb 命令:

import adbutils

adb = adbutils.AdbClient(host="127.0.0.1", port=5037)
print(adb.devices())

运行结果如下:

[AdbDevice(serial=R5CNCOF9QEX), AdbDevice(serial=emulator-5554), AdbDevice(serial=emulator-5556)]

可以看到返回了一个列表, 列表中的每个元素都是 AdbDevice 对象, 这个 AdbDevice 对象包含一 个 serial 属性, 代表设备序列号, 这和运行 adb devcies 命令获取的结果是一致的。

群控实战

为了更加方便地实现群控, 建议将 12.7 节的实战代码封装成单独的一个类, 由这个类维护对应的 device 对象和 poco 对象, 继而执行一系列操作。下面就新建一个类 Controller, 并初始化一些内容:

class Controller(object):

    def __init__(self, device_name, package_name, apk_path, need_reinstall=False):
        self.device_name = device_name
        self.package_name = package_name
        self.apk_path = apk_path
        self.need_reinstall = need_reinstall

对于群控, 需要批量实现一些操作, 包括初始化设备和安装 apk 安装包等。所以这里在构造方法 中声明了 4 个参数:

  • device_name: 刚才使用 adb devices 命令获取的各个设备的序列名称。

  • package_name: 包名。

  • apk_path: 安装包文件的路径, 这个参数是为安装所用的, 因为很多手机可能没有安装过安装包, 所以用该参数来指定安装包的路径。

  • need_install: 是否需要重装安装包, 因为有时候不需要重装, 所以预留该参数来控制是否需 要重装。

然后添加一些常用的初始化方法:

from airtest.core.api import *
from poco.drivers.android.uiautomation import AndroidUiAutomationPoco

class Controller(object):

    def __init__(self, self, device_uri, package_name, apk_path, need_reinstall=False, need_restart=False):
        self.device_uri = device_uri
        self.package_name = package_name
        self.apk_path = apk_path
        self.need_reinstall = need_reinstall
        self.need_restart = need_restart

    def connect_device(self):
        self.device = connect_device(self.device_uri)

    def install_app(self):
        if self.device.check_app(self.package_name) and not self.need_reinstall:
            return
        self.device.uninstall_app(self.package_name)
        self.device.install_app(self.apk_path)

    def start_app(self):
        if self.need_restart:
            self.device.stop_app(self.package_name)
        self.device.start_app(self.package_name)

    def init_device(self):
        self.connect_device()
        self.poco = AndroidUiAutomati onPoco(self.device)
        self.window_width, self.window_height = self.poco.get_screen_size()
        self.install_app()
        self.start_app()

下面介绍一下添加的几个方法。

  • connect_device: 里面直接调用了 Airtest 的 connect_device 方法, 需要传入一个参数 device_uri, 会返回一个 device 对象, 这里其实是 airtest.core.android.android.Android 对象, 并将该对 象赋值给全局变量 device。

  • install_app: 里面使用 device 变量的 check_app 方法检查 App 有没有安装, 使用 need_resintall 方法检查是否需要重装 App, 只有在 App 已经安装且不需要重装的时候才不做任何操作。在其 他情况下, 都需要重装这个 App, 先使用 uninstall_app 方法卸载 App, 再通过 install_app 安装。

  • start_app: 里面使用 need_restart 判断是否需要重启 App, 如果需要, 就先停止 App 再启用, 否则直接启用 App。

  • init_device: 这是一个初始化方法, 里面先调用 connect_device 方法连接了设备, 接着将 device 对象传给 AndroidUiAutomati onPoco 构造了一个 poco 对象, 然后获取了一些基础参数, 例如 Window 屏幕的宽高, 最后调用 install_app 和 start_app 方法完成了 App 的安装和启动。

到这里, 其实我们就能控制手机安装和重启 App 了。下面继续往 Controller 类中添加两个方法:

class Controller(object):
    ...

    def scroll_up(self):
    self.device.swipe((self.window_width * 0.5, self.window_height * 0.8),
    (self.window_width * 0.5, self.window_height * 0.3), duration=1)

    def run(self):
    for _ in range(10):
    self.scroll_up()

添加的方法一个是 scroll_up, 里面调用了 device 对象的 swipe 方法。另一个是 run, 里面调用 了 10 次上滑操作。下面再实现一个总的调用方法:

PACKAGE_NAME = 'com.goldze.mvvmhabit'
APK_PATH = 'scrape-apps.apk'

def run(device_uri):
controller = Controller(device_uri=device_uri,
package_name=PACKAGE_NAME,
apk_path=APK_PATH,
need_reinstall=False,
need_restart=True)

controller.init_device()
controller.run()

注意这里的 scrape-app5.apk 需要下载下来, 和当前代码放在同一文件夹夹下, 这样在安装 apk 的时候才能找到对应的安装包。最后完善 一下群控的调用逻辑即可:

from multiprocessing import Process

if __name__ == '__main__':
    processes = []
    adb = adbutils.AdbClient(host="127.0.0.1", port=5037)
    for device in adb.devices():
        device_name = device.serial
        device_uri = f'Android:///{device_name}'
        p = Process(target=run, args=[device_uri])
        processes.append(p)
        p.start()
    for p in processes:
        p.join()

这里我们就是使用多进程实现了手机群控, 一个爬取进程对应一个 Process 进程, 声明进程的时候直接指定目标方法为 run, 参数就设 置为设备的连接字符串, 格式为 Android:///{device_name}, 例如 Android:///emulator-5554。

在本节的案例中, 由于我连接了三部手机, 所以就新建了三个进 程, 它们同时执行数据爬取操作。运行代码之后, 可以发现三部手机 同时运行着爬取流程, 如图 12-85 所示。

图12-85 运行着爬取流程的手机

总结

本节介绍了手机群控爬取的简单实现，由于我们使用的是 Python 脚本，所以可以直接使用多进程 multiprocessing 库中的 Process 模块为每个爬取进程建立单独的进程，最终成功实现了群控爬取。

本节代码见 https://github.com/Python3WebSpider/AirtestTest

商业服务

以上演示的仅是一个简单案例，通过多进程实现群控爬取自然没有问题不过距离真正商业级的手机群控还是有一定差距的。

现在市面上的手机群控系统支持同时控制上百部手机长时间稳定运行，并不仅仅满足于爬取一个简单的项目，手机也几乎都是真机，被统一放置在一个支架上维护起来，如图 12-86 所示。

图12-86 利用支架维护手机

关于商业级群控系统，由于其类型五花八门且经常发生变动和更新，故这里不再展开介绍，大家可以参考 https://setup.scrape.center/multi-control 了解更多信息。

下面是一些群控系统服务网站（不定期更新）：