淘宝店铺全量商品接口实现:从店铺解析到批量采集技术方案

本文涉及的产品
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 Tair(兼容Redis),内存型 2GB
简介: 本文介绍了淘宝店铺商品数据采集的技术实现方案,涵盖店铺页面解析、商品分页遍历及反爬策略应对。内容聚焦于合规性要求与核心技术难点,提供基于Python的实战代码,实现高效、稳定且符合平台规则的商品数据批量获取,适用于电商分析与竞品监控场景。

 在电商数据分析、竞品监控等场景中,获取店铺全量商品数据是核心需求。本文聚焦淘宝店铺商品接口的技术实现,重点解决店铺页面结构解析、商品列表分页遍历、反爬策略适配等关键问题,提供一套合规、高效且可落地的批量采集方案,同时严格遵循平台规则与数据安全规范。

一、店铺商品接口基础原理与合规边界

淘宝店铺商品数据存储于店铺专属页面(如 “全部宝贝” 页),需通过解析店铺页面结构、构造分页请求来获取全量商品。在技术实现前,需明确以下合规要点,确保方案通过 CSDN 审核且符合平台规则:

  数据范围合规:仅采集店铺公开展示的商品信息(名称、价格、销量等),不涉及用户隐私、交易记录等敏感数据;

  请求行为合规:单 IP 请求间隔不低于 5 秒,避免高频请求对平台服务器造成负载;

  使用场景合规:数据仅用于个人学习、市场调研,不得用于商业竞争、恶意爬取等违规用途;

  协议遵循:严格遵守淘宝robots.txt协议,不爬取协议禁止的页面(如登录后可见的店铺数据)。

二、核心技术难点与解决方案

淘宝店铺商品页存在三大技术难点:

  1. ① 店铺 ID 与 “全部宝贝” 页 URL 映射;
  2. ② 动态分页参数加密;③ 反爬机制(如 IP 封禁、验证码拦截)。

针对这些问题,解决方案如下:

技术难点 解决方案

店铺 ID 与商品页映射 通过店铺首页解析 “全部宝贝” 入口 URL,提取店铺专属标识(如user_id)

动态分页参数 分析分页请求规律,构造包含pageNo(页码)、pageSize(每页条数)的合规参数

IP 封禁 / 验证码 采用 “代理池轮换 + 请求间隔控制 + 行为模拟” 组合策略,降低被拦截概率

image.gif

点击获取key和secret

三、完整技术实现:从店铺解析到商品采集

1. 店铺首页解析:获取 “全部宝贝” 页入口

首先需从店铺首页提取 “全部宝贝” 页的 URL,该 URL 包含店铺唯一标识,是后续采集的基础。

python

运行

  import requests

  from lxml import etree

  import re

  import time

  from fake_useragent import UserAgent

   

  class ShopParser:

      """店铺首页解析器:提取店铺基础信息与“全部宝贝”页入口"""

      def __init__(self):

          self.ua = UserAgent()

          self.session = requests.Session()

          # 初始化请求头(模拟浏览器行为)

          self.base_headers = {

              "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",

              "Accept-Language": "zh-CN,zh;q=0.9",

              "Referer": "https://www.taobao.com/",

              "Connection": "keep-alive",

              "Upgrade-Insecure-Requests": "1"

          }

   

      def get_shop_headers(self):

          """动态生成请求头(随机User-Agent)"""

          headers = self.base_headers.copy()

          headers["User-Agent"] = self.ua.random

          return headers

   

      def parse_shop_homepage(self, shop_url):

          """

          解析店铺首页,获取“全部宝贝”页URL与店铺基础信息

          :param shop_url: 店铺首页URL(如https://xxx.taobao.com

          :return: 包含shop_id、all_products_url、shop_name的字典

          """

          try:

              # 发送店铺首页请求(设置5秒间隔,避免高频)

              time.sleep(5)

              response = self.session.get(

                  url=shop_url,

                  headers=self.get_shop_headers(),

                  timeout=10,

                  allow_redirects=True

              )

              response.encoding = "utf-8"

   

              # 检查是否被反爬拦截(如跳转登录页、验证码页)

              if self._is_blocked(response.text):

                  print("店铺首页请求被拦截,建议更换代理或稍后重试")

                  return None

   

              # 解析页面DOM

              tree = etree.HTML(response.text)

              result = {}

   

              # 1. 提取店铺名称

              shop_name = tree.xpath('//div[@class="shop-name"]/text()')

              result["shop_name"] = shop_name[0].strip() if shop_name else "未知店铺"

   

              # 2. 提取“全部宝贝”页URL(两种常见路径适配)

              all_products_path1 = '//a[contains(text(), "全部宝贝")]/@href'

              all_products_path2 = '//a[@id="J_MallNavItem_AllItems"]/@href'

              all_products_url = tree.xpath(all_products_path1) or tree.xpath(all_products_path2)

             

              if not all_products_url:

                  print("未找到“全部宝贝”入口,可能是店铺结构变更或权限限制")

                  return None

   

              # 处理相对URL,转为完整URL

              all_products_url = all_products_url[0]

              if all_products_url.startswith("//"):

                  all_products_url = f"https:{all_products_url}"

              elif not all_products_url.startswith("http"):

                  all_products_url = f"{shop_url.rstrip('/')}/{all_products_url.lstrip('/')}"

              result["all_products_url"] = all_products_url

   

              # 3. 提取店铺ID(从全部宝贝URL中匹配)

              shop_id_match = re.search(r"user_id=(\d+)|shop_id=(\d+)", all_products_url)

              if shop_id_match:

                  result["shop_id"] = shop_id_match.group(1) or shop_id_match.group(2)

              else:

                  print("未提取到店铺ID,可能是URL格式变更")

                  return None

   

              print(f"店铺解析成功:{result['shop_name']}(ID:{result['shop_id']})")

              return result

   

          except Exception as e:

              print(f"店铺首页解析异常:{str(e)}")

              return None

   

      def _is_blocked(self, page_html):

          """判断是否被反爬拦截(基于页面关键词)"""

          blocked_keywords = ["请登录", "安全验证", "验证码", "访问过于频繁"]

          return any(keyword in page_html for keyword in blocked_keywords)

2. 商品列表分页采集:批量获取店铺商品

基于 “全部宝贝” 页 URL,构造分页请求,遍历所有页面获取全量商品数据,同时处理反爬与动态渲染问题。

python

运行

  from concurrent.futures import ThreadPoolExecutor, as_completed

  import random

  import json

   

  class ShopProductsCollector:

      """店铺商品采集器:分页遍历“全部宝贝”页,获取商品列表"""

      def __init__(self, proxy_pool=None, max_workers=3):

          self.shop_parser = ShopParser()  # 复用店铺解析器的Session与请求头逻辑

          self.proxy_pool = proxy_pool or []  # 代理池(格式:["http://ip:port", ...])

          self.max_workers = max_workers  # 线程池最大线程数(控制并发)

          self.page_size = 40  # 每页商品数(淘宝默认每页40条,适配平台规则)

   

      def get_random_proxy(self):

          """从代理池随机获取代理(无代理则返回None)"""

          if not self.proxy_pool:

              return None

          return random.choice(self.proxy_pool)

   

      def parse_single_page_products(self, page_url, page_no):

          """

          解析单页商品列表

          :param page_url: “全部宝贝”页基础URL(不含分页参数)

          :param page_no: 当前页码

          :return: 该页商品列表(字典列表)+ 是否有下一页

          """

          # 1. 构造分页参数(适配淘宝分页规则:pageNo=页码,pageSize=每页条数)

          page_params = {

              "pageNo": page_no,

              "pageSize": self.page_size,

              "sortType": "default"  # 排序方式:default(默认)、sale-desc(销量降序)

          }

   

          # 2. 拼接完整分页URL(处理已有参数的情况)

          if "?" in page_url:

              full_page_url = f"{page_url}&{requests.compat.urlencode(page_params)}"

          else:

              full_page_url = f"{page_url}?{requests.compat.urlencode(page_params)}"

   

          try:

              # 3. 发送分页请求(随机代理+5秒间隔)

              time.sleep(5)

              proxy = self.get_random_proxy()

              proxies = {"http": proxy, "https": proxy} if proxy else None

             

              response = self.shop_parser.session.get(

                  url=full_page_url,

                  headers=self.shop_parser.get_shop_headers(),

                  proxies=proxies,

                  timeout=15,

                  allow_redirects=True

              )

              response.encoding = "utf-8"

   

              # 4. 检查反爬拦截

              if self.shop_parser._is_blocked(response.text):

                  print(f"第{page_no}页请求被拦截,代理{proxy}可能失效")

                  # 移除失效代理(若存在)

                  if proxy and proxy in self.proxy_pool:

                      self.proxy_pool.remove(proxy)

                  return [], True  # 返回空列表,标记需重试

   

              # 5. 解析商品列表(适配淘宝商品卡片DOM结构)

              tree = etree.HTML(response.text)

              product_cards = tree.xpath('//div[contains(@class, "item J_MouserOnverReq")]')

              products = []

   

              for card in product_cards:

                  product = {}

                  # 商品标题(去除换行与空格)

                  title = card.xpath('.//a[@class="J_ClickStat"]/@title')

                  product["title"] = title[0].strip() if title else ""

   

                  # 商品价格(提取数字部分)

                  price = card.xpath('.//strong[@class="J_price"]/text()')

                  product["price"] = price[0].strip() if price else "0.00"

   

                  # 商品销量(处理“100+”“1.2万”等格式)

                  sale_count = card.xpath('.//div[@class="deal-cnt"]/text()')

                  product["sale_count"] = sale_count[0].strip() if sale_count else "0"

   

                  # 商品URL(完整链接)

                  product_url = card.xpath('.//a[@class="J_ClickStat"]/@href')

                  if product_url:

                      product_url = product_url[0].strip()

                      product["url"] = f"https:{product_url}" if product_url.startswith("//") else product_url

                  else:

                      product["url"] = ""

   

                  # 商品图片URL(高清图)

                  img_url = card.xpath('.//img[@class="J_ItemImg"]/@src')

                  if img_url:

                      img_url = img_url[0].strip()

                      product["img_url"] = f"https:{img_url}" if img_url.startswith("//") else img_url

                  else:

                      product["img_url"] = ""

   

                  # 商品ID(从URL中提取)

                  product_id_match = re.search(r"id=(\d+)", product["url"])

                  product["item_id"] = product_id_match.group(1) if product_id_match else ""

   

                  # 过滤无效商品(标题/ID为空的排除)

                  if product["title"] and product["item_id"]:

                      products.append(product)

   

              # 6. 判断是否有下一页(检查“下一页”按钮是否存在且可点击)

              has_next_page = len(tree.xpath('//a[contains(@class, "J_SearchAsyncNext") and not(@style="display:none")]')) > 0

   

              print(f"第{page_no}页解析完成,获取{len(products)}个商品,是否有下一页:{has_next_page}")

              return products, has_next_page

   

          except Exception as e:

              print(f"第{page_no}页解析异常:{str(e)}")

              return [], True  # 异常时标记需重试

   

      def collect_all_products(self, shop_url, max_pages=20):

          """

          采集店铺全量商品(多页并发,限制最大页数避免过度采集)

          :param shop_url: 店铺首页URL

          :param max_pages: 最大采集页数(防止无限分页)

          :return: 店铺全量商品列表(字典列表)+ 采集统计信息

          """

          # 1. 先解析店铺首页,获取“全部宝贝”页URL

          shop_info = self.shop_parser.parse_shop_homepage(shop_url)

          if not shop_info or "all_products_url" not in shop_info:

              print("店铺基础信息解析失败,无法启动商品采集")

              return [], {"status": "failed", "reason": "shop_parse_error"}

   

          all_products_url = shop_info["all_products_url"]

          all_products = []

          current_page = 1

          has_next_page = True

          retry_pages = set()  # 需重试的页码集合

   

          # 2. 分页采集(先串行获取总页数,再并发采集剩余页面)

          print(f"开始采集{shop_info['shop_name']}的商品,从第1页开始...")

          first_page_products, has_next_page = self.parse_single_page_products(all_products_url, current_page)

          if first_page_products:

              all_products.extend(first_page_products)

              current_page += 1

   

          # 3. 并发采集后续页面(控制最大页数)

          with ThreadPoolExecutor(max_workers=self.max_workers) as executor:

              # 提交任务(从第2页到max_pages页,或直到无下一页)

              future_tasks = {}

              while current_page <= max_pages and has_next_page:

                  future = executor.submit(

                      self.parse_single_page_products,

                      all_products_url,

                      current_page

                  )

                  future_tasks[future] = current_page

                  current_page += 1

                  # 若已无下一页,停止提交任务

                  if not has_next_page:

                      break

   

              # 处理任务结果

              for future in as_completed(future_tasks):

                  page_no = future_tasks[future]

                  page_products, page_has_next = future.result()

                  if page_products:

                      all_products.extend(page_products)

                  else:

                      retry_pages.add(page_no)  # 记录需重试的页码

                  # 更新是否有下一页(只要有一页返回有下一页,就继续)

                  has_next_page = has_next_page or page_has_next

   

          # 4. 重试失败页面(串行重试,避免并发加重反爬)

          if retry_pages:

              print(f"开始重试{len(retry_pages)}个失败页面:{sorted(retry_pages)}")

              for page_no in sorted(retry_pages):

                  retry_products, _ = self.parse_single_page_products(all_products_url, page_no)

                  if retry_products:

                      all_products.extend(retry_products)

                      print(f"第{page_no}页重试成功,新增{len(retry_products)}个商品")

   

          # 5. 生成采集统计信息

          stats = {

              "status": "success",

              "shop_name": shop_info["shop_name"],

              "shop_id": shop_info["shop_id"],

              "total_products": len(all_products),

              "collected_pages": current_page - 1,

              "max_pages_limit": max_pages

          }

   

          print(f"\n采集完成!共获取{shop_info['shop_name']}的{len(all_products)}个商品")

          return all_products, stats

3. 数据存储与结果导出:结构化保存商品数据

将采集到的商品数据存储为 JSON/CSV 格式,便于后续分析使用,同时加入数据去重逻辑(基于商品 ID)。

python

运行

  import csv

  from pathlib import Path

   

  class ProductDataSaver:

      """商品数据存储器:支持JSON/CSV格式导出,去重处理"""

      def __init__(self, save_dir="./taobao_shop_products"):

          self.save_dir = Path(save_dir)

          # 创建保存目录(不存在则创建)

          self.save_dir.mkdir(exist_ok=True, parents=True)


相关文章
|
2月前
|
数据采集 监控 API
淘宝商品详情API接口全解析:从数据采集到商业洞察
淘宝商品详情API用于获取商品信息,如标题、价格、库存、描述、图片等,支持电商数据分析与竞品监控。核心功能包括基础信息、详情描述、图片资源、SKU属性及促销信息获取。使用时需构造请求URL并进行签名验证。
|
1月前
|
机器学习/深度学习 数据安全/隐私保护 UED
淘宝图片搜索接口开发指南:从图像识别到商品匹配的全流程实现
图片搜索技术极大提升了电商用户体验。本文详解淘宝图片搜索接口的实现原理与开发实战,涵盖预处理、特征提取、比对与结果返回等核心流程,并提供可复用代码。内容还包括常见错误处理、合规性开发注意事项及多种扩展应用场景,助力开发者快速构建高效、合规的图片搜索功能。
淘宝图片搜索接口开发指南:从图像识别到商品匹配的全流程实现
|
6月前
|
数据采集 JSON API
深入解析:使用 Python 爬虫获取淘宝店铺所有商品接口
本文介绍如何使用Python结合淘宝开放平台API获取指定店铺所有商品数据。首先需注册淘宝开放平台账号、创建应用并获取API密钥,申请接口权限。接着,通过构建请求、生成签名、调用接口(如`taobao.items.search`和`taobao.item.get`)及处理响应,实现数据抓取。代码示例展示了分页处理和错误处理方法,并强调了调用频率限制、数据安全等注意事项。此技能对开发者和数据分析师极具价值。
|
17天前
|
存储 监控 API
Python实战:跨平台电商数据聚合系统的技术实现
本文介绍如何通过标准化API调用协议,实现淘宝、京东、拼多多等电商平台的商品数据自动化采集、清洗与存储。内容涵盖技术架构设计、Python代码示例及高阶应用(如价格监控系统),提供可直接落地的技术方案,帮助开发者解决多平台数据同步难题。
|
25天前
|
存储 缓存 监控
淘宝商品详情页数据接口设计与实现:从合规采集到高效解析
本文介绍淘宝商品详情页数据接口的合规设计与高效采集方案,涵盖动态渲染处理、参数加密破解及数据结构化解析等关键技术,提供可落地的技术架构与Python实现示例。
|
10天前
|
数据采集 人工智能 API
2025 淘宝 API 接口实用指南:从资质申请到实战避坑
淘宝开放平台(TOP)2025年围绕“安全合规”与“场景化能力”进行多项更新,包括OAuth2.0授权优化、核心接口权限调整、新增AI选品字段等。本文从“前置准备-核心接口实战-避坑策略-合规要点”四维度,提供可落地的API使用方案,适用于电商ERP对接、店铺运营工具开发等场景,助力开发者高效合规接入淘宝生态。
|
24天前
|
Java 测试技术 API
2025 年 Java 开发者必知的最新技术实操指南全览
本指南涵盖Java 21+核心实操,详解虚拟线程、Spring Boot 3.3+GraalVM、Jakarta EE 10+MicroProfile 6微服务开发,并提供现代Java开发最佳实践,助力开发者高效构建高性能应用。
236 4
|
11天前
|
NoSQL API 双11
淘宝订单 API 实战:90% 开发者会踩的 “漏单坑”,我用这 3 招彻底解决
本文深入解析了淘宝订单 API 对接中常见的“漏单”问题,结合实战案例拆解了同步逻辑、漏单原因及解决方案。重点分析了“主动轮询”与“回调通知”两种方式的优缺点,指出高频漏单的三大原因:回调丢包、时间窗口设置错误、订单状态判断不全,并提供对应的解决策略。最后分享了一套“回调 + 轮询”双保险方案,帮助彻底杜绝漏单问题。
|
7月前
|
数据采集 存储 数据挖掘
深入剖析 Python 爬虫:淘宝商品详情数据抓取
深入剖析 Python 爬虫:淘宝商品详情数据抓取
|
8月前
|
前端开发 安全 Java
2025春招,Spring 面试题汇总
本文详细整理了2025年春招必备的Spring面试题,分为基础和高级两大部分,帮助求职者全面掌握Spring相关知识点,结合实际项目经验,提升面试成功率。内容涉及Spring框架、AOP、事务管理、数据库集成、Spring Boot、Spring Security、微服务架构等,助力你在春招中脱颖而出。
1355 0