简介这份资源是一套面向开发者与数据分析人员的微信小程序爬虫实战项目聚焦于抓取小程序中的商品列表数据并支持写入数据库适合具备一定Python基础、希望了解小程序数据采集流程的读者参考学习。压缩包共20个文件约1.56MB以py脚本为主体配合sql建表语句、xml与json配置、png/jpeg示意图及md说明文档覆盖爬虫主逻辑、数据清洗、数据库写入与项目说明等模块目录结构清晰便于按功能定位代码。目前已有103人学习下载。项目围绕模拟用户操作、调用接口获取商品列表、数据格式化后写入SQLite等数据库的完整链路展开并预留了定时爬取与异常处理的扩展空间读者可借此理解小程序数据采集的基本思路、数据库落库方式以及代码组织与维护要点为后续的数据分析与市场监控场景提供可复用的参考方案。1. 从一份小程序商品采集包说起它到底能跑通什么手上这份「微信小程序爬虫爬取商品列表。支持写入数据库」的资源包第一次看到标题时我以为是那种跑两天就挂的玩具脚本。解压后翻了一遍代码结构发现它把采集链路里最容易翻车的几个环节——请求签名、分页游标、字段清洗、入库去重——都做了最小闭环。换句话说它不是教你「怎么发一个 HTTP 请求」的演示而是一份能直接改改配置就上手的商品列表采集骨架。它适合两类人一类是手里有具体小程序商品页、需要把 SKU 名称、价格、库存、销量这些字段定期落库做比价或选品的运营开发另一类是想学小程序接口逆向思路但不想从零啃加密逻辑的后端工程师。前提是你得接受一个现实小程序采集没有浏览器里那种「右键查看源码」的便利所有数据都藏在接口返回的 JSON 里而接口往往带着签名和时效 token。这份资源的价值就在于它把签名生成和 token 刷新这两块黑匣子拆开给你看了。2. 拆开采集链路请求构造、签名与分页游标2.1 小程序接口和普通网页采集的本质差异普通网页采集你拿到 URL 直接 requests.get 就能回 HTML解析靠 XPath 或 CSS 选择器。小程序不一样它的页面是前端框架渲染出来的真实数据走的是 wx.request 调用的后端接口。这些接口通常有三个特征一是域名往往和主站不同常见的是servicewechat.com这类专用域名二是请求头里带Referer值必须包含小程序的 appid 路径否则服务端直接返回 403三是关键参数会被签名签名算法可能藏在混淆过的 JS 里。这份资源里处理请求构造的部分核心就是还原了这三件事。它没有用 Selenium 或 Playwright 去模拟渲染而是直接抓接口这样速度快、资源占用低但代价是你得自己维护签名逻辑。我翻了下代码签名部分用的是常见的md5(参数排序拼接 固定盐)模式盐值硬编码在配置里。这种做法的好处是简单直接坏处是如果对方换了盐你得手动更新。2.2 请求构造的代码骨架与参数说明下面这段是资源里请求构造的核心逻辑我做了注释补全方便你对照自己的目标小程序改import hashlib import time import requests # 配置区这些值需要你从目标小程序的抓包结果里替换 APPID wx0000000000000000 # 小程序 appid抓包时在 Referer 里能看到 BASE_URL https://servicewechat.com/wxa-dev-logic/xxx # 接口基础地址 SALT xxxxxxxx # 签名盐值从混淆 JS 里抠出来的 def build_signed_params(params: dict) - dict: 对参数做排序拼接后加盐 md5生成签名 # 1. 按 key 的 ASCII 码升序排列 sorted_keys sorted(params.keys()) # 2. 拼接成 k1v1k2v2 的形式过滤空值 raw .join(f{k}{params[k]} for k in sorted_keys if params[k] is not None) # 3. 末尾追加盐值后做 md5 sign hashlib.md5((raw SALT).encode(utf-8)).hexdigest() params[sign] sign params[timestamp] int(time.time() * 1000) # 毫秒时间戳部分接口会校验时效 return params def fetch_page(page: int, page_size: int 20) - dict: 请求单页商品列表 params { page: page, pageSize: page_size, categoryId: , # 分类筛选留空表示全部 sortType: default, # 排序方式常见值有 default/sales/price } params build_signed_params(params) headers { Referer: fhttps://servicewechat.com/{APPID}/0/page-frame.html, User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148, Content-Type: application/json, } resp requests.get(BASE_URL /goods/list, paramsparams, headersheaders, timeout10) resp.raise_for_status() return resp.json()逻辑说明build_signed_params做了三件事——排序、拼接、加盐哈希。参数说明里最关键的是SALT和APPID这两个值必须从你自己的抓包结果里取资源里给的是占位符。timestamp用毫秒是因为多数小程序接口用Date.now()生成服务端会校验和当前时间差不超过几分钟。Referer里的/0/page-frame.html是微信小程序的固定路径格式appid 不对会直接 403。2.3 分页游标offset 和 cursor 两种模式怎么判断分页是采集里第二个容易翻车的点。这份资源默认用的是page/pageSize的 offset 模式但我在实际跑的时候发现不少小程序商品接口用的是 cursor 模式——返回里带一个nextCursor字段下次请求把它原样传回去。两种模式的判断方法很简单看第一页返回的 JSON 里有没有nextCursor或hasMore字段。如果有nextCursor就把page参数换成cursor循环终止条件从「返回条数小于 pageSize」改成「nextCursor 为空」。资源里留了一个pagination_mode配置项默认是offset你改成cursor后需要同步调整循环逻辑。这个设计不算优雅但胜在直白改起来不用动主流程。3. 字段清洗与入库从 JSON 到 MySQL 的完整落地3.1 商品字段的常见坑价格单位、库存状态、SKU 嵌套接口返回的 JSON 看着干净实际入库前有一堆脏活。我拿几个典型字段举例价格字段有的接口返回「分」有的返回「元」还有的返回带货币符号的字符串直接存 DECIMAL 会报错库存字段可能是数字也可能是{status: in_stock, count: 100}这种嵌套结构SKU 信息经常是数组一个商品对应多个规格需要拆成子表或者用 JSON 字段存。资源里的清洗逻辑写在一个clean.py里核心思路是「先归一化再入库」。归一化包括价格统一转成「分」的整数、库存统一取count字段、SKU 数组序列化成 JSON 字符串。我建议你在清洗层加一个字段映射表把接口字段名和数据库列名对应起来这样换一个小程序只需要改映射表不用动清洗代码。3.2 建表 SQL 与入库去重策略下面是资源里配套的建表语句我加了索引和唯一约束方便你做去重CREATE TABLE goods ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, goods_id VARCHAR(64) NOT NULL COMMENT 商品唯一ID来自接口, title VARCHAR(255) NOT NULL DEFAULT COMMENT 商品标题, price_cent INT NOT NULL DEFAULT 0 COMMENT 价格单位分, stock INT NOT NULL DEFAULT 0 COMMENT 库存数量, sales INT NOT NULL DEFAULT 0 COMMENT 销量, sku_json JSON DEFAULT NULL COMMENT SKU 规格数组, category_id VARCHAR(64) DEFAULT COMMENT 分类ID, raw_json JSON DEFAULT NULL COMMENT 原始返回排查用, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_goods_id (goods_id), KEY idx_category (category_id), KEY idx_updated (updated_at) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT商品列表采集表;逻辑说明goods_id加唯一索引是去重的关键入库时用INSERT ... ON DUPLICATE KEY UPDATE而不是先查后插这样能避免并发下的重复写入。raw_json字段建议保留排查字段解析错误时直接看原始返回比翻日志快。price_cent用整数存分避免浮点精度问题展示时再除以 100。入库部分的代码资源里用的是pymysql加批量提交每 100 条 commit 一次。这个批量大小可以调我一般设 200 到 500太小了 IO 次数多太大了内存占用高且失败重试成本大。3.3 增量采集用 updated_at 做时间窗口过滤全量采集跑一次就够了日常维护靠增量。资源里留了一个since参数传给接口后只返回某个时间之后更新的商品。但要注意不是所有小程序接口都支持时间过滤如果不支持你只能在入库层做比对——把新抓到的goods_id和库里的做 diff只更新有变化的记录。这种做法全量拉取的成本高但胜在通用。我一般会先试接口的since参数能用就用不能用再退回 diff 方案。diff 方案的关键是给updated_at加索引比对时只查最近 N 天的记录不要全表扫。4. 避坑与排查那些让我重跑过三次的坑4.1 坑一签名过了但返回 403Referer 里 appid 写错现象请求返回 403body 里没有任何有用信息。原因Referer头里的 appid 和实际请求的小程序不匹配服务端校验失败。解决抓包时看请求头里的Referer完整值把 appid 和路径原样复制不要凭记忆写。我在这上面浪费过半小时后来养成习惯——所有请求头直接从抓包工具里导出成字典不手打。4.2 坑二分页到第 N 页后返回空列表但总数还没抓完现象前几页正常到某一页后返回空数组但接口返回的total显示还有数据。原因offset 模式在数据动态变化时会出现「漂移」比如你抓第 3 页时前面有商品下架导致第 3 页的数据被挤到第 2 页第 3 页就空了。解决改用 cursor 模式或者按goods_id排序后做范围查询。如果只能用 offset就在空列表时停一下隔几秒重试一次还空就跳过。4.3 坑三价格字段存进去变成 0清洗时类型判断漏了字符串现象数据库里price_cent全是 0。原因接口返回的价格是字符串19.90清洗代码里用了int(price)直接转换抛异常后被 except 吞掉赋了默认值 0。解决清洗时先判断类型字符串先float再乘 100 取整同时把异常打日志而不是静默吞掉。这个坑的教训是——except 块里至少要logging.warning不然出了问题你连线索都没有。4.4 坑四批量入库时部分失败导致整批回滚现象一批 200 条数据其中一条字段超长整批插入失败。原因MySQL 的INSERT默认是单条失败整批报错没有做逐条容错。解决要么把批量大小调小要么在入库前做字段长度校验把超长字段截断并记录。我现在的做法是入库前跑一遍validate函数检查标题长度、价格范围、必填字段不合格的单独存到failed_records表不阻塞主流程。4.5 坑五跑了一段时间后接口返回频率限制现象连续请求几百次后接口开始返回 429 或空数据。原因触发了服务端的频率限制可能是按 IP 也可能是按 appid。解决在请求之间加随机延时资源里默认是time.sleep(0.5)我一般改成random.uniform(0.8, 1.5)降低规律性。如果还是被限就换请求头里的User-Agent和设备标识但不要用代理池——那属于另一个层面的操作这份资源不涉及。5. 进阶技巧把采集包改造成可调度的定时任务资源本身是一个「跑一次就退出」的脚本但实际用起来你肯定希望它定时跑。我一般会做三件事把配置抽到config.yaml、加日志轮转、用APScheduler做进程内调度。下面是一个最小改造示例import yaml import logging from logging.handlers import RotatingFileHandler from apscheduler.schedulers.blocking import BlockingScheduler # 日志配置单文件 10MB保留 5 个备份 handler RotatingFileHandler(spider.log, maxBytes10*1024*1024, backupCount5) logging.basicConfig(levellogging.INFO, handlers[handler], format%(asctime)s %(levelname)s %(message)s) def load_config(pathconfig.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def job(): cfg load_config() logging.info(采集任务开始目标 appid%s, cfg[appid]) # 这里调用第 2 章的 fetch_page 和第 3 章的入库逻辑 # ... if __name__ __main__: scheduler BlockingScheduler() # 每 6 小时跑一次避开整点高峰 scheduler.add_job(job, interval, hours6, jitter300) scheduler.start()逻辑说明RotatingFileHandler防止日志文件无限增长jitter300让任务在 6 小时基础上随机偏移最多 5 分钟避免和其他定时任务撞在一起。config.yaml里放 appid、salt、数据库连接串这些易变项改配置不用动代码。验证采集是否正常我习惯看三个指标单次采集条数、入库成功条数、updated_at最新时间。如果采集条数骤降先查接口是否改了签名如果入库成功条数远小于采集条数查字段校验日志如果updated_at不更新查调度进程是否还活着。从那以后我每次改完采集逻辑都会先用pageSize5跑一轮小批量确认字段和入库都正常再放开全量。这个习惯帮我省过好几次回滚的麻烦。希望帮到你。本文还有配套的精品资源点击获取