在电商数据分析、竞品监控等场景中,获取店铺全量商品数据是核心需求。本文聚焦淘宝店铺商品接口的技术实现,重点解决店铺页面结构解析、商品列表分页遍历、反爬策略适配等关键问题,提供一套合规、高效且可落地的批量采集方案,同时严格遵循平台规则与数据安全规范。
一、店铺商品接口基础原理与合规边界
淘宝店铺商品数据存储于店铺专属页面(如 “全部宝贝” 页),需通过解析店铺页面结构、构造分页请求来获取全量商品。在技术实现前,需明确以下合规要点,确保方案通过 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)