Python（一）实现一个爬取微信小程序数据并定时秒杀的爬虫+工程化初步实践_微信小程序爬虫

文章目录

• 前言
• 用Charles 抓包 iOS 微信小程序
• • 在Mac端和iOS端安装Charles 自签名证书
  • • Mac端
    • iOS端
  • 能抓到Safari浏览器的包但是抓不到微信小程序的包
  • 直接在iOS 上抓包的App
  • 如何抓取Android 7.0 以上/Harmony OS微信小程序包
  • 抓包获取token和请求数据格式
  • • token的过期时间
• Python 项目工程化
• • pip 切换为国内镜像源
  • 工程化参考
  • • 脚手架
  • Python 虚拟环境
• 实现爬虫
• • 动态IP
  • • 确保代理服务器的延迟够低
    • 付费HTTP代理选购指南
    • • 明确自己的需求
      • 按量付费 vs 静态IP
  • 设置User-Agent
  • 发起爬虫请求
  • • 设置请求的证书
    • 关闭其他科学上网工具
    • 优化爬虫速度
    • • 使用多线程提高并发
      • 提前完成TCP三次握手以建立HTTP连接
      • 根据过去请求延迟统计，提前N毫秒发起请求
      • 应对竞争爬虫
  • 推送爬取结果到iOS
  • 未解决的问题
  • • 打码平台破解验证码
• 备注

前言

公司准备用Python替换传统的Shell来做自动化运维，最近正好在做这方面的code review，试着用Python写一个小爬虫，顺便入门一下Python。

该爬虫的功能是：

1. 对目标小程序定时发起下订单请求
2. 将下订单请求结果推送到iOS以提醒成功或失败

用Charles 抓包 iOS 微信小程序

原理是：电脑和手机处于同一网络中，在电脑上安装Charles，电脑和手机都安装Charles 自签名证书，然后更改手机网络设置，将手机上的网络请求转发至电脑上的Charles以实现抓包

在Mac端和iOS端安装Charles 自签名证书

Mac端

设置步骤见Charles 文档 > SSL Certificates > MacOS部分。

1. 安装完毕 在 钥匙串访问 中设置为始终信任
   
2. 在 Proxy > SSL Proxy Settings > Include 中添加任意域名 ，然后在浏览器中访问该网站来测试 Charles是否可以解析HTTPS数据包。如果想对所有流量进行抓包，域名设置为*, 端口设置为443，不过不建议这样做，这样做会有大量的我们不关心的包，会对我们造成干扰。这种做法建议只用于测试配置是否正确

iOS端

设置步骤见Charles 文档 > SSL Certificates > iOS 部分 。

1. 在iOS 浏览器中下载且安装完证书之后，在设置 > 通用 > 关于本机 > 证书信任设置 中启用该证书
2. 修改网络设置 > 配置代理 > 手动 > IP （可以在Charles > Help > Local IP Address 中找到），Port 一般为8888
3. 打开浏览器 访问任意网站，注意：这里要和Charles 中的 SSL Proxy Settings > Include 对应。查看Charles是否可以解析HTTPS数据

能抓到Safari浏览器的包但是抓不到微信小程序的包

经过上面的测试之后，开始正式抓小程序的包。 Mac端和iOS端 的浏览器都可以抓到包，不过却发现无法抓小程序的包。 Google了一下发现，需要打开 设置 > App > 微信 > 本地网络

感谢这位博主的分享

直接在iOS 上抓包的App

为什么会有这个需求呢？是因为 我觉得在iOS上抓包这么做太麻烦了，想着有没有直接可以安装在iOS上的App，还真有

1. Charles iOS版，58 人民币，还未购买使用
2. Stream，免费。但是经过实际测试，该App已经很久没更新了，无法抓HTTPS包
3. 蜻蜓抓包，免费。未经过测试使用

如果还有其他好用的iOS抓包 App，欢迎评论区留言推荐

如何抓取Android 7.0 以上/Harmony OS微信小程序包

在另一台华为手机 微信中抓包发现抓不到，系统为harmony os 4.2。Google了一下，发现很多人都有这个问题。简单来说，在 Android7.0 及以上的系统中，App只信任系统预装证书而不信任用户安装的证书。由于主力机是iPhone，我就没有深入研究如何解决这个问题，想解决这个问题可参考 知乎回答 和 另外一位博主分享

抓包获取token和请求数据格式

经过抓包，得到了小程序下单的请求数据格式；小程序自定义的token来自于其内部的 /api/base/wxLogin，请求响应里包含了token；该请求参数只有一个名为code的参数，应该是wx.login返回的。后面经过了解，发现该小程序应该用的是静默登录，关于小程序的静默登录流程以及原理建议阅读这位博主的文章

token的过期时间

经过实际测试，这个自定义的token失效时间是不确定的，经过上面的文章了解，我猜测这个小程序的token失效时间应该是以微信的session_key维护时间为准的，也就是，只要session_key有效，那么这个token就是有效的。而关于session_key的过期时间，这篇文章提到

微信不会把 session_key 的有效期告知开发者。我们会根据用户使用小程序的行为对 session_key 进行续期。用户越频繁使用小程序，session_key 有效期越长

Python 项目工程化

经过上面的抓包，拿到了目标小程序的请求格式和数据格式。开始写Python脚本，既然是个项目，不如从一开始就规范起来，使用企业级的Python项目工程化结构，包括：使用流行的包管理工具，代码风格，代码风格检测，单元测试，打包等等

pip 切换为国内镜像源

最好切换为国内镜像源，这样下载包更快更稳定。

我使用清华大学的镜像源

pip config set global.index-url https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simplepip config set global.trusted-host mirrors.tuna.tsinghua.edu.cn

工程化参考

关于Python项目的工程化我使用的python项目工程化指南里提到的一些组件，建议阅读该指南并学习其中 快速上手 章节的小例子

脚手架

1. 上述指南中使用的脚手架是cookiecutter，如果你也使用，需要在初始化好之后升级一些依赖的版本，脚手架生成的一些依赖的版本现在比较低了，如果直接使用会报错
2. cookiecutter 的提交信息提示已经是5年前了，另外一个比较活跃的脚手架项目是pyscaffold。我还没研究，等后面研究透了再在其他博文中详细介绍。

本文使用的cookiecutter

Python 虚拟环境

Python虚拟环境推荐使用poetry。激活Python项目的虚拟环境命令见poetry > Managing environments文档

实现爬虫

根据上述指南生成项目脚手架且升级了相关依赖之后，开始正式写爬虫代码。核心爬虫逻辑就几行，构造下订单数据，然后使用requests发起请求

动态IP

在测试过程中，发现目标小程序添加了同一个IP 1秒内不能访问2次的限制。找了一家国内付费的TLS代理服务的公司，买了个IP池。国内付费HTTP/TLS代理的公司对于个人用户推出的套餐基本差不多，有按照数量计费的，有按照小时/天计费的。我选的是按数量计费并且所选IP失效为5分钟左右。

在选择具体产品时要关注

1. IP得是 高度匿名代理，这样才不会被目标服务器追踪到原始客户端IP
2. IP质量，代理IP要够稳定且延迟低

确保代理服务器的延迟够低

要让代理请求到目标小程序的延迟够低，首先选择的IP最好和目标小程序所在区域一样。经过上面的抓包，得知小程序的域名，使用dig命令通过域名查到该域名的IP

dig www.xxxx.com

再通过IP归属地查询即可得知该主机IP位于哪里

在接入HTTP代理服务时，商家一般提供可选城市，选择离主机IP最近的城市即可。

当然了还要综合考量，不一定离目标网站所在地越近的代理IP速度越快，应该还和商家自身的硬件部署有关，所以如果发现离目标网站所在地的IP反而更慢，那就果断切换至其他节点

即使完成了上述步骤，商家给你的IP池不一定每一个延迟都很低，所以在拿到IP之后，要测试一下该IP到目标小程序的速度，如果速度不满足则丢弃重新获取新的IP，重复上述步骤直至获得满足延迟的IP

\"\"\"get low latency proxy servers\"\"\"import jsonimport loggingimport timeimport requestslogger = logging.getLogger(__name__)PROXY_SERVICE_PROVIDER_URL = (\"这里是HTTP代理服务商家的接入API\")TARGET_SERVER_URL = \"这里是目标小程序的API\"def get_proxy_ips_latency_less_than_one_second(need_ip_nums: int) -> dict: \"\"\"return a set of ip latency less than 1 second\"\"\" good_proxy_ip = {} request_start_time = time.time() proxy_index = 1 while len(good_proxy_ip) < need_ip_nums : reponse = requests.get(PROXY_SERVICE_PROVIDER_URL, timeout=5) logger.debug(json.dumps(reponse.json(), indent=4, ensure_ascii=False)) ip = reponse.json()[\"data\"][0][\"ip\"] port = reponse.json()[\"data\"][0][\"port\"] proxy_ip = f\'http://{ip}:{port}\' proxies = { \"http\": proxy_ip, \"https\": proxy_ip } if test_proxy_delay(proxies, TARGET_SERVER_URL, 1): good_proxy_ip[proxy_index] = proxies proxy_index += 1 request_end_time = time.time() logger.info(\"successfully get a batch of proxy IPs with a delay\" \"of less than or equals 1 second, cost %d seconds\", request_end_time - request_start_time) logger.info(\"IP proxies：\\n%s\", good_proxy_ip) return good_proxy_ipdef test_proxy_delay(request_proxies, target_url: str, request_timeout: int) -> bool: \"test proxy ip\'s latency whether less than timeout seconds\" request_start_time = time.time() logger.debug(\"start time: %s\", time.strftime(\"%H:%M:%S\", time.localtime(request_start_time))) # add try-except block to avoid program crashes try: response = requests.get(target_url, timeout=5, proxies=request_proxies) if response.status_code == 200 and response.json().get(\"code\") == 200 and response.json().get(\"msg\") == \"操作成功\": request_end_time = time.time() logger.debug(\"end time: %s\", time.strftime(\"%H:%M:%S\", time.localtime(request_end_time))) delay = request_end_time - request_start_time logger.debug(\"Proxy %s response time: %.2f seconds\", request_proxies, delay) return delay <= request_timeout return False except requests.exceptions.RequestException as e: logger.error(\"Error testing proxy %s: %s\", request_proxies, str(e)) return False

付费HTTP代理选购指南

明确自己的需求

我的需求就是代理服务器到目标服务器的延迟足够低，由于是秒杀业务，所以如果代理延迟高，错过开售的那一两秒，后面就没什么意义了。我对代理的延迟要求在100ms以内

按量付费 vs 静态IP

1. 对于我来说，调查了几家HTTP代理商的按量付费套餐的IP，发现这个套餐基本做不到我要的低延迟，例如100ms以内
2. 静态IP 没怎么试用过，不过根据各家厂商宣传，静态IP延迟足够低，都是自己机房的专线。很心动，但是静态IP实在是太贵了

设置User-Agent

伪造 User-Agent，使用fake-useragent。使用该库时注意，多次请求/多线程 应只使用同一个对象，避免多次初始化对象浪费时间

ua = UserAgent(platforms=\'mobile\')ua.random

发起爬虫请求

最终在发起请求时，设置reqeusts的proxies即可

response = requests.post(request_url,request_json_data, headers=request_header, timeout=5, verify=CERT_PATH, proxies=request_proxies)

设置请求的证书

分三种情况

1. 如果你想在请求过程中开着Charles，这时的证书来自Charles。把Charles的证书保存下来，这时CERT_PATH是Charles自己证书的路径。Charles 官方文档 > Python 提到了这一点
2. 如果你已经抓到了所需要的目标小程序的数据格式，那么请求时无需再开Charles代理。这时的证书来自 浏览器打开 目标小程序 域名 > 查看证书 > 下载 。下载前点击证书详情，并确保这个证书的名字不带有Charles，如果有，则关闭Charles并在浏览器中 删除目标小程序的cookie，重新加载，即可获得小程序后台服务器的证书
3. 也可以在请求时不指定verify参数

关闭其他科学上网工具

有的科学上网工具，如果你没设置好，无论国内国外的流量都会先经过它的节点，这样请求反而是慢了很多

优化爬虫速度

使用多线程提高并发

我这里还没有使用scheduler，只是使用 threading.Thread方法。线程的执行逻辑是，线程启动之后，即准备数据，即

1. 获得一个延迟足够低的代理IP
2. 构造请求数据
3. 在requests.post前一行计算当前时间距离目标时间的毫秒数，然后time.sleep 休眠。等到目标时间一到，所有线程立刻同时发起请求而无需等待其他步骤

提前完成TCP三次握手以建立HTTP连接

由于现在使用的是按量付费套餐，代理IP延迟基本在1秒内。所以如果时间到了，再发起请求建立TCP连接，请求数据，如果不算代理延迟，起码慢了几百个毫秒，这几百个毫秒就是用来完成TCP三次握手的；如果算上代理延迟，那基本请求到达服务器要到开始时间的2秒以后了。

于是优化代码逻辑，在爬虫开始前几秒，对目标网站发起一个HEAD 请求，此请求的目的是完成TCP三次握手以建立HTTP连接。同时使用requests.Session保证HTTP连接keep-alive。由于我们的请求是到代理服务器，代理服务器到目标网站，这中间的两个连接是不是长连接不确定。根据日志来看这两个长连接是保持住的了

如何得知的？

1. 项目开启debug模式
2. 发起请求
3. 请求时urllib3的connectionpool.py会打印如下日志

 Starting new HTTPS connection

4. 过几秒再次请求，发现没有如上日志，说明用的是同一个连接

就这一个小小的优化，让我的爬虫有了质的提高，领先其他人一步。所以说，计算机基础还是很重要滴！

根据过去请求延迟统计，提前N毫秒发起请求

每次爬虫都会记录发起请求时间和结束请求时间，根据统计，就可以估计出代理服务器到目标服务器的延迟，所以在秒杀开售前的N毫秒发起请求，保证在门票开售时，请求就到达了服务器。

例如：秒杀开始时间晚上8:30，根据统计，代理服务器到目标服务器延迟大概在300-400ms左右，那么就是8:29:59:700 时发起请求

应对竞争爬虫

在实际爬取过程中，发现还存在其他竞争对手。如何打败对方呢？从实际情况来看，要面对的不止竞争爬虫还有正常用户发起的请求，如何让服务器先处理自己发起的请求呢？

1. 更低的延迟。但是目前在使用动态IP按量付费的套餐下，延迟没办法控制。倒是可以使用自己家高质量的网络而不使用代理，但是我又不想暴露自己IP
2. 更多的请求。增加线程数，例如增加到几十个，甚至上百个，以量抢占服务器处理请求的线程资源（例如Tomcat用来处理请求的线程），如果服务器用来处理请求的线程都被我们的爬虫占满，那其他请求自然要排队。不过如果控制不好，就变成了DDoS攻击

推送爬取结果到iOS

由于该爬虫是定时执行，有的时候不一定在家。所以需要一个将爬取结果推送到iOS上。

鉴于APNs即Apple Push Notification service 有点复杂且不想花时间在上面，于是使用Bark来帮助快速开发。接入步骤非常简单，建议阅读文档

未解决的问题

经过上述步骤，mini版的爬虫基本满足了自己的需求，即定时下单并推送消息到iOS提醒我付款。但是有以下几个问题未解决

1. 由于猜测token失效时间是根据session_key来的，所以在拿到了一个有效token之后，真正爬虫开始之前的几个小时，要偶尔使用下该小程序，以保证session_key不过期。不过可以另辟蹊跷，可以每天手动将小程序删除，然后重新访问，即可在爬取之前得到一个全新的token，就无需考虑token续期的问题
2. 该小程序防止爬虫请求只在用户ID层面制订了防护策略，即只允许同一用户下两单，并没有接入验证码之类的防护

打码平台破解验证码

如果未来该小程序接入了验证码，那么我决定使用付费的打码平台来进行破解。

有关打码平台的介绍，可参考打码平台是如何高效的破解市面上各家验证码平台的各种形式验证码的？ 这篇文章

备注

由于该爬虫核心逻辑（就是发起一个API请求）足够简单，就不再提供示例源码了，只提供上述思路。有关Python的其他最佳实践日后会在其他博文中介绍，本篇只是让各位同学对Python工程化和基本的爬虫技术有个整体的了解