在电商行业，竞品价格监控是运营决策的核心依据之一。作为爬虫工程师，单纯依赖 API 或爬虫都存在局限：API 数据规范但可能缺失部分字段，爬虫灵活但易触发反爬。本文将带你实战 “API + 爬虫” 组合方案，高效爬取竞品商品数据并生成可视化价格对比表，全程附可运行代码。

一、前置准备：工具与环境配置

1.1 核心技术栈

选择轻量且高效的技术组合，确保兼容性与可维护性：

• 开发语言：Python 3.8+（推荐 3.10，避免依赖库版本冲突）

• API 请求：requests（处理 HTTP 请求）、pycryptodome（部分平台 API 签名加密）

• 爬虫解析：BeautifulSoup4（静态页面解析）、selenium（动态页面渲染，可选）

• 数据处理：pandas（数据清洗与整合）

• 表格生成：openpyxl（Excel 格式输出，支持样式定制）

• 定时任务（可选）：schedule（实现周期性爬取）

1.2 环境安装

通过 pip 一键安装依赖：

pip install requests==2.31.0 beautifulsoup4==4.12.2 pandas==2.0.3 openpyxl==3.1.2 pycryptodome==3.20.0 schedule==1.2.0 selenium==4.15.2

1.3 关键资源准备

        电商 API 密钥：以京东开放平台为例（其他平台流程类似）

• 登录京东开放平台开发者中心，创建应用并申请 “商品查询 API” 权限
• 记录appkey、appsecret（核心密钥，需加密存储，避免硬编码
  
  反爬资源：
   
• IP 代理池（推荐阿布云、芝麻代理，应对 IP 封禁）
• User-Agent 池（模拟不同浏览器，减少特征识别）
• Cookie 池（针对需要登录的平台，可选）

二、核心实现：API + 爬虫协同爬取

2.1 第一步：API 调用爬取规范数据

以京东 “商品详情查询 API” 为例，重点解决签名验证与频率控制问题。

2.1.1 API 请求逻辑（含签名生成）

京东 API 要求按 “参数排序 + MD5 加密” 生成签名，核心代码如下：

import requests
import time
import hashlib
import json

class JDAPI:
    def __init__(self, appkey, appsecret):
        self.appkey = appkey
        self.appsecret = appsecret
        self.base_url = "https://api.jd.com/routerjson"  # 京东API网关
    
    def generate_sign(self, params):
        """生成API签名（按京东规则：参数ASCII排序+拼接appsecret+MD5加密）"""
        # 1. 排除sign参数，按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接为"key=value&key=value"格式
        param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
        # 3. 末尾拼接appsecret，MD5加密后转大写
        sign_str = param_str + self.appsecret
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    
    def get_product_detail(self, sku_id):
        """获取单个商品详情（sku_id为京东商品唯一标识）"""
        params = {
            "method": "jingdong.product.get",  # API接口名
            "app_key": self.appkey,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "format": "json",
            "v": "2.0",
            "skuId": sku_id,  # 目标商品SKU
            "sign_method": "md5"
        }
        # 生成签名并添加到参数
        params["sign"] = self.generate_sign(params)
        
        try:
            # 控制请求频率（京东API默认QPS=2，避免429错误）
            time.sleep(0.5)
            response = requests.get(self.base_url, params=params, timeout=10)
            response.raise_for_status()  # 触发HTTP错误（如403、500）
            data = response.json()
            
            # 提取核心字段（按需求调整）
            if "product" in data["jingdong_product_get_response"]:
                product = data["jingdong_product_get_response"]["product"]
                return {
                    "source": "JD_API",  # 数据来源标识
                    "sku_id": sku_id,
                    "name": product.get("name", ""),  # 商品名称
                    "price": product.get("jdPrice", {}).get("price", 0),  # 京东价
                    "market_price": product.get("marketPrice", 0),  # 市场价
                    "store_name": product.get("shopName", ""),  # 店铺名称
                    "stock": product.get("stockState", 0)  # 库存状态（0=无货，1=有货）
                }
            else:
                print(f"SKU {sku_id} 无数据")
                return None
        except Exception as e:
            print(f"API请求失败（SKU: {sku_id}）：{str(e)}")
            return None

# 初始化API（替换为你的appkey和appsecret）
jd_api = JDAPI(appkey="你的京东appkey", appsecret="你的京东appsecret")
# 测试获取单个商品数据
test_sku = "100012345678"  # 替换为目标竞品SKU
api_data = jd_api.get_product_detail(test_sku)
print("API获取数据：", api_data)

2.1.2 API 调用避坑指南

1. 签名错误：确保参数排序、编码格式（UTF-8）、appsecret 正确性；
2. 频率限制：通过time.sleep()控制 QPS，或使用 API 提供的 “流量控制接口” 查询剩余配额；
3. 权限不足：确认申请的 API 权限包含 “价格”“库存” 等敏感字段（部分字段需额外审核）。

2.2 第二步：爬虫补充 API 缺失数据

部分平台（如小众电商）无开放 API，或 API 不返回 “促销标签”“用户评价数” 等字段，需用爬虫补充。以 “某电商商品列表页” 为例，讲解静态页面解析与反爬策略。

2.2.1 爬虫核心代码（含反爬）

import requests
from bs4 import BeautifulSoup
import random
from fake_useragent import UserAgent  # 需额外安装：pip install fake-useragent

class ProductSpider:
    def __init__(self, proxy_pool=None):
        self.ua = UserAgent()  # 自动生成随机User-Agent
        self.proxy_pool = proxy_pool or [
            "http://123.45.67.89:8080",
            "http://98.76.54.32:8888"  # 替换为你的代理IP
        ]
    
    def get_random_proxy(self):
        """从代理池随机选择代理"""
        return {"http": random.choice(self.proxy_pool), "https": random.choice(self.proxy_pool)}
    
    def get_page_html(self, url):
        """获取页面HTML（带反爬策略）"""
        headers = {
            "User-Agent": self.ua.random,
            "Cookie": "你的Cookie（可选，登录后获取）",
            "Referer": "https://www.target-platform.com/"  # 模拟真实来源
        }
        
        try:
            # 随机代理+延迟（避免反爬）
            proxy = self.get_random_proxy()
            time.sleep(random.uniform(1, 3))  # 1-3秒随机延迟
            response = requests.get(
                url,
                headers=headers,
                proxies=proxy,
                timeout=15,
                verify=False  # 忽略SSL验证（部分代理需开启）
            )
            response.encoding = response.apparent_encoding  # 自动识别编码
            return response.text
        except Exception as e:
            print(f"页面请求失败（URL: {url}）：{str(e)}")
            return None
    
    def parse_product_list(self, html):
        """解析商品列表页，提取商品数据"""
        if not html:
            return []
        
        soup = BeautifulSoup(html, "lxml")
        product_items = soup.find_all("div", class_="product-item")  # 按页面结构调整选择器
        product_list = []
        
        for item in product_items:
            try:
                # 提取字段（需按目标页面HTML结构调整）
                sku_id = item.get("data-sku", "")
                name = item.find("div", class_="product-name").get_text(strip=True)
                price = float(item.find("span", class_="price").get_text(strip=True).replace("¥", ""))
                store_name = item.find("div", class_="store-name").get_text(strip=True)
                # 补充API缺失的字段（如促销标签）
                promotion = item.find("span", class_="promotion-tag").get_text(strip=True) if item.find("span", class_="promotion-tag") else "无"
                
                product_list.append({
                    "source": "Spider",
                    "sku_id": sku_id,
                    "name": name,
                    "price": price,
                    "market_price": 0,  # 若页面无市场价，可设为0或通过其他接口补充
                    "store_name": store_name,
                    "stock": 1,  # 简化处理：有商品即视为有货
                    "promotion": promotion  # API缺失的促销字段
                })
            except Exception as e:
                print(f"解析商品失败：{str(e)}")
                continue
        
        return product_list

# 初始化爬虫（替换为你的代理池）
spider = ProductSpider(proxy_pool=["http://你的代理IP1:端口", "http://你的代理IP2:端口"])
# 目标商品列表页URL（替换为实际URL）
target_url = "https://www.target-platform.com/category/123?page=1"
html = spider.get_page_html(target_url)
spider_data = spider.parse_product_list(html)
print("爬虫获取数据：", spider_data[:2])  # 打印前2条数据

2.2.2 爬虫反爬关键策略

1. 动态 User-Agent：用fake-useragent避免固定 UA 被识别；
2. IP 代理池：定期更换代理，应对 “单 IP 高频请求封禁”；
3. 随机延迟：避免请求间隔一致，模拟人工浏览；
4. 动态页面处理：若页面用 JS 渲染（如 Vue/React），可改用selenium或抓包获取 XHR 接口（F12→Network→XHR，直接请求真实数据接口效率更高）。

三、数据整合：清洗与统一格式

API 与爬虫获取的数据格式不一致（如字段名、数据类型），需用pandas统一清洗。

3.1 数据整合核心代码

import pandas as pd

def integrate_data(api_data_list, spider_data_list):
    """
    整合API与爬虫数据
    :param api_data_list: API获取的商品列表（list of dict）
    :param spider_data_list: 爬虫获取的商品列表（list of dict）
    :return: 清洗后的DataFrame
    """
    # 1. 转换为DataFrame
    df_api = pd.DataFrame(api_data_list)
    df_spider = pd.DataFrame(spider_data_list)
    
    # 2. 统一字段名（按需求调整，确保两表字段一致）
    # 例：爬虫数据的"promotion"字段，API数据无，补充为"无"
    if "promotion" not in df_api.columns:
        df_api["promotion"] = "无"
    # 例：API数据的"market_price"字段，爬虫数据无，补充为0
    if "market_price" not in df_spider.columns:
        df_spider["market_price"] = 0
    
    # 3. 合并数据（按sku_id去重，保留最新数据）
    df_combined = pd.concat([df_api, df_spider], ignore_index=True)
    # 按sku_id分组，保留"price"最新的一条（假设price更新时间越近越小）
    df_combined = df_combined.sort_values("price").drop_duplicates("sku_id", keep="first")
    
    # 4. 数据清洗（处理缺失值、异常值）
    df_combined["price"] = df_combined["price"].fillna(0)  # 价格缺失填充为0
    df_combined["name"] = df_combined["name"].str.slice(0, 50)  # 商品名称截断（避免Excel列过宽）
    df_combined = df_combined[df_combined["sku_id"] != ""]  # 过滤无SKU的无效数据
    
    # 5. 调整列顺序（按业务优先级）
    column_order = ["sku_id", "name", "price", "market_price", "store_name", "promotion", "stock", "source"]
    df_combined = df_combined[column_order]
    
    return df_combined

# 模拟多商品数据（实际场景中可循环调用API/爬虫获取）
api_data_list = [jd_api.get_product_detail(sku) for sku in ["100012345678", "100012345679"]]
spider_data_list = spider.parse_product_list(spider.get_page_html(target_url))

# 整合数据
df_clean = integrate_data(api_data_list, spider_data_list)
print("清洗后数据：")
print(df_clean.head())

四、生成价格对比表：Excel 可视化

用openpyxl生成带样式的 Excel 表格，便于运营直接使用。

4.1 表格生成核心代码

from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.utils.dataframe import dataframe_to_rows

def generate_price_table(df, output_path="竞品价格对比表.xlsx"):
    """
    生成带样式的价格对比Excel表
    :param df: 清洗后的DataFrame
    :param output_path: 输出路径
    """
    # 1. 创建工作簿
    wb = Workbook()
    ws = wb.active
    ws.title = "竞品价格表"
    
    # 2. 写入数据（含表头）
    for r_idx, row in enumerate(dataframe_to_rows(df, index=False, header=True), 1):
        for c_idx, value in enumerate(row, 1):
            cell = ws.cell(row=r_idx, column=c_idx, value=value)
    
    # 3. 样式定制（提升可读性）
    # 表头样式：加粗、蓝色背景
    header_font = Font(bold=True, color="FFFFFF")
    header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
    for c_idx in range(1, len(df.columns) + 1):
        header_cell = ws.cell(row=1, column=c_idx)
        header_cell.font = header_font
        header_cell.fill = header_fill
        header_cell.alignment = Alignment(horizontal="center")
    
    # 价格列样式：红色字体（突出核心字段）
    price_font = Font(color="FF0000")
    price_column = df.columns.get_loc("price") + 1  # 转换为Excel列号（1开始）
    for r_idx in range(2, len(df) + 2):
        price_cell = ws.cell(row=r_idx, column=price_column)
        price_cell.font = price_font
        price_cell.alignment = Alignment(horizontal="center")
    
    # 调整列宽
    column_widths = [15, 50, 12, 12, 20, 15, 10, 10]  # 按字段长度设置
    for c_idx, width in enumerate(column_widths, 1):
        ws.column_dimensions[chr(64 + c_idx)].width = width  # chr(65)=A列
    
    # 4. 保存文件
    wb.save(output_path)
    print(f"价格对比表已生成：{output_path}")

# 生成表格
generate_price_table(df_clean)

五、进阶优化：自动化与扩展

5.1 定时自动爬取

用schedule库实现每日固定时间爬取，示例代码：

import schedule

def daily_crawl():
    """每日爬取任务"""
    print(f"开始每日爬取：{time.strftime('%Y-%m-%d %H:%M:%S')}")
    # 1. 调用API获取数据
    api_skus = ["100012345678", "100012345679"]  # 可从配置文件读取
    api_data_list = [jd_api.get_product_detail(sku) for sku in api_skus]
    # 2. 调用爬虫获取数据
    spider_urls = ["https://www.target-platform.com/category/123?page=1", "https://www.target-platform.com/category/123?page=2"]
    spider_data_list = []
    for url in spider_urls:
        html = spider.get_page_html(url)
        spider_data_list.extend(spider.parse_product_list(html))
    # 3. 整合数据并生成表格
    df_clean = integrate_data(api_data_list, spider_data_list)
    generate_price_table(df_clean, output_path=f"竞品价格对比表_{time.strftime('%Y%m%d')}.xlsx")
    print(f"每日爬取完成：{time.strftime('%Y-%m-%d %H:%M:%S')}")

# 设定每日9:00自动爬取
schedule.every().day.at("09:00").do(daily_crawl)

# 持续运行
while True:
    schedule.run_pending()
    time.sleep(60)  # 每分钟检查一次任务

5.2 扩展方向

1. 数据存储：将历史数据存入 MySQL/MongoDB，支持价格趋势分析；
2. 可视化：用matplotlib/Tableau生成价格波动图表；
3. 告警机制：当竞品价格低于阈值时，通过邮件 / 企业微信推送告警；
4. 多平台支持：扩展淘宝、拼多多等平台的 API 与爬虫（核心逻辑一致，仅需调整参数与解析规则）。

六、合规与风险提示

        遵守平台规则：

• 严格遵循电商平台robots.txt协议（如京东）；
• API 调用需遵守平台《开发者协议》，不得超量请求或滥用数据；
  
  数据用途合规：爬取的数据仅用于企业内部分析，不得泄露或用于商业竞争；
  反爬适度：避免使用暴力爬取（如多线程无延迟），防止影响平台服务器稳定。

总结

本文通过 “API + 爬虫” 组合方案，解决了电商竞品数据爬取的 “完整性” 与 “效率” 问题。核心逻辑可复用于不同电商平台，只需调整 API 参数与爬虫解析规则。后续可结合数据库与可视化工具，构建完整的竞品价格监控系统。