简介面向需要采集速卖通商品数据的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?SearchTextwirelessearbudspage2sortdefault参数拆开看就三块wholesale是搜索路由SearchText是关键词page是页码。sort常用值有default综合排序、price_asc、price_desc、sale_desc销量优先。我一般固定用default和sale_desc两个字段做对比分析因为采购类需求通常最关心销量梯队。分页规则值得先说清楚速卖通的列表页并不承诺无限翻页一个关键词翻到 40 到 70 页之间返回数据会开始变成空壳或重复推荐商品具体上限受词的热度和站点地区影响。这个现象我后面会在避坑章节展开。所以写调度逻辑时一定要设计连续两页取到 0 个商品就停的终止条件而不是傻傻地把页码计数器翻到天荒地老。另外页码是 1 起始的整数URL 里没有 page 参数时等价于 page1。构造 URL 用urlencode比手工拼字符串省心可以避免关键词里带空格、、时把查询串炸掉。2.3 先把页面稳定抓下来requests 会话、请求头与最小抓取函数确定方案后第一步是把页面稳定地拿回来。我用requests.Session()而不是每次新建 requests因为 Session 会自动维持 cookie而速卖通对无 cookie 的陌生人请求非常敏感。第一次访问时带上一个常规浏览器的 User-AgentAccept-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/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: en-US,en;q0.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, timeout15) resp.raise_for_status() return resp.text逻辑说明build_search_url用urlencode处理参数编码关键词里的空格会被安全转成或%20不会把请求串弄坏。fetch_search_page里我加了raise_for_status()HTTP 401/403/503 时第一时间抛异常这样主流程能尽早发现 IP 或参数层面的问题而不是带着错误页面继续往下走。参数说明timeout15单位是秒同时覆盖连接和读取两段设太小在慢网环境会大量超时设太大会拖慢整体重试节奏。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(rscript[^]*(.*?)/script, html, flagsre.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, flagsre.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 对象它就能成功解出后面多出来的分号或杂散字符会被忽略不用自己写括号配平逻辑。参数说明flagsre.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: fhttps://www.aliexpress.com/item/{product_id}.html, }逻辑说明walk_product_cards是个深度优先的生成器任务是把嵌在任意层级里的商品卡对象捞出来不依赖具体的 JSON 路径——速卖通只要调整过一次数据嵌套层级写死的data.list.items[...]几乎必然失效递归反而能多撑几个版本。parse_product_card负责字段名归一化因为同一字段在不同接口版本里可能叫title、subject或productTitle常见做法是把候选字段名按优先级一次性全查。参数说明max_depth8是控制递归深度的保险丝防止某个嵌套极深的异常数据在遍历时消耗过多内存如果返回结果太少可以适当调大到 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 fhttps://www.aliexpress.com/item/{product_id}.html resp session.get(url, timeout15) 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 实体nbsp;或中文站点的人民币符号替换的情况。我的清洗函数统一用正则抽数字再取区间端点。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.0Python 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 但页面里没有商品数据第一次跑脚本的人最容易遇上这个状态码 200HTML 也拿到了但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 里的quot;和amp;这类 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_workers3) 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_INTERVAL2.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, encodingutf-8) as fp: return int(fp.read().strip()) except (FileNotFoundError, ValueError): return 1逻辑说明verify_sample抓最新落库的样本重查用新旧数据的偏差比例做容忍判断。resume_from_checkpoint是断点续跑的入口读取上次保存的页数文件不存在或内容损坏时默认从第 1 页开始。把这两个函数组装进主流程后整个抓取链路就有了暂停—续跑—验错的闭环而不是一次性的黑箱任务。我一直保留的习惯是每轮抓完先看空页终止的位置再抽验样本最后才把数据交给分析。校验永远是最后一道闸门它比任何解析技巧都能拦住数据质量的崩盘后果。希望这篇把方案选型、核心源码、参数细节和异常排查都摊开的笔记能帮到你少走一轮我当年走过的弯路。本文还有配套的精品资源点击获取