当前位置:   article > 正文

手机群控爬取实战

手机群控爬取实战

准备工作

准备多部手机,或者模拟器然后将它们于与电脑连接,然后通过 adb 命令查看连接状态

模拟器设置链接: 写文章-CSDN创作中心

安装一个库: pip install adbutils

测试是否链接

输入 adb devices 

如果结果中的第二列显示的不是 device 则有可能手机没设置好

或者 结束 adb 进程  adb kill-server

重新输入  adb devices  尝试几次

群控

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

import adbutils

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

[AdbDevice(serial=127.0.0.1:62001), AdbDevice(serial=127.0.0.1:7555), AdbDevice(serial=emulator-5554)]

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

群控实战

新建一个类 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_reinstall : 是否需要重装安装包,因为有时候不需要重装,所以预留该参数来控制是否需要重装

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

注意: 这里在 __init__ 中新加了两个初始化内容

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

class Controller(object):
    def __init__(self, device_uri, device_name, package_name, apk_path, need_reinstall=False, need_restart=False):
        self.device_uri = device_uri
        self.device_name = device_name
        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 = AndroidUiautomationPoco(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.Anfroid 对象,并将该对象赋值给全局变量 device

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

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

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

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

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 次上滑操作。 下面在实现一个总调用方法

scrape-app5.apk 的下载地址 : https://app5.scrape.center

PACKAGE_NAME = 'com.goldze.mvvmhabit'
APK_PATH = 'scrape-app5.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.device_list():
        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

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

关于爬取数据的流程代码

写文章-CSDN创作中心

最后需要,自己手动将两个方法进行整理才能最终实现爬取

关于商业级群控系统参考: https://setup.scrape.center/multi-control  

本文内容由网友自发贡献,转载请注明出处:https://www.wpsshop.cn/w/码创造者/article/detail/1009161
推荐阅读
相关标签
  

闽ICP备14008679号