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

 

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

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

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

数据范围合规：仅采集店铺公开展示的商品信息（名称、价格、销量等），不涉及用户隐私、交易记录等敏感数据；
   请求行为合规：单 IP 请求间隔不低于 5 秒，避免高频请求对平台服务器造成负载；
   使用场景合规：数据仅用于个人学习、市场调研，不得用于商业竞争、恶意爬取等违规用途；
   协议遵循：严格遵守淘宝robots.txt协议，不爬取协议禁止的页面（如登录后可见的店铺数据）。

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

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

① 店铺 ID 与 “全部宝贝” 页 URL 映射；

② 动态分页参数加密；③ 反爬机制（如 IP 封禁、验证码拦截）。

针对这些问题，解决方案如下：

技术难点 解决方案
店铺 ID 与商品页映射 通过店铺首页解析 “全部宝贝” 入口 URL，提取店铺专属标识（如user_id）
动态分页参数 分析分页请求规律，构造包含pageNo（页码）、pageSize（每页条数）的合规参数
IP 封禁 / 验证码 采用 “代理池轮换 + 请求间隔控制 + 行为模拟” 组合策略，降低被拦截概率

 

点击获取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)

 

审核编辑 黄宇