文章目录
前言
公司准备用Python替换传统的Shell来做自动化运维，最近正好在做这方面的code review，试着用Python写一个小爬虫，顺便入门一下Python。
该爬虫的功能是：
- 对目标小程序定时发起下订单请求
- 将下订单请求结果推送到iOS以提醒成功或失败
用Charles 抓包 iOS 微信小程序
原理是：电脑和手机处于同一网络中，在电脑上安装Charles，电脑和手机都安装Charles 自签名证书，然后更改手机网络设置，将手机上的网络请求转发至电脑上的Charles以实现抓包
在Mac端和iOS端安装Charles 自签名证书
Mac端
设置步骤见Charles 文档 > SSL Certificates > MacOS部分。
- 安装完毕 在 钥匙串访问 中设置为始终信任
- 在 Proxy > SSL Proxy Settings > Include 中添加任意域名 ，然后在浏览器中访问该网站来测试 Charles是否可以解析HTTPS数据包。如果想对所有流量进行抓包，域名设置为*, 端口设置为443，不过不建议这样做，这样做会有大量的我们不关心的包，会对我们造成干扰。这种做法建议只用于测试配置是否正确
iOS端
设置步骤见Charles 文档 > SSL Certificates > iOS 部分 。
- 在iOS 浏览器中下载且安装完证书之后，在设置 > 通用 > 关于本机 > 证书信任设置 中启用该证书
- 修改网络设置 > 配置代理 > 手动 > IP （可以在Charles > Help > Local IP Address 中找到），Port 一般为8888
- 打开浏览器 访问任意网站，注意：这里要和Charles 中的 SSL Proxy Settings > Include 对应。查看Charles是否可以解析HTTPS数据
能抓到Safari浏览器的包但是抓不到微信小程序的包
经过上面的测试之后，开始正式抓小程序的包。 Mac端和iOS端 的浏览器都可以抓到包，不过却发现无法抓小程序的包。 Google了一下发现，需要打开 设置 > App > 微信 > 本地网络
感谢这位博主的分享
直接在iOS 上抓包的App
为什么会有这个需求呢？是因为 我觉得在iOS上抓包这么做太麻烦了，想着有没有直接可以安装在iOS上的App，还真有
- Charles iOS版，58 人民币，还未购买使用
- Stream，免费。但是经过实际测试，该App已经很久没更新了，无法抓HTTPS包
- 蜻蜓抓包，免费。未经过测试使用
如果还有其他好用的iOS抓包 App，欢迎评论区留言推荐
如何抓取Android 7.0 以上/Harmony OS微信小程序包
在另一台华为手机 微信中抓包发现抓不到，系统为harmony os 4.2。Google了一下，发现很多人都有这个问题。简单来说，在 Android7.0 及以上的系统中，App只信任系统预装证书而不信任用户安装的证书。由于主力机是iPhone，我就没有深入研究如何解决这个问题，想解决这个问题可参考 知乎回答 和 另外一位博主分享
Python 项目工程化
经过上面的抓包，拿到了目标小程序的请求格式和数据格式。开始写Python脚本，既然是个项目，不如从一开始就规范起来，使用企业级的Python项目工程化结构，包括：使用流行的包管理工具，代码风格，代码风格检测，单元测试，打包等等
pip 切换为国内镜像源
最好切换为国内镜像源，这样下载包更快更稳定。
我使用清华大学的镜像源
pip config set global.index-url https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
pip config set global.trusted-host mirrors.tuna.tsinghua.edu.cn
工程化参考
关于Python项目的工程化我使用的python项目工程化指南里提到的一些组件，建议阅读该指南并学习其中 快速上手 章节的小例子
脚手架
- 上述指南中使用的脚手架是cookiecutter，如果你也使用，需要在初始化好之后升级一些依赖的版本，脚手架生成的一些依赖的版本现在比较低了，如果直接使用会报错
- cookiecutter 的提交信息提示已经是5年前了，另外一个比较活跃的脚手架项目是pyscaffold。我还没研究，等后面研究透了再在其他博文中详细介绍。
本文使用的cookiecutter
Python 虚拟环境
Python虚拟环境推荐使用poetry。激活Python项目的虚拟环境命令见poetry > Managing environments文档
实现爬虫
根据上述指南生成项目脚手架且升级了相关依赖之后，开始正式写爬虫代码。核心爬虫逻辑就几行，构造下订单数据，然后使用requests发起请求
动态IP
在测试过程中，发现目标小程序添加了同一个IP 1秒内不能访问2次的限制。找了一家国内付费的TLS代理服务的公司，买了个IP池。国内付费HTTP/TLS代理的公司对于个人用户推出的套餐基本差不多，有按照数量计费的，有按照小时/天计费的。我选的是按数量计费并且所选IP失效为5分钟左右。
在选择具体产品时要关注
- IP得是 高度匿名代理，这样才不会被目标服务器追踪到原始客户端IP
- IP质量，代理IP要够稳定且延迟低
确保代理服务器的延迟够低
要让代理请求到目标小程序的延迟够低，首先选择的IP最好和目标小程序所在区域一样。经过上面的抓包，得知小程序的域名，使用dig命令通过域名查到该域名的IP
dig www.xxxx.com
再通过IP归属地查询得知，该小程序部署在阿里云 > 华北区域 > 青岛
在接入HTTP代理服务时，商家一般提供可选城市，选择离青岛最近的城市即可。
当然了还要综合考量，不一定离目标网站所在地越近的代理IP速度越快，应该还和商家自身的硬件部署有关，所以如果发现离目标网站所在地的IP反而更慢，那就果断切换至其他节点
即使完成了上述步骤，商家给你的IP池不一定每一个延迟都很低，所以在拿到IP之后，要测试一下该IP到目标小程序的速度，如果速度不满足则丢弃重新获取新的IP，重复上述步骤直至获得满足延迟的IP
"""get low latency proxy servers"""
import json
import logging
import time
import requests
logger = 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_ip
def 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
设置User-Agent
伪造 User-Agent，使用fake-useragent。使用该库时注意，多次请求/多线程 应只使用同一个对象，避免多次初始化对象浪费时间
ua = UserAgent(platforms='mobile')
ua.random
发起爬虫请求
response = requests.post(request_url,request_json_data, headers=request_header,
timeout=5, verify=CERT_PATH, proxies=request_proxies)
设置请求的证书
分两种情况
- 如果你想在请求过程中开着Charles，这时的证书来自Charles。把Charles的证书保存下来，这时CERT_PATH是Charles自己证书的路径。Charles 官方文档 > Python 提到了这一点
- 如果你已经抓到了所需要的目标小程序的数据格式，那么请求时无需再开Charles代理。这时的证书来自 浏览器打开 目标小程序 域名 > 查看证书 > 下载 。下载前点击证书详情，并确保这个证书的名字不带有Charles，如果有，则关闭Charles并在浏览器中 删除目标小程序的cookie，重新加载，即可获得小程序后台服务器的证书
关闭其他科学上网工具
有的科学上网工具，如果你没设置好，无论国内国外的流量都会先经过它的节点，这样请求反而是慢了很多
使用多线程提高并发
我这里还没有使用scheduler，只是使用 threading.Thread方法。线程的执行逻辑是，线程启动之后，即准备数据，即
- 获得一个延迟足够低的代理IP
- 构造请求数据
- 在requests.post前一行计算当前时间距离目标时间的毫秒数，然后time.sleep 休眠。等到目标时间一到，所有线程立刻同时发起请求而无需等待其他步骤
推送爬取结果到iOS
由于该爬虫是定时执行，有的时候不一定在家。所以需要一个将爬取结果推送到iOS上。
鉴于APNs即Apple Push Notification service 有点复杂且不想花时间在上面，于是使用Bark来帮助快速开发。接入步骤非常简单，建议阅读文档
未解决的问题
经过上述步骤，mini版的爬虫基本满足了自己的需求，即定时下单并推送消息到iOS提醒我付款。但是有以下几个问题未解决
- token 失效和刷新机制尚不清楚且获取token的请求我也没找到。导致每次真正爬取之前都要验证下token是否失效，如果失效了，还得重新抓取请求获得有效token
- 该小程序防止爬虫请求只在用户ID层面制订了防护策略，即只允许同一用户下两单，并没有接入验证码之类的防护
打码平台破解验证码
如果未来该小程序接入了验证码，那么我决定使用付费的打码平台来进行破解。
有关打码平台的介绍，可参考打码平台是如何高效的破解市面上各家验证码平台的各种形式验证码的？ 这篇文章
备注
由于该爬虫足够简单，就不再提供示例源码。有关Python的其他最佳实践日后会在其他博文中介绍，本篇只是让各位同学对Python工程化和基本的爬虫技术有个整体的了解