资讯详情

资讯详情

小程序数据采集完整指南:从拆包抓包到模拟请求落地

做小程序采集的人多了之后网上能搜到的资料其实很碎片有人讲抓包有人讲反编译但很少有一篇能把“从拿到一个小程序到最终把数据存进数据库”整条链路讲透。我自己最开始做这个需求时也走了不少弯路先是愣用 requests 去抓 HTML结果连个正经页面都拿不到后来才意识到小程序这套东西跟普通网页爬虫完全是两套玩法。这篇就把我自己跑通的完整思路整理出来从拆包到抓包再从模拟请求到落地框架适合想做小程序数据采集、又不清楚该从哪下手的开发者参考。1. 它跟普通网站爬虫差在哪儿先搞清楚小程序的运行逻辑很多人第一次见到“小程序爬虫”这个需求下意识会把它当成普通网页爬虫来做拿到 URL、伪造 Header、发 GET 请求、解析 HTML。但小程序页面根本不是服务端直接渲染的 HTML它的数据全要从后端 API 拿而且是跑在一个封闭客户端容器里的。你直接拿 Python 请求那些页面 URL返回的往往只有一段基础配置没有业务数据。1.1 为什么 requests 抓不到页面要理解这件事得先明白小程序的基本架构。小程序是“双线程模型”视图层负责渲染 WXML/WXSS逻辑层负责处理数据两层通过微信这个宿主环境通信。它里面没有传统意义上的“网页”只有一个提前打包好的代码包也就是我们后面要拆的 wxapkg里面是一堆 JS、JSON、WXML 文件。当用户打开一个小程序时小程序本身会向它的后端接口发起 wx.request 调用后端返回 JSON 数据前端再把数据渲染到页面上。所以真正有价值的数据全在那一个个 API 请求里而不是在静态页面上。想直接从外面拿到数据本质上就三条路一是拆开小程序代码包把 JS 读透找出所有接口定义和参数规则二是在小程序运行过程中把网络请求拦下来观察真实 API 请求三是把前两步得到的信息整合起来用代码直接模拟请求。1.2 三个常见采集维度我遇到的真实需求里最典型的三类分别是列表类采集比如某个服务小程序里的商品列表、门店列表、动态列表。这类接口大多是分页 JSON结构清晰适合做定时增量采集。详情类采集点进某个条目后的详情页数据往往需要带一个业务 ID 去请求详情接口。搜索/筛选类采集需要构造关键字、筛选条件、排序参数这类接口通常对签名和风控更敏感后面会单独说。另外还有一种比较特殊的场景有些小程序的数据不是通过接口拿的而是直接打包在代码包里。比如一些活动页面的静态配置、Banner 图地址、基础字典表。这类数据不需要发请求就能拿到拆包之后直接去找 JSON 文件就行。1.3 拆解目标小程序的整体思路我的建议是接到采集任务后不要急着开抓包工具先想清楚你拿的小程序是哪种类型内容型小程序重点看代码包里的接口列表和页面渲染逻辑。工具型小程序重点看用户操作触发的请求链路。纯前端展示型可能只需要拆包不需要模拟请求。想清楚路子后面才能少做无用功。接下来我按一条可落地的链路展开讲先拆包拿源码再盯网络层抓真实请求最后写代码模拟接口。2. 拆包从 wxapkg 到可读 JS前端逻辑先摊在桌上小程序发布到线上之前会被编译并打包成.wxapkg文件。这个文件是微信自己的一套格式不是 zip 也不是 tar直接用解压工具打不开。想读懂里面代码必须先把它拆开复原。2.1 wxapkg 包一般躺在哪里要拿到某个小程序的 wxapkg 包最省事的方式是让它在本地环境跑起来再去缓存目录里捞文件。我平时主要用安卓模拟器来完成这一步。在模拟器里装好微信并打开目标小程序随便逛几个页面微信会把小程序包下载到本地存储。常见路径大致是这样的/data/data/com.tencent.mm/MicroMsg/一串哈希/appbrand/pkg/注意这里的一串哈希是每个微信号对应的独立目录不同账号可能不一样。目录下通常会有__APP__.wxapkg这种看似随机的文件名也可能带个版本号后缀。把这类文件按修改时间排序最新修改的那个基本就是刚加载过的小程序主包。提示如果你在跑自动化采集建议把模拟器的系统镜像固定住不要频繁切换账号。换账号意味着重新走一遍下载流程缓存路径会变之前的分析结果也要跟着重新核对。2.2 怎么把包拆开还原社区里目前有不少现成的反编译/解包脚本搜一下wxappUnpacker这个名字就能找到一堆衍生版本。用法大同小异核心是先把 wxapkg 的头信息解析出来再按索引区、文件区、内容区分别提取。拿一个包举例你可以用 Node 跑解包脚本node wuw.js __APP__.wxapkg -o out/-o out/指定输出目录。跑完之后输出目录里一般会出现小程序主页面的几个核心文件out/ ├── app.js ├── app.json ├── app.wxss ├── pages/ │ ├── index/ │ │ ├── index.js │ │ ├── index.json │ │ ├── index.wxml │ │ └── index.wxss │ └── detail/ │ ├── detail.js │ └── ...到了这一步你拿到的是“可读性如何”取决于包本身是否被混淆过。现在很多生产环境小程序做完代码压缩变量名是乱七八糟的短名字a1;b2;cfunction(){}。但不管变量名多难看文件结构、请求 URL、跳转路径这些字符串信息一般是保留的因为运行时必须要用。2.3 拆包后先看什么、再看什么打开解包目录后我的固定顺序是先看app.json。它会列出所有页面路由和全局配置。页面列表能帮你快速确定这个小程序有哪些功能模块以及你关心的数据在哪个页面里。再搜https字符串。直接在pages目录里 grephttps://把所有接口域名找出来通常就一两个主域名。第三次搜关键字。比如你要采集商品数据就搜goods、spu、list、page如果你要采集动态就搜feed、post、comment。顺着这些关键字能定位到具体页面 JS 里的请求函数。以我拆过的一个模拟项目为例pages/index/index.js里通常会有类似这样的片段request({ url: /api/home/list, data: { page: this.data.page, pageSize: 20 }, success: res { this.setData({ list: res.data.list }); } });这里哪是函数名、哪是接口地址、哪是分页字段一目了然。有了这段信息再去抓包验证就快得多。2.4 拆包时的翻车现场拆包不是永远顺利我踩过几个高频坑列出来给后来人提个醒子包缺失小程序超过一定体积会启用分包机制主包只保留入口业务代码被放在subpackages目录下的子包中。你从缓存目录里捞到的可能只有主包后续要进分包页面时分包会被再次下载。解决办法是把目标小程序的每个 Tab/页面都点一遍让所有分包都落到本地再回去重新抓取。wxapkg 加密部分小程序会对包体做自定义加密解包脚本解出来的 JS 散成一堆乱码。这种情况能做的有限一般只能靠抓包去还原接口源码只能当辅助参考。版本不一致微信更新后解包脚本的解析逻辑可能失效。如果你发现脚本解包失败优先去社区找一个更新维护的版本不要自己手改二进制格式。3. 抓包把真实请求拦下来看 API 长什么样拆包能告诉你“代码里写了什么”但代码和运行时的真实行为不一定完全一致。服务端可能根据用户身份、登录态、设备环境动态生成参数也可能在多个接口里拼接数据。这个时候抓包就成了最直接的验证方式。3.1 两大抓包方案怎么选我实际用的方案主要有两个各有利弊方案优点缺点适用场景安卓模拟器 代理抓包工具完整还原移动端行为所见即所得需要处理证书信任模拟器性能吃紧大多数小程序桌面端代理 抓包工具链路简单证书好装部分小程序桌面端交互不同请求会少只需要快速看接口时具体工具方面愿意折腾的可以试开源方案不想折腾的用商业抓包工具也行。核心流程都一样起一个本机代理手机或模拟器把流量指向这个代理代理再对 HTTPS 流量做中间人解密最后把请求展示出来。3.2 证书信任和常见失败原因抓包最坑的一步是 HTTPS 证书。新版本的安卓系统默认不信任用户安装的证书小程序做 HTTPS 请求时如果用了系统的证书校验你把抓包工具证书装进模拟器里也没用。解决办法一般是把抓包工具的 CA 证书作为“系统证书”导入比较绕但网上资料多。这就导致了一个常见局面新手在这步卡一晚上证书来回装最终抓到的还是一堆CONNECT连接请求看不到明文数据。我个人习惯的排查顺序是确认模拟器和电脑能互通代理 IP 没填错。确认代理工具已开启 HTTPS 解密并且证书已安装到模拟器系统信任区。关闭微信或重启模拟器重新打开小程序触发一次新请求。看抓包工具里是否有TCP SSL Error之类的报错有就查证书没有就查代理配置。3.3 从一堆请求里挑出真正的数据接口微信小程序打开一个页面经常会在瞬间发起好几十个请求上报日志、拉配置、检查版本、预加载数据、渲染组件……如果一股脑全看很容易迷失。我一般按三步来过滤第一步按域名过滤。把拆包阶段找到的主接口域名输进过滤框只保留来自这个域名的请求。第二步按响应类型过滤。数据接口返回的通常是 JSON日志和静态资源一般返回的是图片、字体、二进制。找到响应体是 JSON 的那几个请求。第三步看请求参数和时间线。正常情况下列表数据请求和页面渲染是同步发生的。你一边在界面上滑动页面一边观察抓包工具多出来的请求往往就是列表翻页请求它的参数里一定会有page或pageSize这类字段。判断哪个接口是“正主”还有一个技巧直接看接口名字。小程序后端接口命名通常很直白比如/api/goods/list、/api/article/detail、/api/user/center基本不用猜。3.4 一个小细节接口数据未必全在 JSON 里现在有些小程序为了省流量会在 JSON 里塞encodeURIComponent过的字符串或者直接用 base64 包一层。你在抓包工具里看到一堆看起来很像乱码的字符串先别头大把它复制出来解一下。常见的几种“假乱码”base64 编码特征是一串字母数字加结尾。URL 编码特征是%开头。用小程序的加密算法封装过的字段特征是字段名里带encrypt或cipher。前两种直接解码就行第三种就需要回头读 JS 源码看它到底走的是什么加密流程。这里先留个印象后面专门讲模拟请求时再说。4. 模拟请求封装参数、签名跟上用代码代替手工点拆包和抓包都是“观察”真正的采集工作要从“模拟请求”开始。小程序接口的防护通常分几层登录态、请求头校验、参数签名、返回内容加密。我们一个一个说。4.1 找到登录态和基础 Header小程序请求基本都会带登录凭证。常用做法是请求头带Authorization: Bearer token请求头带Cookie自定义 Header 带token或ticket。token 一般是通过微信登录流程换来的通常是先拿微信端返回的 code再请求后端换取登录态。自动化场景下最简单的办法是从抓包工具里直接抄一份有效 token 出来先用。但 token 会过期尤其一些小程序把有效时间设成 30 分钟甚至更短所以你后面必须把它做成自动续期不能写死。另外很多服务端会校验User-Agent不是普通浏览器的 UA而是微信小程序的 UA。这种 UA 里一般带着MicroMessenger字样。建议直接从抓包工具里复制真实 UA不要自己编。4.2 签名参数怎么找、怎么还原最让人头疼的是签名。现在随便一个稍微正规点的小程序接口都会带sign、nonce、timestamp这类字段。如果缺少或算错服务端直接返回参数错误。找签名算法的唯一可靠路径还是回到 JS 源码里搜。常见的关键字有sign signature nonce timestamp hmac sha256 md5 RSA AES搜到相关函数后在 JS 里往前找它的输入参数。一般签名逻辑是把业务参数 一个固定或动态的 key 拼起来再经过哈希算法得到结果。举个简化示例假设你在源码里看到这样的逻辑function buildSign(params, secretKey) { let keys Object.keys(params).sort(); let raw keys.map(k k params[k]).join(); return md5(raw key secretKey); }那么你在 Python 里要模拟的请求生成逻辑就是import hashlib def build_sign(params: dict, secret_key: str) - str: raw .join( f{k}{params[k]} for k in sorted(params) if params[k] is not None ) return hashlib.md5((raw key secret_key).encode(utf-8)).hexdigest()这里有一点要注意很多源码里会先把空值、undefined、空字符串过滤掉再排序拼接。你照着写的时候也要把过滤逻辑带上否则签名永远对不上。4.3 构造一个能跑通的最小请求假设我们已经通过抓包确定了接口地址https://api.example-domain.com/api/goods/list请求方式POST请求体{page: 1, pageSize: 20, categoryId: 101}HeaderContent-Type: application/json、token、User-Agent那么最小请求代码就是这样import requests import hashlib import json BASE_URL https://api.example-domain.com/api/goods/list def make_request(page: int, token: str, secret_key: str): payload { page: page, pageSize: 20, categoryId: 101, timestamp: 1700000000, nonce: abc123 } payload[sign] build_sign(payload, secret_key) headers { Content-Type: application/json, User-Agent: Mozilla/5.0 MicroMessenger/xxx, token: token, } resp requests.post(BASE_URL, jsonpayload, headersheaders, timeout10) return resp.json()跑通之后先去对比返回数据和抓包工具里看到的数据结构确认完全一致再继续往下封装。4.4 返回内容加密怎么办前面提到过有些接口的响应体是密文。一般这种加密用对称加密算法密钥大概率是从代码包里泄露的或者由另一个接口下发。在 JS 源码里搜decrypt、AES、CryptoJS通常能找到解密函数。常见的做法是 AES-128-CBC密钥固定偏移量固定或者也被藏在前端代码里。Python 端可以用pycryptodome来对应实现from Crypto.Cipher import AES from Crypto.Util.Padding import unpad import base64 def decrypt_response(data: str, key: bytes, iv: bytes) - str: cipher AES.new(key, AES.MODE_CBC, iv) raw base64.b64decode(data) plain unpad(cipher.decrypt(raw), AES.block_size) return plain.decode(utf-8)说实话每次遇到这种加密字段我都建议你回头再把源码里的解密逻辑完完整整读一遍特别注意它会不会对密文做二次处理比如先 base64 再 hex。这东西直接套官方函数很容易翻车还是照着源码逐行换算靠谱。4.5 模拟请求阶段最常见的四个坑签名参数排序不一致小程序端有些会按 ASCII 排序有些会按参数名称的 Unicode 码排序还有些会把公共参数单独放前面。你没看源码靠猜基本猜不对。timestamp 校验服务端会校验请求时间戳和服务器时间的差值超过几分钟直接拒绝。写代码时用int(time.time())动态生成时间戳不要用一个固定值。nonce 重复校验同一个随机串短时间内使用两次服务端会视为重放攻击。你要在自己代码里每次都换一个新串。headers 里的动态字段除了 token还可能有requestId、deviceId、appVersion这类字段。这些字段的生成逻辑也在源码里搜一下就能定位。5. 落地一套最小采集框架多线程 SQLite 重试模拟请求跑通后下一步是把它整理成一个能长期稳定运行的采集脚本。我建议不要只写一个单文件后面加监控、加数据库、改字段都麻烦。用一个简单但清晰的结构。5.1 目录划分我自己常用的小工程结构大概是collector/ ├── main.py # 入口 ├── api_client.py # 封装请求逻辑 ├── storage.py # 数据库操作 ├── config.py # 参数配置 └── logs/如果你需要多平台采集可以把api_client.py拆成多个模块一个平台一个实现。但目录别过度设计收采一个小程序时这四件套完全够用。5.2 核心代码实现api_client.py里封装一个类import time import random import requests import threading class ApiClient: def __init__(self, token, secret_key, max_retries3): self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 MicroMessenger/xxx, token: token, }) self.secret_key secret_key self.max_retries max_retries def fetch_page(self, page, category_id): payload { page: page, pageSize: 20, categoryId: category_id, timestamp: int(time.time()), nonce: str(random.randint(100000, 999999)), } payload[sign] build_sign(payload, self.secret_key) for retry in range(self.max_retries): try: resp self.session.post( /api/goods/list, jsonpayload, timeout10, ) data resp.json() if data.get(code) in (0, 200): return data[data] # 风控或 token 过期的情况 if data.get(code) 401: self._refresh_token() continue if data.get(code) 429: time.sleep(2 retry * 2) continue except requests.RequestException: time.sleep(2) return None这里_refresh_token()要按小型自动续期通常就是拿旧的 token 去换新的。如果实在换不了就先写一个通知让后续流程暂停等人工处理。storage.py里用 SQLite 存数据import sqlite3 import json class Storage: def __init__(self, db_pathdata.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.lock threading.Lock() self._create_table() def _create_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS goods ( id INTEGER PRIMARY KEY, goods_id TEXT, category_id TEXT, data TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) def insert_goods(self, goods_id, category_id, data_dict): with self.lock: self.conn.execute( INSERT OR REPLACE INTO goods (goods_id, category_id, data) VALUES (?, ?, ?), (goods_id, category_id, json.dumps(data_dict, ensure_asciiFalse)) ) self.conn.commit()main.py里做多线程调度import threading from api_client import ApiClient from storage import Storage def worker(category_id, client, storage, pages): for page in range(1, pages 1): data client.fetch_page(page, category_id) if not data or not data.get(list): break for item in data[list]: storage.insert_goods(item[id], category_id, item) print(fcategory {category_id} done) def main(): storage Storage() client ApiClient(tokenxxx, secret_keyxxx) threads [] for category_id in [101, 102, 103]: t threading.Thread(targetworker, args(category_id, client, storage, 50)) t.start() threads.append(t) for t in threads: t.join() if __name__ __main__: main()这个框架虽然简单但已经具备抓取、去重、并发、错误重试、人工告警入口。实际跑一个小程序基本够用。5.3 落地之后大概率会遇到的现实问题页面数判断错误很多列表接口不是直接把总页数打给你而是返回一个hasMore或hasNext字段。你判断终止条件时优先看这个字段不要硬编码“假设有 50 页”。并发数别拍脑袋设小程序的接口大多有限流。我之前试过开 20 个线程结果没跑多久就被风控了后来乖乖降到 3 到 5 个线程加了随机延时才稳定下来。数据库表设计别只存现字段小程序接口的字段可能隔三差五变一下。把整条 JSON 存下来之后字段变了也不怕重新解析原始数据就行。这是我现在非常推荐的做法。6. 采集中一定要守住的边界写爬虫的人如果不谈边界很容易把自己和项目方都坑进去。很多数据虽然技术上能抓到但能不能抓、抓了怎么用完全是另一回事。我在这方面的原则很简单不去碰不该碰的数据不拿采集数据做不该做的事。6.1 哪些内容可以在合理范围内采集从合规和风险角度出发可以采集的一般是那些面向所有用户公开、不需要登录就能看到的内容比如公开的商品介绍、公开的活动信息、公开的列表数据。这类数据类似“公共场所可见的信息”技术上去采集法律和道德风险相对低。但有几个雷区要特别注意用户隐私信息手机号、微信号、头像、性别、地理位置等一律不要碰。就算接口里返回了也不建议落到自己库里。需要登录态才能看到的“个人数据”比如我的订单、我的收藏、我的余额。这类不是公开数据采集行为有很大争议我强烈不建议做。平台明确禁止商业爬取的内容许多小程序的服务条款里明确写了禁止未经授权抓取数据。明知故犯后续风险太高。6.2 频率控制和技术红线技术上有能力不等于可以随便跑。我做一个项目时会把单接口的请求频率压到每秒 1 次以下批处理时间尽量放在深夜或业务低峰期。宁可慢一点也要减少对目标服务的压力。如果有些接口对单 IP 限流特别严格就放弃“高并发”方案用单线程加延时慢慢排。代码层面建议加一个简单的“频率护栏”import time class RateLimiter: def __init__(self, interval1.0): self.interval interval self.last_time 0.0 def wait(self): now time.time() delta now - self.last_time if delta self.interval: time.sleep(self.interval - delta) self.last_time time.time()把它加进fetch_page的逻辑里每次请求前强制等一拍省得哪天手一抖把自己的 IP 送进黑名单。6.3 数据保存和使用的底线采集到的数据存多久、给谁用、怎么用这些也要想清楚。我现在的做法是能不用真实 ID 的不记真实 ID只留业务标识。数据库里分“原始数据”和“清洗数据”两张表保留时间按需求设定到期自动清理。绝不把采集到的数据包装成所谓“数据集”转卖给别人这是最容易踩到法律红线的一条。另外还有一点经常被忽略采集脚本的日志里可能会带 token、签名 key 这些敏感信息。项目文件一旦外泄等于把账户的钥匙也交出去了。所以我建议把 token、密钥、数据库连接串全部丢进环境变量或独立配置文件不要写死在代码里。6.4 最后一句话留给后来人如果你只是做一个内部数据观察项目把目标控制在小范围、低频、公开数据这三个界定的交集里风险是相对可控的。但如果你的主要目的已经超出了这个边界那不管技术方案多华丽我都劝你及时停下来。能爬和该爬从来不是同一件事。我自己在这个领域踩过的坑不算少拆包拆到凌晨、签名算错到怀疑人生、风控封号后整库重来……但把这些链路理顺之后小程序采集其实也就是“拆包找接口、抓包验参数、代码模拟请求、框架去稳定运行”这四步。希望这篇能让你少走几个我走过的弯路。
觉得有用,分享给同行:

为您的企业打造数字门面

稳重轻奢商务风格,端正雅致视觉,长效耐看不易过时。

立即咨询 →