简介：面向需要采集速卖通商品数据的Python爬虫学习者与电商数据研究人员，这份源码包演示了从发送HTTP请求、设置请求头与代理，到BeautifulSoup解析HTML、定位商品标题与价格，再到使用Selenium处理JavaScript动态加载销量信息，以及利用pandas落盘CSV的完整流程。压缩包共9个文件，主体是spider.py爬虫脚本，同时包含XML工程配置、说明txt以及工程元数据等辅助文件，整体仅6KB，轻量易读。当前已有792人学习或下载。代码中清晰划分了请求、解析、动态内容获取和数据清洗等模块，并提及多线程与反反爬注意事项，适合作为速卖通或同类电商平台抓取项目的起步模板，便于按需修改选择器与代理策略。
1. 速卖通商品信息抓取：别急着上模拟浏览器，先把内嵌数据结构吃透
做速卖通商品信息抓取，很多人第一反应是上 Selenium 或 Playwright，让浏览器自动化去点、去滚、去等渲染。这个方向不是不行，而是贵——别人一个页面 30KB 能拿到的数据，你开浏览器要多消耗几十倍内存和带宽，爬 3000 个商品就够把一台 4G 的服务器拖到告警。我的血泪经验是：速卖通的列表页和详情页本身就内置了大量结构化 JSON 数据，藏在<script>标签里。把这部分吃透，用requests加正则定位，加一层递归提取，就能在完全不碰渲染引擎的前提下把商品ID、标题、价格、销量、评价数完整抓下来。这篇文章给的就是这一整套完整源码的拆解，从请求构造到解析、落库、避坑，一次讲透。适合要做竞品监控、选品分析或价格追踪，且预算撑不住每秒钟都开浏览器的团队和个人开发者。
2. 抓取方案怎么选：三种数据来源的取舍与速卖通页面结构分析
2.1 三种抓取途径：直接解析、内嵌 JSON 与浏览器渲染的取舍
速卖通商品信息的获取，业界常见的做法是往三个方向走。第一种是直接解析 HTML 标签，用 BeautifulSoup 去find_all('a', class_=...)找商品链接和标题。这条路实现简单，但页面改版一次，选择器就废一半，而且很多字段（比如真实价格、优惠区间）并不直接渲染在 HTML 里，而是由 JavaScript 加载后填充。第二种是从内嵌 JSON 里提取，这也是我推荐的主力方案。页面源码里的<script>标签会初始化一个或多个全局对象，商品数据以 JSON 字符串的形式躺在里面，包含的字段比页面上肉眼可见的还要全。第三种是浏览器渲染抓取，靠 Selenium 或 Playwright 模拟用户，适合必须拿到动态二次请求结果的场景，比如某些需要登录态才展示的直播价、区域库存。
三种方式我在抓取不同站点时都轮过一遍，直接解析的体验就是"黑匣子"，选个 class 名还得赌它明天下班前不被改掉；浏览器渲染则是在性能和稳定性上做减法。内嵌 JSON 的方案多数情况雷打不动，因为速卖通的页面服务端渲染框架需要把首屏数据交给浏览器里的 JS 去 hydrate，这份数据就在原 HTML 里，属于"必须公开"的部分。下表是三个方案的对比，按我踩过的坑排列：
| 方案 | 请求开销 | 解析复杂度 | 反爬触发概率 | 字段完整度 | 场景适配 |
|---|---|---|---|---|---|
| HTML 标签解析 | 低 | 高（选择器易碎） | 低 | 中（缺价格/优惠） | 快速原型、小规模验证 |
| 内嵌 JSON 提取 | 低 | 中（定位+递归） | 低 | 高（含隐藏字段） | 批量抓取、长期监控 |
| 浏览器渲染 | 极高 | 低（所见即所得） | 高（特征明显） | 高 | 登录态、强动态页面 |
2.2 速卖通列表页 URL 参数与分页规则：一个关键词最多能翻多少页
先看请求层面的构造。速卖通搜索列表页的典型 URL 大概长这样：
https://www.aliexpress.com/wholesale?SearchText=wireless+earbuds&page=2&sort=default参数拆开看就三块：wholesale是搜索路由，SearchText是关键词，page是页码。sort常用值有default（综合排序）、price_asc、price_desc、sale_desc（销量优先）。我一般固定用default和sale_desc两个字段做对比分析，因为采购类需求通常最关心销量梯队。
分页规则值得先说清楚：速卖通的列表页并不承诺无限翻页，一个关键词翻到 40 到 70 页之间，返回数据会开始变成空壳或重复推荐商品，具体上限受词的热度和站点地区影响。这个现象我后面会在避坑章节展开。所以写调度逻辑时，一定要设计"连续两页取到 0 个商品就停"的终止条件，而不是傻傻地把页码计数器翻到天荒地老。另外，页码是 1 起始的整数，URL 里没有 page 参数时等价于 page=1。构造 URL 用urlencode比手工拼字符串省心，可以避免关键词里带空格、&、=时把查询串炸掉。
2.3 先把页面稳定抓下来：requests 会话、请求头与最小抓取函数
确定方案后，第一步是把页面稳定地拿回来。我用requests.Session()而不是每次新建 requests，因为 Session 会自动维持 cookie，而速卖通对无 cookie 的陌生人请求非常敏感。第一次访问时带上一个常规浏览器的 User-Agent，Accept-Language 设成英文或目标站点语言，能减少被重定向到站内引导页的机会。
import requests from urllib.parse import urlencode session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9", "Referer": "https://www.aliexpress.com/", }) def build_search_url(keyword: str, page: int = 1, sort: str = "default") -> str: params = { "SearchText": keyword, "page": page, "sort": sort, } return "https://www.aliexpress.com/wholesale?" + urlencode(params) def fetch_search_page(keyword: str, page: int = 1) -> str: url = build_search_url(keyword, page) resp = session.get(url, timeout=15) resp.raise_for_status() return resp.text逻辑说明：build_search_url用urlencode处理参数编码，关键词里的空格会被安全转成+或%20，不会把请求串弄坏。fetch_search_page里我加了raise_for_status()，HTTP 401/403/503 时第一时间抛异常，这样主流程能尽早发现 IP 或参数层面的问题，而不是带着错误页面继续往下走。
参数说明：timeout=15单位是秒，同时覆盖连接和读取两段；设太小在慢网环境会大量超时，设太大会拖慢整体重试节奏。User-Agent 我建议固定一个完整版本，不要用python-requests/2.x这种默认值，否则在服务端日志里和主动说"我是脚本"没区别。Referer 设置为站点首页是模拟正常访问路径的常规操作，能降低一部分请求特征分。
3. 核心解析源码：把商品 ID、价格、销量从杂乱 HTML 里抠出来
3.1 从 script 脚本里定位商品 JSON 数据块
拿到 HTML 之后，直接去标签树里找商品节点是很脆弱的做法。我一般会把目标收窄到<script>标签，尤其那些以window.xxx = {...}或var xxx = {...}开头的初始化脚本。难点在于这类 JSON 对象的结尾往往不干净：后面会跟着分号、换行甚至下一个赋值语句。用普通的json.loads会直接报错，这时候JSONDecoder.raw_decode是更可靠的解药，它能在解析完一个完整 JSON 对象后，返回剩余未消费的字符串位置，而不是要求整段文本都是合法 JSON。
import json import re def extract_embedded_json(html: str) -> list: """ 遍历页面里所有 <script> 标签，把以赋值形式暴露的 JSON 对象解出来。 返回一个 list，里面每个元素都是一个 dict 或 list。 """ scripts = re.findall(r'<script[^>]*>(.*?)</script>', html, flags=re.S) results = [] for script in scripts: script = script.strip() if not (script.startswith("window.") or script.startswith("var ")): continue m = re.search(r'=\s*(\{.*)', script, flags=re.S) if not m: continue try: data, _ = json.JSONDecoder().raw_decode(m.group(1)) results.append(data) except json.JSONDecodeError: continue return results逻辑说明：先用正则把整个 HTML 里所有<script>内容抽出来，避免在样式表或注释里浪费时间；然后只保留以window.或var开头的脚本块；再通过= \s*(\{.*)拿到从{到结尾的候选字符串，交给raw_decode做非严格解析。raw_decode的好处是，只要开头是一个完整合法的 JSON 对象，它就能成功解出，后面多出来的分号或杂散字符会被忽略，不用自己写括号配平逻辑。
参数说明：flags=re.S必须加，否则.匹配不了换行。raw_decode返回的第二个值是剩余字符串长度，我们这里用不上，所以显式忽略。如果这段脚本开头是方括号[而不是花括号{，当前正则会漏掉，实际遇到的话把正则改成=\s*([\[\{].*)即可，这个我通常会在调用方预留一个扩展点。
3.2 商品列表字段映射：价格区间、订单量、评价数的解析代码
拿到一个或多个全局 JSON 对象后，下一步是递归地把商品卡片找出来。速卖通前后端数据结构里有几个稳定标识：商品卡对象几乎必带itemId或productId，两者的值是一段数字字符串。我的方式是不做具体路径假设，写一个带深度限制的递归遍历，按标识字段去"碰运气式"收集卡片对象，收集完再做字段校验和清洗。
def walk_product_cards(obj, depth: int = 0, max_depth: int = 8): if depth > max_depth: return if isinstance(obj, dict): if "itemId" in obj or "productId" in obj: yield obj for value in obj.values(): yield from walk_product_cards(value, depth + 1) elif isinstance(obj, list): for item in obj: yield from walk_product_cards(item, depth + 1) def parse_product_card(card: dict) -> dict: """把原始商品卡片对象映射成整洁的记录。字段缺失时返回 None 再由上层过滤。""" product_id = str(card.get("itemId") or card.get("productId") or "") if not product_id: return None title = card.get("title") or card.get("subject") or card.get("productTitle") or "" raw_price = (card.get("formatedPrice") or card.get("skuTitle") or card.get("price") or "") price_min, price_max = parse_price_range(str(raw_price)) orders = parse_orders(str(card.get("tradeDesc") or card.get("orderCount") or card.get("salesCount") or "")) rating = 0.0 try: rating = float(card.get("averageStar") or card.get("itemRating") or 0) except (TypeError, ValueError): rating = 0.0 return { "product_id": product_id, "title": title.strip(), "price_min": price_min, "price_max": price_max, "orders": orders, "rating": rating, "image_url": card.get("image") or card.get("imgUrl") or "", "detail_url": f"https://www.aliexpress.com/item/{product_id}.html", }逻辑说明：walk_product_cards是个深度优先的生成器，任务是把嵌在任意层级里的商品卡对象捞出来，不依赖具体的 JSON 路径——速卖通只要调整过一次数据嵌套层级，写死的data.list.items[...]几乎必然失效，递归反而能多撑几个版本。parse_product_card负责字段名归一化，因为同一字段在不同接口版本里可能叫title、subject或productTitle，常见做法是把候选字段名按优先级一次性全查。
参数说明：max_depth=8是控制递归深度的保险丝，防止某个嵌套极深的异常数据在遍历时消耗过多内存；如果返回结果太少，可以适当调大到 12，但要注意单个页面对象索引过深时，Python 递归栈本身就变成风险。价格和订单量的清洗函数在下一节展开，这里传的是原始字符串和数字。
3.3 详情页与 SKU 解析：补全描述、规格与店铺信息
列表页数据再全，也会缺长描述和多规格 SKU 明细。这两块信息在详情页里。详情页的 URL 结构稳定，就是把商品 ID 拼到/item/{id}.html后面。详情页 HTML 里的内嵌 JSON 位置和列表页不同，常见是挂在runParams或类似的全局变量下，但提取思路完全一样：复用extract_embedded_json找到包含该商品 ID 的顶层对象，再按字段取skuModule、descriptionModule、storeModule等子结构。
def fetch_product_detail(product_id: str) -> dict: url = f"https://www.aliexpress.com/item/{product_id}.html" resp = session.get(url, timeout=15) resp.raise_for_status() html = resp.text for data in extract_embedded_json(html): sku = iter_deep_get(data, "skuModule", "productSKUPropertyList") if sku: return { "product_id": product_id, "sku_properties": sku, "description": iter_deep_get(data, "descriptionModule") or iter_deep_get(data, "productDescription"), "store_name": iter_deep_get(data, "storeModule", "storeName"), "store_url": iter_deep_get(data, "storeModule", "storeUrl"), } return {"product_id": product_id} def iter_deep_get(obj: dict, *keys): """按 key 链向下取，中途找不返回 None，不做 KeyError 抛错。""" for key in keys: if not isinstance(obj, dict): return None obj = obj.get(key) return obj逻辑说明：iter_deep_get做的是链式取值，好处是任何一层缺失都直接返回None，不用写一长串try/except KeyError。fetch_product_detail会遍历详情页里的所有内嵌 JSON 对象，只找携带skuModule的那个，因为 SKU 数据是详情页独有的，是确认"抓的就是这个商品详情"的强信号。描述字段有长文本和 HTML 片段两种存在形态，落库时记得做标签剥离。
参数说明：详情页不需要刻意模拟浏览器的滚动行为，内嵌数据是首屏就存在的。如果发现解析结果缺 SKU，优先怀疑是地区站或语言站变了数据字段名，而不是请求本身失败。把当前页面的 script 标签数量和字符数分别打印出来，能快速判断是不是拿到了验证码页或空壳页。
4. 数据落库：去重、增量更新与 SQLite 存储的完整实现
4.1 清洗规范化：价格区间、销量单位与缺失值处理
抓下来的原始字符串离能入库还差两步：价格统一成数值、销量把K/M后缀展开。价速卖通的列表页价格展示通常有两种格式：单个价US $1.99和区间价US $1.59 - US $2.39，偶尔还有货币符号被 HTML 实体 或中文站点的人民币符号替换的情况。我的清洗函数统一用正则抽数字，再取区间端点。
import re PRICE_RE = re.compile(r'[\$€£¥]\s*([0-9][0-9,]*(?:\.[0-9]{1,2})?)') def parse_price_range(raw_price: str): if not raw_price: return None, None matches = PRICE_RE.findall(raw_price.replace("\u00a0", " ")) if not matches: return None, None values = [float(m.replace(",", "")) for m in matches] if len(values) == 1: return values[0], values[0] return min(values), max(values) def parse_orders(text: str) -> int: if not text: return 0 m = re.search(r'([\d.,]+)\s*([KkMm]?)\s*\+?\s*(?:orders?|sold)', text) if not m: return 0 num = float(m.group(1).replace(",", "")) unit = m.group(2) if unit and unit.lower() == "k": num *= 1000 elif unit and unit.lower() == "m": num *= 1_000_000 return int(num)逻辑说明：PRICE_RE匹配货币符号加数字，findall一次拿回所有价格实例。区间价会产生两个匹配，取min/max是为了规避页面上"划线价在前、活动价在后"这类顺序反直觉的展示；如果只有单个匹配，上下限给同一个值，后续按price_min排序时不会丢数据。parse_orders处理的是1,234 sold、2.5K+ orders这两种最常见样式的销量文本。
参数说明：正则里的货币符号集合\$€£¥是我实际抓到的范围，美元做主货币、欧元英镑做备选。如果目标站设定为巴西或俄罗斯站点，货币符号会超出这个集合，到时在字符组里补R$、₽即可。缺失值统一用None或0而不是空字符串，这样在 SQLite 里查询价格区间时不会拿到脏类型。
4.2 主键与 upsert：增量更新怎么做到不重复
抓取脚本如果每次都全量重跑，商品数据表会膨胀得很离谱——同一个商品ID重复出现十几次，排重全靠后期处理，完全是给自己制造麻烦。正确的落库姿势是以product_id做主键，用 SQLite 的ON CONFLICT做 upsert：插入时发现主键已存在，就更新价格、销量、评价数、最后抓取时间这些可变字段；而首见时间first_seen永远保持第一次插入的值。
import sqlite3 from datetime import date DB_PATH = "aliexpress_products.db" SCHEMA = """ CREATE TABLE IF NOT EXISTS products ( product_id TEXT PRIMARY KEY, title TEXT, price_min REAL, price_max REAL, orders INTEGER, rating REAL, image_url TEXT, detail_url TEXT, first_seen TEXT, last_seen TEXT ); """ UPSERT_SQL = """ INSERT INTO products ( product_id, title, price_min, price_max, orders, rating, image_url, detail_url, first_seen, last_seen ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(product_id) DO UPDATE SET price_min = excluded.price_min, price_max = excluded.price_max, orders = excluded.orders, rating = excluded.rating, image_url = excluded.image_url, last_seen = excluded.last_seen; """ def save_product(conn: sqlite3.Connection, record: dict) -> None: today = date.today().isoformat() conn.execute(UPSERT_SQL, ( record["product_id"], record["title"], record["price_min"], record["price_max"], record["orders"], record["rating"], record["image_url"], record["detail_url"], today, today, )) conn.commit()逻辑说明：excluded.xxx是 SQLite upsert 语法里的特殊前缀，表示"本次 INSERT 打算写入的新值"。ON CONFLICT子句只在product_id冲突时触发，所以同一个商品多次抓取后，first_seen保持第一次落地的时间，last_seen不断刷新，天然形成一套可追溯商品价格史的最小结构，后续做降价提醒时直接查first_seen != last_seen的记录即可。
参数说明：conn.commit()我放在每条记录后，单机每日几千条数据时性能完全够用；如果以后数据量涨到十万条以上，改成每抓完一页 commit 一次，再把一次批量插入的 records 列表用executemany送进去，磁盘 IO 会好看很多。SQLite 的 upsert 语法要求数据库版本不低于 3.24.0，Python 3.8 之后自带的 sqlite3 模块基本都满足，老环境可以用SELECT + UPDATE + INSERT三段式替代。
4.3 完整入库代码与主流程串联
把抓列表页、解析、落库串起来，就是前面所有代码块的汇总。注意主循环里一定要有"双空页终止"的判断，因为速卖通翻到尾部时数据源开始放空，继续请求只会白白烧配额。
def save_products(conn: sqlite3.Connection, products: list) -> int: saved = 0 for record in products: if record is None or record["product_id"] in ("", "None"): continue save_product(conn, record) saved += 1 return saved def crawl_keyword(keyword: str, max_pages: int = 50): empty_streak = 0 conn = sqlite3.connect(DB_PATH) conn.execute(SCHEMA) for page in range(1, max_pages + 1): html = fetch_search_page(keyword, page) products = [] for data in extract_embedded_json(html): for card in walk_product_cards(data): record = parse_product_card(card) if record: products.append(record) saved = save_products(conn, products) print(f"[{page}] page, {len(products)} cards, {saved} new/updated") if saved == 0: empty_streak += 1 if empty_streak >= 2: print("two consecutive empty pages, stop crawling") break else: empty_streak = 0 time.sleep(3 + random.random() * 2) conn.close()逻辑说明：empty_streak连续计数两次空页就断掉循环，这是应对速卖通翻页尾部数据衰减的兜底逻辑。save_products里先过滤掉没有商品 ID 的空记录，避免把异常对象写进库。整个循环里session复用同一个连接，cookie 从头保持到尾，比每个请求新建连接要稳得多。
参数说明：max_pages默认 50，实际能跑多少取决于关键词热度。每条请求之间的sleep(3 + random.random() * 2)是刻意的，3 秒保底加 0 到 2 秒的随机抖动，让请求间隔不是固定节拍；完全均匀的间隔在时序统计上反而更容易被判为机器行为。
5. 速卖通抓取避坑手册：限流、验证码与解析翻车的排查清单
5.1 现象：HTTP 200 但页面里没有商品数据
第一次跑脚本的人最容易遇上这个：状态码 200，HTML 也拿到了，但extract_embedded_json一个对象都解不出来，walk_product_cards自然一个卡片都没有。真实原因通常是会话里根本没有 cookie，或者首次访问直接命中低配页面。
排查路径：把 HTML 里<script>标签数量打印出来看看，正常列表页一般有几十个 script 块，少于 5 个基本就是空壳页；再检查页面里有没有验证码关键字，比如captcha、verify这类英文单词。解决方式是每次启动脚本前，先用session.get("https://www.aliexpress.com/")逛一下首页完成 cookie 初始化，再开始抓列表页。
5.2 现象：前 100 页正常，第 150 页开始一直返回同一段空结果
这种属于翻页翻到了数据源承诺的尾部。我曾经用sale_desc排序抓一个热门词，前 120 页都好好的，之后连续十几页都是同样的推荐商品集合，金额还越抓越少。原因不是被封，而是排序接口在深层页码里把结果切成了"推荐流"，不再是真实的搜索结果。
解决方式就是前面说的空页终止逻辑。不过这里有个细节：空页终止要把"解析到 0 个商品"和"解析到同一批重复商品"分开判断。重复流的特点是商品 ID 集合和上一页几乎完全一致，这时应该把重复连续计数也纳入终止条件，否则会在一堆重复数据上浪费时间。
5.3 现象：正则能找到片段但 json.loads 一直报错
内嵌 JSON 最常见的问题是两层：一是赋值语句后面还跟着其他代码，json.loads这种整串解析必然失败；二是数据内部有转义字符，比如\u002F、\x2F这类 unicode 表示，肉眼看着像乱码。我第一版脚本就在这里翻过车，后来切到JSONDecoder.raw_decode加json.loads两步走，前者处理结尾不干净，后者正常解析剩余完整串，基本把这类问题压干净了。
另一个容易被忽略的坑是 script 里的"和&这类 HTML 实体。它们不是合法 JSON 字符，直接塞给解析器会报 JSONDecodeError。破解办法是解析前先做一次简单反转义，html.unescape()，把实体还原成原始字符再进 JSON 解析器。
5.4 现象：价格字段拆不开或出现乱码字符
价格解析出问题上限不要慌，先把原始字符串打印出来。速卖通不同国家站点的价格格式差异很大：英文站是US $1.99，俄罗斯站是US $1.99 / 2.79 руб，巴西站可能混进R$符号，还有把价格区间写成$0.99~$1.99的版本。我的PRICE_RE正则已经覆盖了主流货币符号，遇到没覆盖的格式，在字符组里加符号即可，不要在正则里硬编码"US $"这种完整前缀。
另外注意\u00a0这是不间断空格，很多时候页面把数字和货币符号之间的空格转成了这个特殊字符。代码里replace("\u00a0", " ")就是给它做归一化的。如果解析出的价格全部是None，优先怀疑是这种不可见字符把正则匹配搞断了。
5.5 现象：本地正常运行，部署到服务器立刻触发验证码
本地抓得好好的，代码原封不动挪到服务器，请求几次就触发验证码，是典型的运行环境特征差异问题。服务器出口 IP 的网段如果被站点标记为数据中心地址，信任度天然比家庭宽带低；再加上服务器上往往多个任务同时跑，请求频率叠加后就更容易触发人机验证。这算是上线后遇到的"玄学"问题，其实原因很机械：IP 信誉和频率叠加。
解决思路是降速、错峰，别把抓取任务叠在同一个出口 IP 上并发冲锋。把每页间隔从 3 秒提高到 6 到 8 秒，命中验证码的概率会显著下降；同时把任务错开在不同时段跑。如果你还叠加了其他用到同一出口的自动化任务，一定要错开时间窗口。大规模场景下正确做法是横向扩展多个出口，而不是在一个出口上硬顶并发。
6. 进阶技巧：把单机脚本改成可暂停、可续跑、可校验的小并发任务
6.1 用信号量与线程池做可控并发
抓取速度的瓶颈往往不在解析，而在网络往返。单线程一页一页抓，一百页要跑十几分钟；但裸上ThreadPoolExecutor并发容易让请求频率失控，把避坑章节里好不容易建立的低频优势全丢掉。我的习惯是给并发套一个信号量，把同时飞在路上的请求数量锁在 3 到 5 个，并且每个线程抓完一个页面后必须经过同一把"节流锁"才能发出下一个请求。
import threading import time from concurrent.futures import ThreadPoolExecutor _semaphore = threading.Semaphore(3) _page_sleep_lock = threading.Lock() _last_request_ts = 0.0 MIN_INTERVAL = 2.0 def throttled_fetch(keyword: str, page: int) -> list: global _last_request_ts with _semaphore: with _page_sleep_lock: now = time.time() wait = MIN_INTERVAL - (now - _last_request_ts) if wait > 0: time.sleep(wait) _last_request_ts = time.time() html = fetch_search_page(keyword, page) products = [] for data in extract_embedded_json(html): for card in walk_product_cards(data): record = parse_product_card(card) if record: products.append(record) return products def crawl_with_pool(keyword: str, pages: range): results = [] with ThreadPoolExecutor(max_workers=3) as executor: futures = {executor.submit(throttled_fetch, keyword, p): p for p in pages} for future in as_completed(futures): results.extend(future.result()) return results逻辑说明：Semaphore(3)把同时在执行网络请求的线程数锁在 3 个以内，_page_sleep_lock配合_last_request_ts实现全局的最小请求间隔 2 秒。这两个限制叠加的效果是：并发上限 3，但最密请求间隔 2 秒，任何线程都不能绕过节流控制直接打站点。
参数说明：MIN_INTERVAL和Semaphore的值在生产环境需要按出口 IP 的信誉动态调。初次使用我建议MIN_INTERVAL=2.0、并发 3，跑半小时没触发验证码再逐步加压；发现限流先把并发降到 1，间隔升到 5 秒。全局节流锁有个副作用：并发收益会被高延迟摊薄，因为锁只限制了请求发出频率，没限制每个请求自身的耗时。
6.2 断点续跑与数据校验：抓完怎么确认没抓漏
脚本中断是常态，内存挂了、服务器重启、网络抖动都可能导致任务半路断掉。我习惯把"当前抓到的最大页数"写进一个 checkpoint 文件，每次页面成功入库后更新一次，下次启动从上次断点续跑，而不是从头再来。
数据校验方面，不能只靠"没报错"来判断抓取成功。我会在框架跑完后随机抽 10 个商品 ID 重新拉一次详情页，比较列表页记录里的orders、rating两个字段是否在误差范围内；如果超过 20% 的抽查商品对不上，说明解析逻辑可能已经跟页面真实结构脱节，这一轮的数据直接作废重抓。
def verify_sample(conn: sqlite3.Connection, sample_size: int = 10): rows = conn.execute( "SELECT product_id, orders, rating FROM products ORDER BY last_seen DESC LIMIT ?", (sample_size,), ).fetchall() mismatch = 0 for product_id, old_orders, old_rating in rows: detail = fetch_product_detail(product_id) # 这里以详情页为准重新解析一次销量和评分 new_orders = parse_orders(str(detail.get("orders", ""))) if new_orders and abs(new_orders - old_orders) > max(10, new_orders * 0.2): mismatch += 1 return mismatch def resume_from_checkpoint(path: str = "checkpoint.txt") -> int: try: with open(path, "r", encoding="utf-8") as fp: return int(fp.read().strip()) except (FileNotFoundError, ValueError): return 1逻辑说明：verify_sample抓最新落库的样本重查，用新旧数据的偏差比例做容忍判断。resume_from_checkpoint是断点续跑的入口，读取上次保存的页数，文件不存在或内容损坏时默认从第 1 页开始。把这两个函数组装进主流程后，整个抓取链路就有了"暂停—续跑—验错"的闭环，而不是一次性的黑箱任务。
我一直保留的习惯是：每轮抓完先看空页终止的位置，再抽验样本，最后才把数据交给分析。校验永远是最后一道闸门，它比任何解析技巧都能拦住数据质量的崩盘后果。希望这篇把方案选型、核心源码、参数细节和异常排查都摊开的笔记能帮到你，少走一轮我当年走过的弯路。
本文还有配套的精品资源，点击获取
