自动化+接口拦截方案的实现

为什么自动化程序需要拦截接口

传统的自动化数据采集，比如浏览器自动化，App自动化都是以来于自动化测试工具实现的，如Selenium，Puppeteer，Appium，U2等。

这种采集方法秉承着可见及可爬的原则，依靠对浏览器网页或者App界面元素的匹配 (如xpath，re等匹配库) 来采集数据。这种方法虽然可行，但是有时候会有返回数据不全，或者漏抓的情况出现。

情况A【数据漏抓】：

有一个下拉刷新的页面，通过观察可以知道每页可以返回100条数据，但是铺满整个屏幕只能看到30条数据，这个时候你可能会计算浏览器屏幕高度和宽度，然后把结果去重一下，结果发现还是会漏抓。

情况B【数据不全】：

抓取商品评论的时候有些网站只展示用户昵称而没有用户id（uuid等唯一标识符），而网站允许用户的昵称相同。那么如果相同的用户评论的内容也相同，则会少抓取一个用户。

那么，如果有一个方法可以在自动化操作的时候，实时监听和拦截数据接口的返回，是不是就可以解决掉这个问题了。 DrissionPage 可以做到。

DrissionPage

特性

DrissionPage 是一个基于 python 的网页自动化工具。
它既能控制浏览器，也能收发数据包，还能把两者合而为一。
可兼顾浏览器自动化的便利性和 requests 的高效率。
它功能强大，内置无数人性化设计和便捷功能。
语法简洁而优雅，代码量少，对新手友好。

环境依赖

支持系统：Windows、Linux、Mac
Python 版本：3.6 及以上
支持应用：Chromium 内核浏览器（如 Chrome、Edge），electron 应用

安装

Bash

pip install DrissionPage
pip install DrissionPage --upgrade

快速启动

默认状态下，程序会自动在系统内查找 Chrome 路径。执行以下代码，浏览器启动并且访问了项目文档，说明可直接使用，跳过后面的步骤即可。

Python

from DrissionPage import ChromiumPage

page = ChromiumPage()
page.get('http://DrissionPage.cn')

监听网络数据

许多网页的数据来自接口，在网站使用过程中动态加载，如使用 JS 加载内容的翻页列表。

这些数据通常以 json 形式发送，浏览器接收后，对其进行解析，再加载到 DOM 相应位置。

做数据采集的时候，我们往往从 DOM 中去获取解析后数据的，可能存在数据不全、加载响应不及时、难以判断加载完成等问题。如果我们可以拿到浏览器收发的数据包，根据数据包状态判断下一步操作，甚至直接获取数据，岂不是爽爆了？

DrissionPage 每个页面对象（包括 Tab 和 Frame 对象）内置了一个监听器，专门用于抓取浏览器数据包。可以提供等待和捕获指定数据包，实时返回指定数据包功能。

下面是一些示例代码：

点击下一页，等待数据包，再点击下一页，循环。

Python

from DrissionPage import ChromiumPage

page = ChromiumPage()
page.get('https://gitee.com/explore/all')  # 访问网址，这行产生的数据包不监听

page.listen.start('gitee.com/explore')  # 开始监听，指定获取包含该文本的数据包
for _ in range(5):
    page('@rel=next').click()  # 点击下一页
    res = page.listen.wait()  # 等待并获取一个数据包
    print(res.url)  # 打印数据包url
    print(res.response.body)  # 打印数据包内容

listen.start() 此方法用于启动监听器，启动同时可设置获取的目标特征。可设置多个特征，符合条件的数据包会被获取。如果监听未停止时调用这个方法，可清除已抓取的队列。
listen.wait() 此方法用于等待符合要求的数据包到达指定数量。所有符合条件的数据包都会存储到队列，wait()实际上是逐个从队列中取结果，不用担心页面已刷走而丢包。

DrissionPage 是一个基于 Python 的网页自动化工具，它既能控制浏览器，也能收发数据包，还能把两者合而为一。

DrissionPage 的监听器功能，可以实时监听和拦截数据接口的返回，可以解决掉数据漏抓和数据不全的问题了，是一个值得推荐的自动化工具。

使用示例

Python

import json
import random
import time

from DrissionPage import ChromiumOptions, WebPage
from loguru import logger


class PDD(object):
    def __init__(self, initial_tag_times):
        self.initial_tag_times = initial_tag_times  # 新增属性来记录初始的tag_times
        self.url = 'https://mobile.mock.com/list_id=oWGE3Y6kLU'
        co = ChromiumOptions().headless(False)
        self.page = WebPage(chromium_options=co)

    def get_html(self):
        """ 根据initial_tag_times决定是否重新加载页面或直接监听数据接口 """
        if self.initial_tag_times == 0:
            self.page.get(self.url)
        self.page.listen.start('https://mobile.mock.com/api/search')
        self.get_data()

    def get_data(self):
        """ 获取数据并处理，记录刷新次数 """
        tag_times = 0
        while tag_times < 300:
            for c in range(10):
                self.page.run_js(f"window.scrollBy(0, {300 * c});")
                time.sleep(random.randint(1, 2))
            pack = self.page.listen.wait()
            html = pack.response.body
            logger.debug(html)
            if html:
                if isinstance(html, str):
                    html = json.loads(html)
                items = html.get('items')
                self.process_tag(items)
                time.sleep(random.randint(1, 2))
                tag_times += 1

使用了 self.page.listen.start('https://mobile.mock.com/api/search') ，在 get_data 方法中添加滑动页面的逻辑，并在滑动的同时继续监听数据接口。

从而实现一边滑动页面一边获取数据。如果接口的 URL 与预期的一致，就处理获取到的数据。