Skip to main content

自动化框架

自动化框架

从我们访问一个网址开始,到在浏览器看到漂亮的界面,这个过程经历了许多步骤。然后当你断开网络会发现,页面内容依然没有消失,他们在哪里呢?

我们知道网页分为三部分,互相协作:

  • HTML 定义了网页的内容
  • CSS 描述了网页的布局
  • JavaScript 控制了网页的行为

其实还可以加上一项

  • 资源(多媒体)在合适的时候被调用

我们通过查询谷歌浏览器开发者协议(Chrome DevTools Protocol)可以找到这条命令,其实所有浏览器开发者协议都可以,这里只是以谷歌为例。

那接着我们要做的事情就是:通过浏览器工具,获取完整的浏览器缓存,接着读取缓存就能找到对应的数据了,你可以抓取浏览器中的流数据,并根据流数据类型做分类保存;对下载好的页面做修改并截图实现高亮部分文字的效果;下载多媒体等。当你做到这一步时,你的爬虫认知就从定向爬虫转变为了通用型爬虫了。

现在市面上常见的模拟浏览器工具有以下几种,我建议熟读其中 playwright 的开发文档,再阅读其他框架的文档时,会发现大部分都是相通的。

selenium:老牌模拟浏览器 仅同步

Pyppeteer: Puppeteer 的 python 实现,仅支持谷歌浏览器,异步

playwright:支持三种不同浏览器,异步,微软月更

接下来我们通过一个自动搜寻验证码图片并解析的例子来熟悉下整个流程。代码中所有路径部分请替换为你自己的地址,过程中会有极少的数据提取操作,可以参见后续数据提取教程。

selenium

如果你使用的是 selenium,可以参考这段代码

selenium 代码官方文档

from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
import time
import json
import base64
from PIL import Image
from urllib.parse import urlparse
from io import BytesIO

options = webdriver.ChromeOptions()
proxy = False # 代理设置:关闭
if proxy:
options.add_argument('--proxy-server={}'.format(proxy))


d = DesiredCapabilities.CHROME
d['loggingPrefs'] = {'performance': 'ALL'}
options.add_experimental_option('w3c', False) # 实验室选项,w3c协议 关闭

'''你应该经历过有部分网站证书过期,会跳出是否继续访问。这里我们忽视这些错误'''
options.add_argument('--ignore-certificate-errors') # 忽略证书错误
options.add_argument('--ignore-ssl-errors') #忽略ssl错误
options.add_argument('--headless') # 选择无头模式,加载更快
options.binary_location = 'C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe'

driver = webdriver.Chrome('D:\\MyProgram File\\python\\chromedriver.exe',
options=options,
desired_capabilities=d)

driver.execute_cdp_cmd("Network.enable", {}) # 启用网络跟踪,网络事件现在将传递给客户端。

driver.get('https://id.qq.com/vc.html') # 打开一个网址
time.sleep(2)

for f in driver.find_elements_by_tag_name("form"): # 找到所有表单类信息
urls = [i.get_attribute("src") for i in f.find_elements_by_tag_name("img")] #找到表单内所有的图片(大概率只有一个,且是验证码)


for log in driver.get_log("performance"):
log = json.loads(log["message"]) # 减小我们的检索范围
if log["message"]["method"] == "Network.responseReceived": # 精准检索收到的数据

for url in urls:
url = urlparse(url).path #避免网页内的url中的https的有无导致匹配不上

if url in log['message']['params']['response']['url']: # 特征匹配
request_id = log['message']['params']['requestId'] # 获取requestid
data = driver.execute_cdp_cmd("Network.getResponseBody", {"requestId": request_id})['body'] # 提取id对应的数据
data = base64.b64decode(data) # 从cdp开发文档可以得知返回的是base64编码的数据,我们他把转回二进制
source_img =Image.open(BytesIO(data))# 接着读取二进制数据
source_img.show() #打开看看,获取部分到这里就结束了。

如果你遇到 selenium 元素无法点击,可以推测下原因:

  • 窗口没有最大化导致目标元素不可见
  • 没有等待页面渲染完成
  • 没有等待 AJX 异步加载完成
  • 元素在框架内,没有跳转到框架中

Pyppeteer

Pyppeteer 对流抓取支持的不是很好,但是对 Js 定位与执行做的非常不错,所以这段代码稍短一些。

pyppeteer 代码官方文档

import asyncio
from pyppeteer import launch
from urllib import request

class Get_pic(): ## 自动获取验证码图片
async def get_refer(self): # 定义主函数
browser = await launch(headless=False)
page = await browser.newPage()
await page.goto('https://id.qq.com/vc.html') # 打开网址
image_src = await page.Jeval('#img', 'el => el.src') # 锁定图片,获取网络对象
# 第一个参数代表定位方式【.】代表属性【#】代表id 【>】表示下属
# 第二个参数代表要执行的js代码, => 表示匿名函数,输出 src路径
request.urlretrieve(image_src, 'image.png') # 把网络对象复制到本地
await browser.close() # 关闭浏览器

new = Get_pic() #实例化
asyncio.run(new.get_refer()) # 调用方法

playwright

如果你使用的 playwright,可以参考下方的代码。我使用了 2 种方法来实现:

playwright 的监听器整合了 cdp 协议中的接口,因此代码更加简洁。 playwright 通用支持选中元素之后执行 JS 语法。

playwright 代码文档

from PIL import Image
from urllib.parse import urlparse
from io import BytesIO
import asyncio
from urllib import request
from playwright.async_api import async_playwright

class Get_pic(): ## 自动获取验证码图片
async def get_refer(self): # 定义主函数
async with async_playwright() as p: # 创建一个playwright对象
browser =await p.chromium.launch(headless=False) # 启动浏览器,配置一下无头模式
context = await browser.new_context(ignore_https_errors = True,bypass_csp=True) # 切换绕过页面的 Content-Security-Policy
page = await context.new_page() # 创建页面
self.img_data = {}
page.on("response", self.on_response) # 注册一个监听器,读取已加载内容,并把读取到的response传递给on_response方法
await page.goto('https://id.qq.com/vc.html') # 打开网址
image_src = await page.eval_on_selector('#img','el => el.src')
request.urlretrieve(image_src, 'image.png') # 把网络对象复制到本地
image = page.locator("img")
url = await image.get_attribute("src") # 获取页面内的img标签的src属性
self.urls = urlparse(url).query #避免网页内的url中的https的有无导致匹配不上
for u in self.img_data:
if self.urls in u:
source_img =Image.open(BytesIO(self.img_data[u]))
source_img.show()
await browser.close() # 关闭浏览器

async def on_response(self,response):
'''
response 为playwright 的 response对象
'''
self.img_data[response.url] = await response.body()

new = Get_pic() #实例化
asyncio.run(new.get_refer()) # 调用方法

移动端自动化

Frida

移动端的自动化往往基于adb,你可以通过下载雷电模拟器或其他模拟器来获得它,然后在模拟器上进行自动化测试。这里以雷电模拟器为例,介绍如何使用Frida进行移动端自动化。

1. 安装 Frida

首先,我们需要在 Python 环境中安装 Frida。打开终端,输入以下命令:

pip install frida-tools
pip install frida

2. 下载 Frida 服务端

接着,我们需要下载 Frida 服务端。你可以在 Frida 的 GitHub 仓库中找到对应的版本。

例如,如果你需要的版本是 16.1.11,你可以下载 frida-server-16.1.11-android-x86_64.xz。请根据你的设备和需求选择合适的版本。

3. 解压 Frida 服务端

下载完成后,我们需要对 Frida 服务端进行解压。解压后,你会得到一个名为 frida-server-16.1.11-android-x86_64 的文件。

4. 准备 Frida 服务端

为了方便后续操作,我们将 Frida 服务端重命名并移动到一个方便的位置。例如,你可以将它重命名为 frida-s,并移动到 C:\leidian\LDPlayer9\ 目录下(雷电模拟器自带adb)。

5. 设置 Android 设备

接下来,我们需要在 Android 设备上进行一些设置。

如果你是使用自己的手机插桩:打开设备的开发者模式,并允许 USB 调试。然后,设置设备的 Root 权限。 如果你是使用雷电模拟器插桩:在设置-其他设置-打开root权限后重启。

6. 重启 ADB 服务

为了确保 ADB 服务正常运行,我们需要先停止,然后再启动 ADB 服务。在终端中输入以下命令:

CMD
cd C:\leidian\LDPlayer9\
adb kill-server
adb start-server
# 将 Frida 服务端上传到 Android 设备上
adb push frida-s /data/local/tmp/
# 启动 Frida 服务端
adb shell
CMD-adb shell
# 获取 root 权限
su
# 切换到 Frida 服务端所在的目录
cd data/local/tmp
# 设置 Frida 服务端的权限
chmod 777 frida-s
# 启动 Frida 服务端
./frida-s

至此,你已经成功安装并启动了 Frida。你可以开始使用 Frida 进行动态代码插桩了(成功的代码没有任何提示,No news is good news)。

另起一个CMD窗口,输入以下命令:

CMD
cd C:\leidian\LDPlayer9
adb forward tcp:27042 tcp:27042

9. 使用 Frida

C:\leidian\LDPlayer9\test.py
import frida

# 获取远程设备
remote_dev = frida.get_remote_device()
print(remote_dev)

# 获取前台应用
front_app = remote_dev.get_frontmost_application()
print(front_app)

# 枚举进程
processes = remote_dev.enumerate_processes()
for process in processes:
    print(process)

# 输出结果如下
'''
Device(id="socket", name="Local Socket", type='remote')
Application(identifier="jp.co.hit_point.tabikaeru", name="旅行青蛙", pid=3249, parameters={})
.....
Process(pid=2925, name="logcat", parameters={})
Process(pid=2940, name="memfd:frida-helper-32 (deleted)", parameters={})
'''