简介面向爬虫初学者与电商数据采集人员的速卖通AliExpress商品信息抓取完整 Python 源码包覆盖从页面请求、HTML 解析、动态渲染到数据存储的全流程。源码按实际抓取流程组织从商品页面结构入手给出请求头设置与代理 IP 轮换以降低封禁风险用 requests 获取页面、BeautifulSoup 解析 HTML针对动态加载的销量、评价引入 Selenium 等待渲染再用 pandas 清洗数据并存储 CSV同时兼顾多线程提速与 robots.txt 约束。整套代码结构清晰、注释适当可作为入门项目直接运行或二次开发。资源包为 RAR 压缩格式共 9 个文件核心是 spider.py 主脚本另含 5 个 XML 工程配置、1 个忽略规则文件及 1 个说明文档整体仅 6KB轻量便携。该资源目前已有 792 人学习浏览适合希望快速掌握商品信息定位、动态加载处理和本地存储全流程的开发者参考。1. 速卖通商品信息抓取从接口到入库的完整落地链路做跨境电商选品、竞品监控、价格比对的团队每天打开浏览器在速卖通上手动翻商品标题、价格、销量、评分再一条条复制到表格里这是最原始也最费人的数据采集方式。当你需要维护几千个SKU的价格变化时手动方案就彻底撑不住了。速卖通商品信息抓取本质上就是把浏览器里发生的每一步交互改写成脚本按你自己的节奏批量执行。一个反直觉的结论先放在这里速卖通的商品数据并不藏在复杂的页面DOM里而是通过几个比较固定的JSON接口下发的。真正让你拿不到数据的瓶颈从来不是解析HTML而是三件事——会话Cookie的签名校验、请求频率的节奏控制、以及TLS层设备指纹的识别。这三件事处理不干净哪怕你把选择器写得再漂亮第二天也会被安全验证页拦在门外。这篇文章直接给出一套能落地的抓取链路从抓包定位真实接口到Python完整源码再到反爬对抗策略和五个典型的翻车场景。适合想自己维护采集程序的产品、运营和Python工程师阅读。跟着走完你能清楚回答三个问题数据源接口在哪、最小可用代码长什么样、哪些坑会让你白干一晚上。2. 先抓包再写码定位速卖通商品列表与详情页的真实数据接口2.1 为什么先抓包而不是直接解析页面早期网上的采集教程爱用BeautifulSoup直接从HTML里抠标题和价格。这套路在今天的速卖通页面上已经不好用了。页面是SSR和客户端渲染混着来的类名经常带随机后缀变量名被webpack压实过你今天写好的解析规则下个月页面发一次版就全部作废。凡是靠DOM结构找数据的方案本质都是在跟前端团队的前端发布节奏赛跑你永远跑不赢。抓包的价值在于绕过页面直接看到浏览器与服务器之间交换的JSON数据。接口路径相对固定字段结构比DOM稳定得多。你一旦确认了接口后续工作就变成「按参数请求→拿JSON→清洗入库」彻底不依赖DOM。这也是我每次接手这类采集任务的第一步先花半小时抓包而不是急着写代码。操作步骤很简单。打开浏览器无痕窗口访问速卖通首页按F12切到Network面板勾选Preserve log保留日志。在页面上搜索一个关键词比如dress然后回到Network面板逐个点开新出现的XHR请求在Response里搜productId。哪个请求能搜到哪个就是你要找的列表数据接口。确认之后右键复制成cURL后面转成Python代码就很快。初学者最容易犯的错是去解析那个巨大的HTML文档请求形如wholesale-dress.html。那个响应确实有几十KB甚至上百KB但你要找的不是它而是返回纯JSON的XHR请求。判断标准很简单响应以{开头并且能在里面搜到商品字段的才是正确的数据接口。2.2 列表页的数据接口与关键参数以批发搜索页为例列表数据接口的地址形如https://www.aliexpress.com/w/wholesale-{keyword}.html。注意它返回的仍是HTML加内嵌JSON的混合体而不是纯JSON。你需要从页面文本里把那段初始化数据抽取出来。常见的关键参数有两个page控制页码从1开始每页固定返回若干条商品SortType控制排序方式。默认排序是综合匹配但做价格监控时我一般会用total_tran_cnt_desc按销量降序这样抓到的商品更接近该关键词下的热卖款比默认排序的波动小。另外部分接口也接受SearchText参数传入关键词具体以你抓包时看到的query为准。拿到响应文本后在源码里搜索productId或sellerItems这两个标记就能定位数据块的起点。你可以先写一个应急用的正则把商品ID抠出来确认连通性import re # 假设 text 是列表页返回的完整 HTML matches re.findall(r(productId|productIdStr):?(\d)?, text) print(len(matches)) # 打印提取到的商品 ID 数量这段正则只能用来验证接口连通性不能当作正式方案。因为页面里的JSON块可能被转义直接全局匹配容易把其他数字也捞进来。正式的抽取逻辑要么找到内嵌JSON的变量名后整体json.loads要么用__NEXT_DATA__这类标准数据节点解析这些会在第三章的源码里写清楚。列表接口返回的字段中重点关注这几个productId是后续详情接口的主键subject是商品标题skuInfos里是SKU级别的价格区间tradeVolume是近期销量evaluateScore是评分。不同国家或地区的响应里字段名可能略有差异第一次跑通时要先打印原始JSON确认你的字段路径没写错。2.3 详情页接口的参数结构与动态令牌打开任意一个商品详情页地址形如/item/1005001234567.htmlNetwork面板里会出现一个名称为detail的XHR请求完整地址形如https://www.aliexpress.com/aeglodetailweb/api/product/detail。这个接口返回的JSON很完整SKU价格、库存、运费模板、卖家信息都在里面。这个接口的参数比较固定参数类型说明itemIdstring商品唯一ID必须与页面URL一致currencystring币种USD、EUR等localestring语言区域一般用en_USshipCountrystring收货国家影响最终价格与运费展示shipCountry这个参数容易被忽略但它会让同一个商品的返回价格不一样。比如收货地址填美国还是填英国价格、税费、可用物流方式都会变。如果你要做多站点比价这个参数必须固定下来否则数据口径不统一。真正麻烦的是会话动态令牌。速卖通的接口请求里有一类以_m_开头的Cookie参与请求签名过期时间不长。令牌失效后接口不会明确报错而是直接返回success: false或者把你重定向到验证页。应对办法也很直接每次任务开始前先请求一次首页让服务端种下新的令牌再继续抓详情页。第三章的create_session函数就是干这件事的。有一点必须提前说清楚你在自己机器上抓包看到的接口路径和字段名可能与我这里给出的不完全一致。不同地区、不同账号状态下的接口版本会有差异这是正常的。文章里的代码是完整可运行的骨架遇到差异时核心思路不变改参数名和字段路径即可。3. 用 Python 实现速卖通商品抓取完整源码与逐段解释3.1 依赖选型为什么用 curl_cffi 而不是纯 requests先安装依赖pip install requests curl_cffirequests不用多说Python最常用的HTTP库。curl_cffi是我后来坚持加进项目里的一个库它基于libcurl可以把TLS握手指纹伪装成Chrome浏览器的样子。速卖通这类阿里系站点会对TLS指纹做采集直接用原生requests发请求哪怕Cookie齐全也可能在传输层就被标记为脚本客户端。换成curl_cffi之后同一个接口的请求通过率会明显提高。这个库不能帮你绕过风控它只是让你的请求看起来更像一个真实浏览器。如果UA是Chrome的、TLS指纹却是Python默认的这本身就是最明显的机器特征。用curl_cffi把指纹对齐之后你再配合Cookie和限速整套请求才说得上像一个正常用户。如果你出于某些原因不想引入这个库继续用纯requests也能跑但建议做好频繁被验证的预期准备。下面所有代码的Session都基于curl_cffi。3.2 抓取商品列表页代码与字段抽取import json import re from curl_cffi import requests session requests.Session(impersonatechrome) def fetch_list(keyword: str, page: int): url fhttps://www.aliexpress.com/w/wholesale-{keyword}.html resp session.get(url, params{page: page}, timeout20) # 如果被重定向到安全验证页resp.url 会带 captcha 等特征 if captcha in resp.url.lower() or resp.url.rstrip(/) ! url.rstrip(/): print(被安全校验拦截当前URL:, resp.url) return None text resp.text # 优先抽取内嵌 JSON 块常见变量名是 _dida_config_ patterns [ rwindow\._dida_config_\s*\s*(\{.*?\});, rscript id__NEXT_DATA__ typeapplication/json(.*?)/script, ] for pattern in patterns: m re.search(pattern, text, re.S) if m: try: return json.loads(m.group(1).strip()) except json.JSONDecodeError: continue return Noneimpersonatechrome是curl_cffi的核心用法让整个Session的TLS指纹贴近Chrome。resp.url判断很关键很多反爬不会给你返回403而是静默302到验证页。如果你只看状态码会发现一切都正常但response里根本没有商品数据。加上URL判断能第一时间发现被拦。两个正则对应两种不同的内嵌数据格式。_dida_config_和__NEXT_DATA__是这类SSR页面最常见的两个数据挂载点。如果你抓到的页面里这两个变量都没有打开源码搜索productId手动找到JSON块的起点和终点把正则改一下就行。第一次跑通时建议先print(resp.text[:2000])看结构不要盲改。提取商品ID列表的函数如下def extract_item_ids(page_data): # 根据实际返回结构写多层兼容 try: items page_data[data][root][mods][itemlist][content] except (KeyError, TypeError): items page_data.get(data, {}).get(products, []) ids [] for it in items: pid it.get(productId) or it.get(productIdStr) if pid: ids.append(str(pid)) return ids速卖通不同版本的接口返回结构差别不小。有的版本把商品列表放在itemlist里有的直接叫products。所以这里用try/except写了双层兼容。实际使用中拿到page_data后先print(json.dumps(page_data, indent1)[:2000])看一眼结构确认你的路径写对了再跑全量。3.3 抓取商品详情页从 itemId 到结构化数据def fetch_detail(item_id: str): url https://www.aliexpress.com/aeglodetailweb/api/product/detail params { itemId: item_id, currency: USD, locale: en_US, shipCountry: US, } resp session.get(url, paramsparams, timeout20) if resp.status_code ! 200: return None data resp.json() # 接口成功时 success 为 trueresult.data 里是详情 if not data.get(success) or not data.get(result): return None info data[result][data] return { item_id: item_id, title: info.get(subject), price_min: info.get(skuInfos, {}).get(price), price_max: info.get(skuInfos, {}).get(calPrice), sales: info.get(tradeVolume), rating: info.get(evaluateScore), store: info.get(storeName), detail_url: fhttps://www.aliexpress.com/item/{item_id}.html, }参数说明itemId是必填的必须与列表页拿到的ID一致并且作为字符串传入不要传成整数后面避坑章节会专门讲。currency决定返回价格的币种你存库时用的什么币种这里就要固定否则一张表里美元欧元混在一起后面分析全是错的。locale影响标题的语言版本做中文分析可以试试zh_CN但部分字段在非英文环境下可能缺失我一般统一用en_US保证字段完整。shipCountry影响价格和物流固定成主目标市场即可。返回值里做了字段映射skuInfos里的price是最低价calPrice是划线价或原价tradeVolume是近期销量evaluateScore是评分。这些字段偶有改名的情况拿不到值就打印info.keys()自查不要硬编码死。3.4 数据清洗与 SQLite 落库import sqlite3 def init_db(): con sqlite3.connect(aliexpress.db) con.execute( CREATE TABLE IF NOT EXISTS products ( item_id TEXT PRIMARY KEY, title TEXT, price_min REAL, price_max REAL, sales INTEGER, rating REAL, store TEXT, detail_url TEXT, updated_at TEXT DEFAULT (datetime(now, localtime)) ) ) return con def upsert_product(con, item): con.execute( INSERT INTO products (item_id, title, price_min, price_max, sales, rating, store, detail_url) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(item_id) DO UPDATE SET titleexcluded.title, price_minexcluded.price_min, price_maxexcluded.price_max, salesexcluded.sales, ratingexcluded.rating, storeexcluded.store, updated_atdatetime(now, localtime) , ( item[item_id], item[title], item[price_min], item[price_max], item[sales], item[rating], item[store], item[detail_url], )) con.commit()主键用item_id天然去重。第二次抓到同一条商品时ON CONFLICT分支会自动更新价格、销量、评分和更新时间保留标题与店铺信息。这个设计对价格监控场景很合适你关心的不是历史全量快照而是每条商品最新长什么样。主流程串起来import random import time def main(): con init_db() keywords [dress, watch, headphone] for kw in keywords: for page in range(1, 4): page_data fetch_list(kw, page) if page_data is None: break ids extract_item_ids(page_data) print(f关键词 {kw} 第 {page} 页拿到 {len(ids)} 个商品) for pid in ids: detail fetch_detail(pid) if detail: upsert_product(con, detail) # 每抓一个商品随机等 1~2 秒 time.sleep(random.uniform(0.8, 2.2)) # 翻页之间停久一点降低连续请求特征 time.sleep(random.uniform(3, 8)) if __name__ __main__: main()两个sleep不是摆设而是整套程序能持续跑下去的前提。速卖通的风控不只盯单个请求频率还会计算一段时间内的请求总数和分布。均匀、无间隔的请求序列最容易触发限流而加入随机延时后请求时间分布会接近真实用户被标记的概率大幅下降。4. 速卖通反爬识别机制与持续抓取策略Cookie、签名与换 IP4.1 反爬三板斧会话、节奏与设备特征速卖通的反爬体系我拆成三层来看。第一层是会话校验请求是否携带有效的CookieCookie里的签名令牌是否过期。这一层你没处理好接口直接返回失败。第二层是请求频率单位时间内的请求次数、请求间隔的规律性。如果你每5秒固定请求一次连续一小时不停这种节奏本身就暴露了机器特征。第三层是设备特征TLS指纹、HTTP头顺序、Accept-Language的组合。原生requests的默认请求头太整齐与浏览器差异明显容易在传输层被识别。这三层不是孤立的。你只解决某一层另外两层照样拖后腿。我见过不少案例开发者加了很强的延时策略数据还是拿不到最后一查是TLS指纹的问题。所以务实的做法是三层都做基本对齐不需要做到完美只要比大部分脚本更像浏览器你就比它们都活得久。4.2 让 Session 连贯基础 Cookie 与动态令牌刷新from curl_cffi import requests def create_session(): s requests.Session(impersonatechrome) s.headers.update({ Accept-Language: en-US,en;q0.9, Referer: https://www.aliexpress.com/, }) # 先访问首页让服务端种下基础 Cookie 与签名令牌 s.get(https://www.aliexpress.com/, timeout25) return screate_session做了两件事。第一创建了一个带Chrome指纹的Session对象后续所有请求自动带上第一次访问获得的Cookie。第二先请求一次首页触发服务端下发基础Cookie和签名令牌。这一步很关键如果你直接请求详情接口而没有先访问首页接口大概率返回签名校验失败。这个Session不是永远有效的。签名令牌的时效通常在几十分钟到几小时之间长时间运行的采集任务需要在令牌过期后重新执行一次首页访问来刷新。常见做法是在fetch_detail里捕获签名失败的特征返回然后在主循环里检测到连续失败时重新调用create_session替换全局的session变量。4.3 随机延时与指数退避重试请求过程中超时、断连、签名错误都是常态。一个健壮的抓取程序必须有重试机制但重试不能是无脑环绕。指数退避是最基础的策略第一次失败等2秒第二次等4秒第三次等8秒每次翻倍加上随机抖动。import random import time def request_with_retry(func, max_retries3): for attempt in range(max_retries): try: return func() except Exception: if attempt max_retries - 1: raise # 指数退避 随机抖动 time.sleep(2 ** attempt random.uniform(0, 1))这里有一个额外的细节如果失败原因是签名校验那重试也没有用。所以在实际项目中func内部捕获到签名失败时应当先刷新Session再重试。把刷新动作放在重试之前成功率会好很多。4.4 换 IP 的取舍与正确打开方式不是每个项目都需要换IP。你需要先盘一下自己的数据量级再决定要不要在这上面投入。场景推荐方式说明每天几百条不加IP做好限速控制好延时就够每天几千条数据中心IP池成本低基本够用每天几万条住宅IP池成本高非必要不建议上我的建议是先跑通程序确认业务确实需要更大数据量再考虑IP方案。很多项目卡在第一步就开始买IP结果代码本身还有问题IP换了也是白换。如果确实需要换IP记住几条原则第一单个IP上不要开太多并发5个以内比较保守第二IP失效或者被标记时先停手不要死循环重试第三抓取任务中断是常态记录好断点下次从断点继续而不是从头再来。4.5 验证码出现后的降级策略当你看到滑块验证或者安全验证页时说明你的请求已经被明确标记了。这个时候最省事的方案不是对抗而是让步。具体做法是暂停当前批次的请求打开浏览器手动访问一次目标页面完成验证流程然后把浏览器里的Cookie导出放进你的程序中替换旧的Session。之后再恢复请求时把频率再调低一些比如把随机延时从1秒提升到3秒。不要在同一IP上反复触发验证码那只会让该IP被标记得更久。从长期运行的角度看抓取是一个需要耐心的事数据的可持续性比单次抓取速度重要得多。同时务必控制访问频率尊重平台服务条款抓下来的数据只用于内部分析。5. 速卖通抓取避坑5 个让你数据缺失或断流的典型案例5.1 返回 200 但列表数据为空被安全页替换了响应现象请求返回状态码是200但resp.text里既没有productId也没有内嵌JSON页面内容是大量验证脚本。原因反爬采用的是302重定向加200兜底的组合。你看到的200其实是验证页的200不是目标页的200单看状态码完全发现不了问题。解决在拿到响应后立刻做URL校验检查resp.url是否等于请求URL或者URL中是否出现captcha、security之类的特征。同时把resp.url打进日志排查时直接看日志就能定位。5.2 itemId 没问题但详情接口一直报错参数类型不是数字现象列表页拿到的商品ID确认无误复制到浏览器里也能打开详情页但脚本请求详情接口返回失败。原因速卖通的itemId在详情接口中要求以字符串形式传递。如果你在代码里用整数类型传参某些签名计算逻辑会把参数按字符串拼接导致签名不匹配。另一个坑是商品ID可能以productIdStr的形式返回它的值比productId更长用错字段也会失败。解决统一用str(pid)转成字符串后再传参并优先取productIdStr。我的代码里已经用or做了双字段兼容这个坑在第一次跑通时就会自动避开。5.3 价格字段拿到 None多 SKU 价格藏在多个字段里现象详情接口返回了完整数据但price_min和price_max是None。原因速卖通商品的SKU价格结构并不统一。skuInfos里有一个price字段但部分商品的价格在calPrice部分是rangePrice还有一部分压根只有SKU列表里才有价格。你只取一个字段自然会漏。解决先打印skuInfos的完整结构看清你的目标商品是哪一种价格形态再决定取哪个字段。如果做通用方案就写一个兜底逻辑优先取price没有取calPrice还没有就遍历skuList取第一个有价格的SKU。5.4 凌晨定时任务批量失败动态令牌过期现象白天手动跑程序一切正常凌晨定时任务跑起来后前几十条正常之后全部失败。原因签名令牌有有效期长时间运行后必然过期。凌晨任务往往一跑就是几小时令牌在运行中途过期而你没有任何刷新机制后续请求全部签名失败。解决在请求包装器里捕获签名失败特征一旦出现就刷新Session再重试。刷新动作就是重新请求一次首页拿新Cookie。这个刷新应该作为自动逻辑写进重试机制而不是等任务失败后人工介入。5.5 抓到的商品数量与平台显示不一致分页与排序的坑现象平台页面上显示某个关键词有几千个结果但程序跑完只抓回来一两百条。原因列表接口的分页是有限制的不是所有页面都能逐页翻到底。同时默认排序下部分商品会因为广告位和推荐位占据前几页导致自然结果的实际分布与你预期不同。解决不要依赖默认排序改用明确的排序参数比如按销量或价格排序自然结果更稳定。另外把分页逻辑写成循环直到取不到新商品为止而不是写死页码上限。这样可以避免平台改版导致结果数量变化后你的程序还停在旧页码上。6. 进阶增量更新与异步提速的落地技巧6.1 增量抓取只更新最近变化的商品如果你已经跑通了全量抓取下一步要思考的是怎么避免每天都做全量重抓。速卖通的商品列表是动态的部分商品会下架部分价格会变动每天全量重抓不仅慢还会放大触发风控的概率。更合理的做法是做增量更新。思路是维护一张任务表记录每个item_id第一次出现的时间和最近一次抓取时间CREATE TABLE task_items ( item_id TEXT PRIMARY KEY, keyword TEXT, first_seen TEXT, last_fetch TEXT );每次抓取时只处理两类商品last_fetch距今超过一定时间比如24小时的商品以及从未见过的item_id。其他商品直接跳过。这样你的详情接口请求量会大幅下降同样的IP和Cookie资源能支撑更长期的运行。另一个细节是下架检测。当某个item_id连续多次请求详情返回失败时不要急着删除先在库里标记状态位。因为有些失败是临时性的网络或令牌问题直接删除会导致后续价格趋势分析缺数据。标记3次以上失败再归档是比较稳妥的做法。6.2 用异步替代多线程把吞吐量提上去当你的商品量涨到几千上万条时同步请求加sleep的方案就有点慢了。这时候可以考虑异步改造。httpx配合asyncio是常见的组合比维护线程池省心。import asyncio import httpx SEM asyncio.Semaphore(5) async def fetch_one(client, item_id): async with SEM: resp await client.get( https://www.aliexpress.com/aeglodetailweb/api/product/detail, params{itemId: item_id, currency: USD}, ) return resp.json() async def run(item_ids): async with httpx.AsyncClient(http2True) as client: tasks [fetch_one(client, iid) for iid in item_ids] results await asyncio.gather(*tasks) return results这里的Semaphore(5)是并发上限我建议控制在5到10之间不要贪多。很多人一看到异步就想着并发拉满跑满带宽但在反爬场景下真正的瓶颈不是CPU和网络而是你的IP和Cookie被识别的风险。并发从5提到50数据量并不会线性增长被封的风险却会成倍增加。异步方案配合增量更新可以覆盖到每天几万条数据量的场景。再往上走就要重新评估你的IP资源和合规边界了。我个人吃过一次亏为了追求速度凌晨用50并发跑了一整晚第二天几乎全部请求被验证页拦截之前积累的Cookie和IP全白费。后来我把限速当成业务流程的一部分写进程序里而不是一种妥协抓取反而变得顺了很多。做这个方向先想清楚每天真实需要多少数据再决定跑多快比什么都重要。希望帮到你。本文还有配套的精品资源点击获取
