简介这是一份微信小程序商品列表爬虫的完整实现面向具备一定Python基础、需要采集小程序数据的开发者和数据分析人员。源码基于微信官方接口模拟用户交互在合规范围内抓取商品信息并内置写入MySQL、MongoDB、SQLite等数据库的适配模块同时包含数据清洗与格式化处理便于后续统计和检索。资源压缩包共20个文件、约1.56MB以Python脚本和SQL文件为核心辅以XML配置、项目说明文档、运行截图等目录结构清晰可对照文档快速理解并二次开发。已有98人学习下载代码注释较为完整具备良好的可扩展性便于加入定时爬取、异常管理及数据可视化等功能。借助这套爬虫使用者能快速搭建合规的小程序数据采集流程为电商选品、市场监控和用户行为分析提供数据支持。1. 项目概述与整体思路1.1 这个项目到底解决什么问题平时做电商数据采集或者市场调研的时候最头疼的就是数据来源。网页端做了大量的反爬限制头部电商平台的PC页面各种滑块、字体加密、行为验证码轮番上阵爬取效率低还容易被封。但是不少人忽略了一个入口——微信小程序端。小程序端的接口往往比网页端好拿得多原因也很简单小程序运行在微信这个封闭容器里开发者普遍觉得微信的审核机制和安全机制已经够用所以在接口防护上投入的精力远没有网页端那么多。这就给了爬虫开发者一个相对宽松的采集环境。我拿到的这个压缩包项目核心目标就是通过微信小程序端抓取商品列表数据解析后写入数据库。整个项目结构不算复杂但思路比较完整——从抓包到数据解析再到落库是一条完整的链路。对于想入门小程序数据采集、或者需要批量获取商品信息做分析的人来说是一个可以改改就能用的基础框架。1.2 项目适合谁看需要什么前置知识如果你具备以下基础上手这个项目会非常丝滑Python基础语法、requests库的使用经验、基本的SQL语句、以及一点HTTP协议的知识。群里有人问我是不是必须会小程序开发其实不太需要——你不需要会写小程序的wxml和wxss你只需要会看它的网络请求就够了。前置工具方面需要准备一个安卓模拟器或者真机、Charles或者Fiddler抓包工具、Python 3.7以上环境以及MySQL或者SQLite数据库。Windows和macOS都行我测试用的环境是Windows 10 Python 3.9 MySQL 8.0后面所有步骤都是在这个环境跑的。1.3 方案选型为什么走小程序端这条路其实最开始我琢磨的方案是直接抓网页版商品列表但试了一圈发现阻力很大字体反爬、CSS偏移、JS加密一环扣一环破解成本太高。后来换了个思路转向小程序端效果反而好得多。小程序端的请求结构普遍简单。大多数情况下就是HTTPS请求加上一个签名参数而这个签名算法有时就藏在js里用工具反编译就能看到。你在微信里打开一个商品小程序正常浏览列表页抓包看到的请求大概率是标准的JSON返回只需要把返回的json字段解析出来就行不需要对抗什么字体加密。这就是所谓“降维打击”——与其跟网页端斗智斗勇不如找一个防护薄弱的入口。2. 微信小程序抓包与请求解析2.1 抓包环境搭建与核心操作这一步是整个项目的基础抓不到包后面全白搭。我用的是两个方案Android真机配合Fiddler以及Android模拟器配合Charles。先说真机方案。手机和电脑连同一个WiFi在Fiddler里开启HTTPS解密Tools - Options - HTTPS - Decrypt HTTPS traffic然后手机上设置代理为电脑的IP地址加Fiddler端口8888再安装Fiddler根证书就可以看到小程序发出的所有HTTPS请求了。实际操作中有一个细节容易踩坑——不少小程序会做证书校验直接安装抓包工具的根证书会被对方检测并拒绝连接。解决办法是使用Frida框架来hook掉证书校验逻辑或者用Xposed模块绕过检测。但这部分门槛较高需要了解逆向基础。如果你只是测试环境最省事的方法是用安卓模拟器。模拟器自带root权限可以使用类似“小黄鸟HttpCanary”这类抓包工具直接抓取也可以在模拟器系统里把证书安装到系统证书目录很多校验就能绕过去。我自己测试时用的是MuMu模拟器 HttpCanary实测下来成功率很高。2.2 定位商品列表请求抓包解决了下一个问题是怎么从一堆请求里挑出真正的商品列表接口。小程序启动后会发很多请求初始化配置、用户信息、首页banner、商品分类、购物车数量等等。刚开始抓的时候可能会看花眼我总结了一个快速定位的方法优先找返回JSON体积大、字段结构复杂且包含商品关键词skuId、goodsId、price、stock等的请求列表页下拉刷新时会重新发出的请求基本就是列表接口——你可以操作一次下拉刷新手动触发请求这样在抓包工具的对比视图里更容易定位查看请求URL路径常见的命名规律如/api/goods/list、/api/search、/api/shop/product等一眼能认出是列表接口请求参数里通常会带page、pageSize、sort、categoryId这类分页排序参数特征非常明显。定位到请求之后右键复制为cURL格式然后转成Python requests代码网上有在线转换工具也可以手动改写。这一步之后你就拿到了一个能跑通的数据来源入口。2.3 请求签名与加密参数处理很多大平台的小程序接口不会让你直接白嫖请求通常在header里会携带sign、nonce、timestamp、appKey之类的参数。服务端通过校验sign来确认请求来自合法的客户端而不是你随手用Python模拟的。这个sign参数怎么来最常见的两种方式 一是小程序前端JS计算通过反编译工具提取算法在Python里复现 二是sign直接从云函数或后端接口下发这种情况就不需要你自己算。我处理过的项目里签名算法一般就是MD5、SHA256、HmacSHA256组合拼接比如把业务参数按字典序排序、拼接固定盐值、再取哈希。你需要用反编译工具提取小程序代码后定位这个函数。压缩包里应该也包含了一套完整的小程序解包工具链——主要是node.js环境下的wxappUnpacker用它可以把小程序wxapkg包还原成接近源码的形式。拿到签名算法后在Python里照着实现一遍注意字符串编码、大小写、时间戳的选取要和原来一致。这里最容易翻车的就是盐值写错或拼接顺序不对调试的时候先把抓包拿到的原始请求参数和签名原样重现一遍确认能请求成功再加变量替换。3. 商品列表爬虫代码实现与入库3.1 数据解析从JSON响应到结构化数据成功请求到接口后返回的JSON结构每家略有差异但大体的规律是一致的。拿我抓的一个商品列表举例响应的data字段里有个list数组每个元素对应一个商品对象包含以下核心字段{ code: 0, message: success, data: { list: [ { skuId: 10001293842, title: 某某品牌运动鞋 2024新款, price: 299.00, originalPrice: 499.00, sales: 1200, shopName: 官方旗舰店, category: 运动鞋, stock: 532, coverImage: https://xxx.com/img/123.jpg } ], hasMore: true, page: 1, totalCount: 35682 } }解析的时候有一个建议——不要光盯着title和price这两个字段多解析一些辅助字段。比如sales销量可以用于后续排行分析stock库存判断商品是否下架originalPrice可以算折扣力度。这些字段对后续数据分析很重要建议全部入库存着哪怕暂时用不上也不要丢弃。解析用的库就是Python自带的json模块或者用requests的response.json()方法直转字典。字段取值用dict.get(key, default)比dict[key]更安全避免个别商品缺少某个字段时直接KeyError崩溃。3.2 爬虫主流程设计多页循环与异常处理商品列表的基本抓取逻辑很简单——循环翻页直到没有更多数据为止。但翻页循环里要处理的细节不少我直接给出优化过的核心代码import requests import hashlib import time import random import json class MiniProgramSpider: def __init__(self, api_url, app_key, secret_key): self.api_url api_url self.app_key app_key self.secret_key secret_key self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15, Referer: https://servicewechat.com/wxrandomappid1b2c3d4e5f6g7h8/, Content-Type: application/json }) def generate_sign(self, params): # 常见签名方式参数排序拼接 盐值 MD5 sorted_keys sorted(params.keys()) raw_string for key in sorted_keys: if params[key] ! : raw_string f{key}{params.get(key)} raw_string fkey{self.secret_key} return hashlib.md5(raw_string.encode(utf-8)).hexdigest().upper() def fetch_goods_list(self, page, page_size20): params { page: page, pageSize: page_size, sort: default, timestamp: int(time.time()) } sign self.generate_sign(params) params[sign] sign resp self.session.get(self.api_url, paramsparams) return resp.json() def crawl(self, max_pages50): all_goods [] page 1 while page max_pages: try: data self.fetch_goods_list(page) goods_list data.get(data, {}).get(list, []) has_more data.get(data, {}).get(hasMore, False) if not goods_list: break all_goods.extend(goods_list) print(f页面 {page} 抓取成功获取 {len(goods_list)} 条数据) if not has_more: break page 1 # 随机延时降低请求频率避免触发风控 time.sleep(random.uniform(1.5, 3.5)) except Exception as e: print(f页面 {page} 抓取失败: {e}) # 失败重试机制连续失败3次则跳过 retry_count getattr(self, _retry_count, 0) retry_count 1 if retry_count 3: print(连续失败3次跳过当前页) retry_count 0 page 1 self._retry_count retry_count time.sleep(5) return all_goods这里有几个点值得说下。第一是请求头里的Referer和小程序userAgent。有些接口会校验来源把Referer设置成servicewechat.com域名下的路径可以绕过一部分来源限制。User-Agent用iPhone的微信内置浏览器UA不容易触发风控比裸Python的UA可靠得多。第二是时间戳。这类接口的时间戳通常要求和服务端时间差在几百秒内用time.time()生成的没问题但如果你在本机测试且机器时间偏差过大会直接提示签名错误。务必保证系统时间准确。第三是失败重试的写法。很多新手一遇到请求异常就死循环或者直接崩溃退出但爬虫要做的是记录日志、重试、跳过异常数据保证整体流程能走完。我用的这个简单计数方式只是基础版更稳健的做法是记录fail_pages列表结束后统一重试。3.3 数据入库MySQL表设计与批量写入数据拿到手了但内存里的列表随时可能丢失必须尽快落库。我用的是MySQL表结构设计如下CREATE TABLE goods_list ( id INT NOT NULL AUTO_INCREMENT, sku_id VARCHAR(64) NOT NULL COMMENT 商品ID, title VARCHAR(255) NOT NULL COMMENT 商品标题, price DECIMAL(10,2) DEFAULT 0.00 COMMENT 当前价格, original_price DECIMAL(10,2) DEFAULT 0.00 COMMENT 原价, sales INT DEFAULT 0 COMMENT 销量, stock INT DEFAULT 0 COMMENT 库存, shop_name VARCHAR(128) DEFAULT COMMENT 店铺名, category VARCHAR(128) DEFAULT COMMENT 商品分类, cover_image VARCHAR(512) DEFAULT COMMENT 封面图URL, crawled_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT 抓取时间, PRIMARY KEY (id), UNIQUE KEY uk_sku_id (sku_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT小程序商品列表;入库逻辑建议使用pymysql的批量插入效率远高于逐条insert。关键点在主键冲突处理——同一个小程序页面可能会重复抓取商品数据反复入库会导致重复记录。这里用ON DUPLICATE KEY UPDATE配合唯一索引重复skuId时更新价格和销量import pymysql def save_to_mysql(goods_list): conn pymysql.connect( hostlocalhost, userroot, passwordyour_password, databasespider_data, charsetutf8mb4 ) cursor conn.cursor() batch_size 100 sql INSERT INTO goods_list (sku_id, title, price, original_price, sales, stock, shop_name, category, cover_image) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE price VALUES(price), sales VALUES(sales), stock VALUES(stock), crawled_at CURRENT_TIMESTAMP for i in range(0, len(goods_list), batch_size): batch goods_list[i:ibatch_size] rows [] for item in batch: rows.append(( item.get(skuId, ), item.get(title, ), item.get(price, 0), item.get(originalPrice, 0), item.get(sales, 0), item.get(stock, 0), item.get(shopName, ), item.get(category, ), item.get(coverImage, ) )) cursor.executemany(sql, rows) conn.commit() cursor.close() conn.close()连接数据库前确保MySQL服务启动且创建了对应的database。库名、表名、用户名密码改成你自己环境的配置这块写死肯定不适合所有人。建议把数据库配置单独放到config.py或环境变量里后续维护方便。如果你不想装MySQL也可以先用SQLite顶着只要把连接方式换成sqlite3库改动量不大。SQLite适合数据量不大、单机跑的场景后期数据多了再迁MySQL不迟。4. 常见问题与排查技巧实录做这个项目过程中我踩了不少坑也看到群里有人犯过类似错误整理几个高频问题。4.1 问题数据抓回后全是无效乱码或空值表现接口返回200但content字段是乱码或者list为空。排查方向第一步看字符编码。有些接口返回的是gzip压缩过的内容requests会自动解压但如果你直接用底层urllib就未必了。用resp.encoding查看编码如果是乱码尝试resp.encoding utf-8。第二步看是否请求被降级。服务端检测到异常客户端后经常不是拒绝请求而是返回固定空内容。这时候需要比对抓包工具里正常请求和代码请求的headers差异可能是缺少某个自定义header导致。第三步看签名是否正确。把代码里生成的sign值和抓包抓到的sign值对比如果一样但返回空问题不在签名而在其他头参数。4.2 问题触发风控被限制请求表现请求一段时间后开始出现验证码、请求频率限制错误或者直接返回需要登录的状态码。处理方式控制抓取速度。把延时从1秒调整到3到5秒模拟人工浏览速度。你是在做数据采集不是做压测没必要把并发拉满。随机化请求间延时。固定间隔反而更容易被识别为机器用random.uniform做一个浮动范围更自然。请求时间选在低峰期。白天抓大平台容易碰到风控团队实时监控深夜和凌晨的成功率明显更高。设置异常熔断。当连续多次请求失败时停止一段时间比如10分钟再继续而不是无限重试。4.3 问题小程序appid变化导致抓包失效小程序发版后原来的请求接口可能被替换或增加新参数抓包数据全部失效。这种情况没什么一劳永逸的办法只能定期重新抓包对比。我一般会在抓包工具里订阅接口变化一旦url路径出现新增参数就立刻检查并更新代码里的对应逻辑。这个项目作为基础框架你在更换目标小程序时肯定要调整这些配置没法一套代码打天下。4.4 问题价格字段带千分位或字符串类型商品列表返回的price字段偶尔是字符串1,299.00或者1299.00前缀带货币符号。入库前必须统一清洗否则DECIMAL字段插入会报错或者存入错误值。清洗函数很简单def clean_price(price_val): if isinstance(price_val, (int, float)): return float(price_val) if isinstance(price_val, str): cleaned price_val.replace(,, ).replace(¥, ).replace(, ).strip() return float(cleaned or 0) return 0.0这种字段类型不一致问题在真实数据里非常常见永远不要相信接口返回的数据类型跟文档描述完全一致。4.5 问题批量写入时数据库连接断开爬虫跑得时间长了MySQL默认的wait_timeout可能会断开空闲太久的连接。解决方案是每次写入时重新创建连接或者在连接池里加一个ping操作def ensure_connection(conn): try: conn.ping(reconnectTrue) except Exception: conn pymysql.connect(...) return conn写入方法每次调用前都先走一遍这个逻辑能有效避免程序跑一半突然报Lost connection错误。5. 实操后的经验与扩展建议跑通这个项目的流程后我在本机验证了一轮完整的抓取、清洗、入库过程数据量级在3000到5000条时性能没任何问题单页请求延迟在200毫秒以内批量入库100条一次耗时不到1秒。可以说整体效率是达标的。但我也要说清楚一点——这个项目只是数据采集链路的起点往后面走还有几个可以扩展的方向。第一个扩展点是商品详情页抓取。列表页拿到的字段有限如果需要商品的完整规格、描述图片、评价数据还得进入每个商品详情页抓取。这个在列表页拿到skuId之后可以直接拼URL请求原理一致。第二个扩展点是定时增量抓取。简单写一个schedule任务每天固定时间运行一次爬虫只抓取最新上架或价格变动的商品把变更数据单独存到变更日志表里。这个对追踪竞品价格变化很有用。第三个扩展点是数据可视化。数据入库后用Flask或FastAPI做个小服务前端用ECharts展示各分类商品的销量排行、价格分布区间。这块做出来之后爬虫项目就从“采集工具”升级成了“数据产品”对做竞品分析的人来说价值更大。如果你打算把这个框架用到自己关注的小程序上我的建议是先从抓包入手把接口分析透彻再写代码。很多人一上来就开写爬虫结果接口被反爬拦截又回头去调签名反而浪费大量时间。先花半天时间把抓包工具玩明白后面代码部分基本就是复制粘贴改改参数的事。本文还有配套的精品资源点击获取 微信小程序爬虫实战：商品列表抓取、解析与入库
简介这是一份微信小程序商品列表爬虫的完整实现面向具备一定Python基础、需要采集小程序数据的开发者和数据分析人员。源码基于微信官方接口模拟用户交互在合规范围内抓取商品信息并内置写入MySQL、MongoDB、SQLite等数据库的适配模块同时包含数据清洗与格式化处理便于后续统计和检索。资源压缩包共20个文件、约1.56MB以Python脚本和SQL文件为核心辅以XML配置、项目说明文档、运行截图等目录结构清晰可对照文档快速理解并二次开发。已有98人学习下载代码注释较为完整具备良好的可扩展性便于加入定时爬取、异常管理及数据可视化等功能。借助这套爬虫使用者能快速搭建合规的小程序数据采集流程为电商选品、市场监控和用户行为分析提供数据支持。1. 项目概述与整体思路1.1 这个项目到底解决什么问题平时做电商数据采集或者市场调研的时候最头疼的就是数据来源。网页端做了大量的反爬限制头部电商平台的PC页面各种滑块、字体加密、行为验证码轮番上阵爬取效率低还容易被封。但是不少人忽略了一个入口——微信小程序端。小程序端的接口往往比网页端好拿得多原因也很简单小程序运行在微信这个封闭容器里开发者普遍觉得微信的审核机制和安全机制已经够用所以在接口防护上投入的精力远没有网页端那么多。这就给了爬虫开发者一个相对宽松的采集环境。我拿到的这个压缩包项目核心目标就是通过微信小程序端抓取商品列表数据解析后写入数据库。整个项目结构不算复杂但思路比较完整——从抓包到数据解析再到落库是一条完整的链路。对于想入门小程序数据采集、或者需要批量获取商品信息做分析的人来说是一个可以改改就能用的基础框架。1.2 项目适合谁看需要什么前置知识如果你具备以下基础上手这个项目会非常丝滑Python基础语法、requests库的使用经验、基本的SQL语句、以及一点HTTP协议的知识。群里有人问我是不是必须会小程序开发其实不太需要——你不需要会写小程序的wxml和wxss你只需要会看它的网络请求就够了。前置工具方面需要准备一个安卓模拟器或者真机、Charles或者Fiddler抓包工具、Python 3.7以上环境以及MySQL或者SQLite数据库。Windows和macOS都行我测试用的环境是Windows 10 Python 3.9 MySQL 8.0后面所有步骤都是在这个环境跑的。1.3 方案选型为什么走小程序端这条路其实最开始我琢磨的方案是直接抓网页版商品列表但试了一圈发现阻力很大字体反爬、CSS偏移、JS加密一环扣一环破解成本太高。后来换了个思路转向小程序端效果反而好得多。小程序端的请求结构普遍简单。大多数情况下就是HTTPS请求加上一个签名参数而这个签名算法有时就藏在js里用工具反编译就能看到。你在微信里打开一个商品小程序正常浏览列表页抓包看到的请求大概率是标准的JSON返回只需要把返回的json字段解析出来就行不需要对抗什么字体加密。这就是所谓“降维打击”——与其跟网页端斗智斗勇不如找一个防护薄弱的入口。2. 微信小程序抓包与请求解析2.1 抓包环境搭建与核心操作这一步是整个项目的基础抓不到包后面全白搭。我用的是两个方案Android真机配合Fiddler以及Android模拟器配合Charles。先说真机方案。手机和电脑连同一个WiFi在Fiddler里开启HTTPS解密Tools - Options - HTTPS - Decrypt HTTPS traffic然后手机上设置代理为电脑的IP地址加Fiddler端口8888再安装Fiddler根证书就可以看到小程序发出的所有HTTPS请求了。实际操作中有一个细节容易踩坑——不少小程序会做证书校验直接安装抓包工具的根证书会被对方检测并拒绝连接。解决办法是使用Frida框架来hook掉证书校验逻辑或者用Xposed模块绕过检测。但这部分门槛较高需要了解逆向基础。如果你只是测试环境最省事的方法是用安卓模拟器。模拟器自带root权限可以使用类似“小黄鸟HttpCanary”这类抓包工具直接抓取也可以在模拟器系统里把证书安装到系统证书目录很多校验就能绕过去。我自己测试时用的是MuMu模拟器 HttpCanary实测下来成功率很高。2.2 定位商品列表请求抓包解决了下一个问题是怎么从一堆请求里挑出真正的商品列表接口。小程序启动后会发很多请求初始化配置、用户信息、首页banner、商品分类、购物车数量等等。刚开始抓的时候可能会看花眼我总结了一个快速定位的方法优先找返回JSON体积大、字段结构复杂且包含商品关键词skuId、goodsId、price、stock等的请求列表页下拉刷新时会重新发出的请求基本就是列表接口——你可以操作一次下拉刷新手动触发请求这样在抓包工具的对比视图里更容易定位查看请求URL路径常见的命名规律如/api/goods/list、/api/search、/api/shop/product等一眼能认出是列表接口请求参数里通常会带page、pageSize、sort、categoryId这类分页排序参数特征非常明显。定位到请求之后右键复制为cURL格式然后转成Python requests代码网上有在线转换工具也可以手动改写。这一步之后你就拿到了一个能跑通的数据来源入口。2.3 请求签名与加密参数处理很多大平台的小程序接口不会让你直接白嫖请求通常在header里会携带sign、nonce、timestamp、appKey之类的参数。服务端通过校验sign来确认请求来自合法的客户端而不是你随手用Python模拟的。这个sign参数怎么来最常见的两种方式 一是小程序前端JS计算通过反编译工具提取算法在Python里复现 二是sign直接从云函数或后端接口下发这种情况就不需要你自己算。我处理过的项目里签名算法一般就是MD5、SHA256、HmacSHA256组合拼接比如把业务参数按字典序排序、拼接固定盐值、再取哈希。你需要用反编译工具提取小程序代码后定位这个函数。压缩包里应该也包含了一套完整的小程序解包工具链——主要是node.js环境下的wxappUnpacker用它可以把小程序wxapkg包还原成接近源码的形式。拿到签名算法后在Python里照着实现一遍注意字符串编码、大小写、时间戳的选取要和原来一致。这里最容易翻车的就是盐值写错或拼接顺序不对调试的时候先把抓包拿到的原始请求参数和签名原样重现一遍确认能请求成功再加变量替换。3. 商品列表爬虫代码实现与入库3.1 数据解析从JSON响应到结构化数据成功请求到接口后返回的JSON结构每家略有差异但大体的规律是一致的。拿我抓的一个商品列表举例响应的data字段里有个list数组每个元素对应一个商品对象包含以下核心字段{ code: 0, message: success, data: { list: [ { skuId: 10001293842, title: 某某品牌运动鞋 2024新款, price: 299.00, originalPrice: 499.00, sales: 1200, shopName: 官方旗舰店, category: 运动鞋, stock: 532, coverImage: https://xxx.com/img/123.jpg } ], hasMore: true, page: 1, totalCount: 35682 } }解析的时候有一个建议——不要光盯着title和price这两个字段多解析一些辅助字段。比如sales销量可以用于后续排行分析stock库存判断商品是否下架originalPrice可以算折扣力度。这些字段对后续数据分析很重要建议全部入库存着哪怕暂时用不上也不要丢弃。解析用的库就是Python自带的json模块或者用requests的response.json()方法直转字典。字段取值用dict.get(key, default)比dict[key]更安全避免个别商品缺少某个字段时直接KeyError崩溃。3.2 爬虫主流程设计多页循环与异常处理商品列表的基本抓取逻辑很简单——循环翻页直到没有更多数据为止。但翻页循环里要处理的细节不少我直接给出优化过的核心代码import requests import hashlib import time import random import json class MiniProgramSpider: def __init__(self, api_url, app_key, secret_key): self.api_url api_url self.app_key app_key self.secret_key secret_key self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15, Referer: https://servicewechat.com/wxrandomappid1b2c3d4e5f6g7h8/, Content-Type: application/json }) def generate_sign(self, params): # 常见签名方式参数排序拼接 盐值 MD5 sorted_keys sorted(params.keys()) raw_string for key in sorted_keys: if params[key] ! : raw_string f{key}{params.get(key)} raw_string fkey{self.secret_key} return hashlib.md5(raw_string.encode(utf-8)).hexdigest().upper() def fetch_goods_list(self, page, page_size20): params { page: page, pageSize: page_size, sort: default, timestamp: int(time.time()) } sign self.generate_sign(params) params[sign] sign resp self.session.get(self.api_url, paramsparams) return resp.json() def crawl(self, max_pages50): all_goods [] page 1 while page max_pages: try: data self.fetch_goods_list(page) goods_list data.get(data, {}).get(list, []) has_more data.get(data, {}).get(hasMore, False) if not goods_list: break all_goods.extend(goods_list) print(f页面 {page} 抓取成功获取 {len(goods_list)} 条数据) if not has_more: break page 1 # 随机延时降低请求频率避免触发风控 time.sleep(random.uniform(1.5, 3.5)) except Exception as e: print(f页面 {page} 抓取失败: {e}) # 失败重试机制连续失败3次则跳过 retry_count getattr(self, _retry_count, 0) retry_count 1 if retry_count 3: print(连续失败3次跳过当前页) retry_count 0 page 1 self._retry_count retry_count time.sleep(5) return all_goods这里有几个点值得说下。第一是请求头里的Referer和小程序userAgent。有些接口会校验来源把Referer设置成servicewechat.com域名下的路径可以绕过一部分来源限制。User-Agent用iPhone的微信内置浏览器UA不容易触发风控比裸Python的UA可靠得多。第二是时间戳。这类接口的时间戳通常要求和服务端时间差在几百秒内用time.time()生成的没问题但如果你在本机测试且机器时间偏差过大会直接提示签名错误。务必保证系统时间准确。第三是失败重试的写法。很多新手一遇到请求异常就死循环或者直接崩溃退出但爬虫要做的是记录日志、重试、跳过异常数据保证整体流程能走完。我用的这个简单计数方式只是基础版更稳健的做法是记录fail_pages列表结束后统一重试。3.3 数据入库MySQL表设计与批量写入数据拿到手了但内存里的列表随时可能丢失必须尽快落库。我用的是MySQL表结构设计如下CREATE TABLE goods_list ( id INT NOT NULL AUTO_INCREMENT, sku_id VARCHAR(64) NOT NULL COMMENT 商品ID, title VARCHAR(255) NOT NULL COMMENT 商品标题, price DECIMAL(10,2) DEFAULT 0.00 COMMENT 当前价格, original_price DECIMAL(10,2) DEFAULT 0.00 COMMENT 原价, sales INT DEFAULT 0 COMMENT 销量, stock INT DEFAULT 0 COMMENT 库存, shop_name VARCHAR(128) DEFAULT COMMENT 店铺名, category VARCHAR(128) DEFAULT COMMENT 商品分类, cover_image VARCHAR(512) DEFAULT COMMENT 封面图URL, crawled_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT 抓取时间, PRIMARY KEY (id), UNIQUE KEY uk_sku_id (sku_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT小程序商品列表;入库逻辑建议使用pymysql的批量插入效率远高于逐条insert。关键点在主键冲突处理——同一个小程序页面可能会重复抓取商品数据反复入库会导致重复记录。这里用ON DUPLICATE KEY UPDATE配合唯一索引重复skuId时更新价格和销量import pymysql def save_to_mysql(goods_list): conn pymysql.connect( hostlocalhost, userroot, passwordyour_password, databasespider_data, charsetutf8mb4 ) cursor conn.cursor() batch_size 100 sql INSERT INTO goods_list (sku_id, title, price, original_price, sales, stock, shop_name, category, cover_image) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE price VALUES(price), sales VALUES(sales), stock VALUES(stock), crawled_at CURRENT_TIMESTAMP for i in range(0, len(goods_list), batch_size): batch goods_list[i:ibatch_size] rows [] for item in batch: rows.append(( item.get(skuId, ), item.get(title, ), item.get(price, 0), item.get(originalPrice, 0), item.get(sales, 0), item.get(stock, 0), item.get(shopName, ), item.get(category, ), item.get(coverImage, ) )) cursor.executemany(sql, rows) conn.commit() cursor.close() conn.close()连接数据库前确保MySQL服务启动且创建了对应的database。库名、表名、用户名密码改成你自己环境的配置这块写死肯定不适合所有人。建议把数据库配置单独放到config.py或环境变量里后续维护方便。如果你不想装MySQL也可以先用SQLite顶着只要把连接方式换成sqlite3库改动量不大。SQLite适合数据量不大、单机跑的场景后期数据多了再迁MySQL不迟。4. 常见问题与排查技巧实录做这个项目过程中我踩了不少坑也看到群里有人犯过类似错误整理几个高频问题。4.1 问题数据抓回后全是无效乱码或空值表现接口返回200但content字段是乱码或者list为空。排查方向第一步看字符编码。有些接口返回的是gzip压缩过的内容requests会自动解压但如果你直接用底层urllib就未必了。用resp.encoding查看编码如果是乱码尝试resp.encoding utf-8。第二步看是否请求被降级。服务端检测到异常客户端后经常不是拒绝请求而是返回固定空内容。这时候需要比对抓包工具里正常请求和代码请求的headers差异可能是缺少某个自定义header导致。第三步看签名是否正确。把代码里生成的sign值和抓包抓到的sign值对比如果一样但返回空问题不在签名而在其他头参数。4.2 问题触发风控被限制请求表现请求一段时间后开始出现验证码、请求频率限制错误或者直接返回需要登录的状态码。处理方式控制抓取速度。把延时从1秒调整到3到5秒模拟人工浏览速度。你是在做数据采集不是做压测没必要把并发拉满。随机化请求间延时。固定间隔反而更容易被识别为机器用random.uniform做一个浮动范围更自然。请求时间选在低峰期。白天抓大平台容易碰到风控团队实时监控深夜和凌晨的成功率明显更高。设置异常熔断。当连续多次请求失败时停止一段时间比如10分钟再继续而不是无限重试。4.3 问题小程序appid变化导致抓包失效小程序发版后原来的请求接口可能被替换或增加新参数抓包数据全部失效。这种情况没什么一劳永逸的办法只能定期重新抓包对比。我一般会在抓包工具里订阅接口变化一旦url路径出现新增参数就立刻检查并更新代码里的对应逻辑。这个项目作为基础框架你在更换目标小程序时肯定要调整这些配置没法一套代码打天下。4.4 问题价格字段带千分位或字符串类型商品列表返回的price字段偶尔是字符串1,299.00或者1299.00前缀带货币符号。入库前必须统一清洗否则DECIMAL字段插入会报错或者存入错误值。清洗函数很简单def clean_price(price_val): if isinstance(price_val, (int, float)): return float(price_val) if isinstance(price_val, str): cleaned price_val.replace(,, ).replace(¥, ).replace(, ).strip() return float(cleaned or 0) return 0.0这种字段类型不一致问题在真实数据里非常常见永远不要相信接口返回的数据类型跟文档描述完全一致。4.5 问题批量写入时数据库连接断开爬虫跑得时间长了MySQL默认的wait_timeout可能会断开空闲太久的连接。解决方案是每次写入时重新创建连接或者在连接池里加一个ping操作def ensure_connection(conn): try: conn.ping(reconnectTrue) except Exception: conn pymysql.connect(...) return conn写入方法每次调用前都先走一遍这个逻辑能有效避免程序跑一半突然报Lost connection错误。5. 实操后的经验与扩展建议跑通这个项目的流程后我在本机验证了一轮完整的抓取、清洗、入库过程数据量级在3000到5000条时性能没任何问题单页请求延迟在200毫秒以内批量入库100条一次耗时不到1秒。可以说整体效率是达标的。但我也要说清楚一点——这个项目只是数据采集链路的起点往后面走还有几个可以扩展的方向。第一个扩展点是商品详情页抓取。列表页拿到的字段有限如果需要商品的完整规格、描述图片、评价数据还得进入每个商品详情页抓取。这个在列表页拿到skuId之后可以直接拼URL请求原理一致。第二个扩展点是定时增量抓取。简单写一个schedule任务每天固定时间运行一次爬虫只抓取最新上架或价格变动的商品把变更数据单独存到变更日志表里。这个对追踪竞品价格变化很有用。第三个扩展点是数据可视化。数据入库后用Flask或FastAPI做个小服务前端用ECharts展示各分类商品的销量排行、价格分布区间。这块做出来之后爬虫项目就从“采集工具”升级成了“数据产品”对做竞品分析的人来说价值更大。如果你打算把这个框架用到自己关注的小程序上我的建议是先从抓包入手把接口分析透彻再写代码。很多人一上来就开写爬虫结果接口被反爬拦截又回头去调签名反而浪费大量时间。先花半天时间把抓包工具玩明白后面代码部分基本就是复制粘贴改改参数的事。本文还有配套的精品资源点击获取