微信小程序爬虫实战：商品列表采集与数据库入库全链路

简介：这是一份面向Python爬虫初学者与数据分析人员的微信小程序数据采集实战项目，聚焦于模拟用户操作、通过微信官方接口获取小程序商品列表，并将结果持久化到数据库，适合作为爬虫入门到进阶的练手案例。压缩包共20个文件，约1.56MB，以py脚本为核心，配合sql建表语句、xml与json配置、png/jpeg示意图及md说明文档，另含pyc缓存与gitignore等辅助文件，结构清晰便于按模块阅读。项目已吸引103人学习下载，具备一定参考热度。代码中涵盖商品分类、位置数据处理、数据发送至SQLite等模块，并预留MySQL、MongoDB等数据库写入思路，读者可据此掌握数据清洗、格式化与异常处理流程，理解定时爬取与实时分析的扩展方向，同时注意遵守平台规则与法律法规。

1. 小程序商品数据采集：从接口抓包到入库的完整链路

做电商比价、选品分析或者库存监控的团队，几乎都会碰到同一个需求：把某个小程序里的商品列表稳定地拿下来，落到自己的数据库里。标题里说的「微信小程序爬虫，爬取商品列表，支持写入数据库」，本质就是这条链路——不是去逆向加密算法，而是找到小程序前端调用的那个 HTTP 接口，用合法合规的方式复现请求，把返回的 JSON 解析成结构化字段，再写进 MySQL 或 PostgreSQL。

它适合两类人：一类是刚接触小程序数据采集、想跑通最小闭环的新手；另一类是做过多平台采集、但每次换小程序都要重新摸接口的老手。核心难点从来不是写代码，而是搞清楚请求里带了哪些参数、哪些会过期、分页怎么翻、频率怎么控。这篇笔记按我实际做过的顺序拆开讲，参数和坑都落到具体值上。

2. 抓包定位商品列表接口：工具选择与请求还原

小程序和普通网页最大的区别是它跑在宿主环境里，没有浏览器地址栏，也没有现成的 Network 面板给你看。所以第一步永远是抓包，把小程序发出的真实请求截下来。

2.1 抓包工具的选型与配置思路

常见做法有两类：一类是代理型抓包工具，把手机或模拟器的网络流量导向抓包端；另一类是宿主环境自带的调试能力。代理型工具的好处是通用，任何小程序都能截；代价是要处理证书信任问题，否则 HTTPS 流量全是乱码。

我一般会用一个干净的测试机或模拟器，只装目标小程序，避免其他 App 的流量干扰分析。配置时把代理指向抓包端监听的地址和端口，然后在设备上安装并信任抓包工具的根证书。这一步如果证书没装对，你会看到大量 CONNECT 请求但看不到明文，这是最常见的翻车点。

抓包开始后，在小程序里手动操作：进入商品列表页，往下滑两屏，点进一个商品详情再返回。这几步操作会触发列表接口的首次请求和分页请求，正好把你要的东西全暴露出来。

2.2 从一堆请求里筛出真正的列表接口

抓到的请求可能上百条，怎么快速定位？看三个特征：响应体是 JSON、返回内容里有商品标题和价格字段、请求路径里带 list / search / goods / product 这类词。图片、字体、埋点上报这些都可以直接忽略。

找到候选接口后，重点记录这几项：

这里有个血泪经验：很多小程序的鉴权 token 放在请求头里，而不是请求体。你只复制 body 去重放，会一直返回 401 或「登录失效」，然后误以为是加密问题，其实是头没带全。

2.3 用最小请求验证接口可复现

定位到接口后，先别急着写爬虫，用一条命令验证它能不能脱离小程序独立跑通。这一步能省掉后面大量调试时间。

# 用 curl 重放抓到的请求，验证接口是否可独立调用 # -X POST 指定方法，-H 逐条带上抓包看到的请求头 curl -X POST 'https://example-api.test/api/mall/goods/list' \ -H 'Content-Type: application/json' \ -H 'token: 抓包得到的token值' \ -H 'User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X)' \ -d '{"page":1,"pageSize":20,"categoryId":0}'

逻辑说明：这条命令完全复刻小程序发出的请求。如果返回了正常商品 JSON，说明接口没有做额外的设备指纹或签名校验，后面用 Python 直接请求即可。如果返回错误，优先检查 token 是否过期、请求头是否漏项、body 的字段类型是否对（比如 page 是数字还是字符串）。

参数说明：page 从 1 开始还是从 0 开始，必须实测确认，猜错会导致第一页数据丢失或重复；pageSize 不要一上来就设成 100，很多接口对上限有校验，超过就返回空数组，先用抓包看到的原始值。

3. 用 Python 复现请求并解析商品字段

接口验证通过后，进入正式采集脚本的编写。这一章的目标是：给定一个分类，能把该分类下所有分页的商品全部拉下来，并解析成干净的字典。

3.1 请求封装与分页终止条件

分页最容易出的问题是「什么时候停」。常见做法有三种：返回数组为空、返回数量小于 pageSize、响应里带 total 或 hasMore 字段。我一般优先用 hasMore，没有就用「返回数量小于 pageSize」兜底，同时加一个最大页数保护，防止接口异常时死循环。

import requests import time BASE_URL = "https://example-api.test/api/mall/goods/list" HEADERS = { "Content-Type": "application/json", "token": "从抓包或登录流程获取的token", "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X)", } def fetch_page(page, page_size=20, category_id=0): """拉取单页商品列表，返回 (商品列表, 是否还有下一页)""" payload = {"page": page, "pageSize": page_size, "categoryId": category_id} resp = requests.post(BASE_URL, headers=HEADERS, json=payload, timeout=10) resp.raise_for_status() data = resp.json() # 不同接口字段名不同，这里按抓包实际结构取 items = data.get("data", {}).get("list", []) has_more = data.get("data", {}).get("hasMore", len(items) >= page_size) return items, has_more def crawl_category(category_id, max_pages=50): """按分类翻页采集，带频率控制和页数保护""" all_items = [] for page in range(1, max_pages + 1): items, has_more = fetch_page(page, category_id=category_id) if not items: break all_items.extend(items) if not has_more: break time.sleep(1.2) # 控制频率，避免触发风控 return all_items

逻辑说明：fetch_page 负责单页请求和字段提取，crawl_category 负责翻页循环。把两者拆开的好处是，后面要加并发或换存储时，只改外层不动内层。has_more 的兜底逻辑用 len(items) >= page_size 判断，是因为有些接口不返回 hasMore，但返回满页通常意味着还有下一页。

参数说明：time.sleep(1.2) 这个值不是随便写的，1 秒以下在多数接口上短时间没问题，但连续几百次请求后容易被限流；1.5 秒以上又太慢。1.2 秒是我在几个项目里比较稳的折中值，具体要按目标接口的容忍度调。max_pages 是后悔药，防止接口在最后一页反复返回相同数据导致无限循环。

3.2 商品字段的清洗与类型归一

接口返回的原始字段往往不能直接入库：价格可能是「¥19.90」这种带符号的字符串，销量可能是「1.2万」，库存可能是 null。清洗的目标是让每个字段类型固定，方便建表和后续查询。

import re def parse_price(raw): """把 '¥19.90' / '19.9元' 统一成 float""" if raw is None: return None match = re.search(r"(\d+(\.\d+)?)", str(raw)) return float(match.group(1)) if match else None def parse_sales(raw): """把 '1.2万' 转成 12000，'3000+' 转成 3000""" if raw is None: return 0 text = str(raw).replace("+", "").strip() if "万" in text: num = float(re.search(r"(\d+(\.\d+)?)", text).group(1)) return int(num * 10000) match = re.search(r"(\d+)", text) return int(match.group(1)) if match else 0 def normalize_item(item): """把原始商品字典归一成入库结构""" return { "goods_id": str(item.get("id") or item.get("goodsId")), "title": (item.get("title") or "").strip(), "price": parse_price(item.get("price")), "sales": parse_sales(item.get("sales") or item.get("soldCount")), "category_id": item.get("categoryId"), "shop_name": (item.get("shopName") or "").strip(), }

逻辑说明：parse_price 用正则提取第一个数字，兼容各种货币符号和单位；parse_sales 单独处理「万」这个中文单位，因为直接 int() 会抛异常。normalize_item 把字段名统一，避免不同接口字段名差异污染下游。

参数说明：goods_id 一定要转成字符串，因为有些平台的商品 ID 是长整型，超过 JavaScript 安全整数范围，用 int 存会在某些环节丢精度。title 做 strip 是因为接口返回常带首尾空格，入库后做去重或匹配时会因此对不上。

3.3 增量采集与去重键的选择

全量采集跑一次就够了，日常维护要的是增量：只拿新增或变动的商品。判断依据通常是商品 ID 加更新时间，或者用价格和销量的变化来标记。

def diff_items(new_items, existing_ids): """过滤出数据库中不存在的商品，existing_ids 是已入库 ID 集合""" fresh = [] for item in new_items: norm = normalize_item(item) if norm["goods_id"] not in existing_ids: fresh.append(norm) return fresh

逻辑说明：existing_ids 从数据库一次性查出来放进内存集合，比每条都查库快几个数量级。如果商品量在十万级以内，这种做法完全够用；再大就要考虑用布隆过滤器或分批比对。

参数说明：去重键选 goods_id 而不是 title，因为标题可能被商家改，ID 是稳定的。如果接口不返回稳定 ID，退而求其次用「标题 + 店铺名」的哈希，但要接受一定误判率。

4. 写入数据库：建表、批量插入与更新策略

采集到的数据最终要落库，这一步决定了后面查询和分析顺不顺畅。核心是三件事：表结构设计、批量写入方式、重复数据怎么处理。

4.1 表结构设计与索引取舍

表不用设计得太复杂，但几个关键字段和索引必须提前想清楚，否则数据量上来后查询会很难受。

CREATE TABLE goods ( id BIGINT AUTO_INCREMENT PRIMARY KEY, goods_id VARCHAR(64) NOT NULL COMMENT '平台商品ID', title VARCHAR(255) NOT NULL DEFAULT '', price DECIMAL(10,2) DEFAULT NULL, sales INT DEFAULT 0, category_id INT DEFAULT NULL, shop_name VARCHAR(128) DEFAULT '', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_goods_id (goods_id), KEY idx_category (category_id), KEY idx_price (price) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

逻辑说明：goods_id 上加唯一索引，这样插入时可以用 ON DUPLICATE KEY UPDATE 做「存在则更新」，一条 SQL 同时覆盖新增和变动两种情况，不用先查后写。category_id 和 price 建普通索引，方便按分类筛选和价格区间查询。

参数说明：price 用 DECIMAL(10,2) 而不是 FLOAT，因为浮点数在金额比较时会出现 19.9 存成 19.899999 的情况，做价格排序或等值判断会出错。title 长度给 255 是经验值，多数商品标题不会超，超了就在清洗阶段截断。

4.2 批量插入与 ON DUPLICATE KEY UPDATE

逐条 insert 在几千条数据时还能忍，上万条就明显慢了。批量插入能把往返次数从 N 次降到 N/batch 次，配合唯一索引做更新，是性价比最高的写法。

import pymysql def save_items(items, batch_size=500): """批量写入商品，存在则更新价格和销量""" if not items: return 0 conn = pymysql.connect( host="127.0.0.1", port=3306, user="crawler", password="your_password", database="mall", charset="utf8mb4" ) sql = """ INSERT INTO goods (goods_id, title, price, sales, category_id, shop_name) VALUES (%s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE title = VALUES(title), price = VALUES(price), sales = VALUES(sales), shop_name = VALUES(shop_name) """ total = 0 with conn.cursor() as cur: for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] rows = [(x["goods_id"], x["title"], x["price"], x["sales"], x["category_id"], x["shop_name"]) for x in batch] cur.executemany(sql, rows) conn.commit() total += len(batch) conn.close() return total

逻辑说明：executemany 把一批数据合成一次网络往返，batch_size 取 500 是兼顾内存和速度的常见值。ON DUPLICATE KEY UPDATE 命中唯一索引 uk_goods_id 时执行更新，没命中就插入，省掉了「先 select 再决定 insert 还是 update」的两步操作。

参数说明：batch_size 不要设太大，超过 2000 时单条 SQL 包体可能超过数据库的 max_allowed_packet 限制，报「Packet too large」。如果确实要更大批量，先调数据库参数，或者改用 LOAD DATA LOCAL INFILE。commit 放在每批之后而不是全部结束，是为了失败时已成功的批次不回滚，重跑时靠唯一索引自动去重。

4.3 采集与入库解耦的队列思路

当采集量大或者要同时跑多个分类时，把「抓」和「存」放在一个循环里会互相拖慢。常见做法是采集端把归一化后的数据丢进队列，入库端从队列消费批量写。

import queue import threading data_queue = queue.Queue(maxsize=2000) def producer(category_ids): """采集线程：把归一化数据放入队列""" for cid in category_ids: for item in crawl_category(cid): data_queue.put(normalize_item(item)) data_queue.put(None) # 结束信号 def consumer(): """入库线程：攒批写入""" buffer = [] while True: item = data_queue.get() if item is None: if buffer: save_items(buffer) break buffer.append(item) if len(buffer) >= 500: save_items(buffer) buffer.clear() t = threading.Thread(target=consumer) t.start() producer([1, 2, 3]) t.join()

逻辑说明：生产者只管抓和归一化，消费者只管攒批入库，两边通过有界队列连接。maxsize=2000 起到背压作用，采集太快时 put 会阻塞，避免内存被撑爆。结束信号用 None 传递，比用标志位更直观。

参数说明：消费者攒批阈值 500 和 save_items 的 batch_size 保持一致，避免二次拆分。如果采集端是多线程，队列本身是线程安全的，不用额外加锁。

5. 避坑与排查：token 失效、限流和字段漂移

这一章是我踩过的坑里最值得单独拎出来的几条，每条按现象、原因、解决写，照着排查能省不少时间。

5.1 请求突然全部返回登录失效

现象：脚本跑了几百条后，接口开始返回 401 或「请先登录」，重启脚本也没用。

原因：token 有有效期，通常几小时到几天。抓包时拿到的 token 是当时会话的，过期后所有请求都会被拒。有些接口还会在 token 快过期时返回新 token，但脚本没处理。

解决：把 token 的获取做成独立步骤，不要硬编码在脚本里。如果小程序有登录接口，用账号密码或验证码换 token；如果没有，就在 token 失效时暂停并告警，人工更新后继续。判断失效不要只看状态码，有些接口失效时返回 200 但 body 里 code 是 401，要按业务码判断。

5.2 翻到几十页后返回空数据

现象：前 30 页正常，第 31 页开始返回空数组，但手动在小程序里能滑到更后面。

原因：多数接口对翻页深度有限制，常见上限是 1000 条或 50 页。超过后返回空，不是数据没了，是接口不让你继续翻。

解决：换维度拆分。按分类、按价格区间、按上架时间分段采集，每段控制在限制以内。比如一个分类有 5000 个商品，就按价格 0-50、50-100 这样切，每段单独翻页。拆分维度要选接口支持的筛选参数，抓包时留意请求体里有哪些可选字段。

5.3 价格字段偶尔解析成 None

现象：大部分商品价格正常，少数几条 price 是 None，导致入库后价格为空。

原因：接口对部分商品返回的价格是「面议」「暂无报价」这类文本，正则匹配不到数字。或者字段名不统一，有的商品用 price，有的用 minPrice。

解决：解析函数要覆盖多种字段名和文本格式，匹配不到时返回 None 而不是抛异常。入库后对 price 为空的记录单独标记，人工确认是接口问题还是商品本身没有价格。不要用 0 代替 None，0 会被当成真实价格参与排序和统计。

5.4 高频请求触发限流或封禁

现象：脚本跑得快时，一段时间后所有请求返回 429 或直接超时，换 IP 后恢复。

原因：接口按 IP 或 token 维度做了频率限制，短时间请求过多就触发。有些还会记录异常行为，持续高频会导致更长时间的限制。

解决：控制并发和频率，单 IP 的 QPS 控制在 1 以下比较稳。需要更快时用多个出口分散，但要注意每个出口的请求量都要控制。加随机延迟比固定延迟更难被识别，比如 1 到 2 秒之间随机。遇到 429 不要立即重试，退避等待，等待时间随失败次数递增。

5.5 接口字段悄悄改名导致入库失败

现象：脚本昨天还正常，今天报 KeyError 或大量字段为空。

原因：接口版本更新，字段名从 soldCount 改成 sales，或者把商品数组从 data.list 挪到 data.items。

解决：解析层不要用 item["price"] 这种硬取值，统一用 item.get() 加默认值。同时把原始响应存一份到日志或临时表，字段异常时能对比前后差异。上线前加一个字段校验步骤，关键字段缺失超过阈值就告警，而不是默默写入空数据。

6. 让采集长期稳定：调度、监控与合规边界

跑通一次不难，难的是让它连续跑几个月不出事。这一章讲几个让方案能长期用的具体技巧，以及必须守住的边界。

6.1 用调度器替代手动执行

手动跑脚本只适合验证阶段。长期采集要交给调度器，按固定周期触发。常见做法是用系统自带的定时任务，或者用调度框架管理多个采集任务。

# 每天凌晨 2 点执行一次全量增量采集 # crontab -e 添加下面这行 0 2 * * * /usr/bin/python3 /opt/crawler/run_daily.py >> /var/log/crawler.log 2>&1

逻辑说明：把标准输出和错误输出都重定向到日志文件，出问题时能回溯。run_daily.py 里负责读取分类列表、调用采集和入库、记录本次采集的统计信息。

参数说明：时间选在凌晨是因为目标接口此时负载低，限流概率小。如果数据时效性要求高，可以改成每小时一次，但要把频率控制参数相应调大。

6.2 采集健康度的三个监控指标

不监控的采集脚本等于黑匣子，出问题往往几天后才发现。我一般盯三个指标：单次采集条数、入库成功率和接口错误率。

这三个指标写进日志，配合简单的阈值告警就够了。不需要一上来就上复杂的监控系统，先把数据记下来，出问题时能查。

6.3 采集频率与数据使用的合规边界

技术能跑通不代表可以随便跑。采集前要确认目标数据是否公开、采集行为是否会影响对方服务、采集频率是否在合理范围。只采集业务必需字段，不碰用户隐私数据，不绕过明显的访问控制。数据用途限定在分析和比价这类场景，不用于批量转售或恶意竞争。

我自己的习惯是：任何采集任务上线前，先问三个问题——这个接口是不是公开可访问的、我的频率会不会给对方造成负担、采下来的数据我打算怎么用。三个问题有一个答不上来，就先停下来想清楚。这条线守住了，方案才能长期用下去，而不是跑几天就被封、被迫推倒重来。

希望帮到你。

本文还有配套的精品资源，点击获取