Open‑Claw 实战｜无需逆向，快速搭建电商商品监控与数据分析系统

前言

做电商运营、选品、比价，很多时候我们需要持续盯紧商品的售价、库存、规格、店铺信息。手动复制粘贴做表格效率极低，大促期间价格频繁变动，人工几乎跟不上节奏。

很多同学尝试直接写网页解析，却遇到各种反爬、动态渲染、JS 加密，调试几天拿不到完整的 SKU 规格、原价、发货地等信息。今天分享一套落地思路，借助 Open‑Claw 工具能力，绕开逆向难题，快速实现商品的持续监控，把采集到的数据做清洗统计，完成从数据获取到分析的完整流程。

适合人群：电商运营、数据分析爱好者、Python 开发人员；环境要求 Python3.7+，会基础 Python 语法即可。

一、业务目标梳理

我们想要达成三件事：

单商品完整信息获取：拿到标题、当前售价、划线原价、库存、多规格 SKU、店铺名称、商品主图、发货地等全量资料。
定时持续监控：循环巡检，记录每一次价格、库存快照，捕捉调价、缺货、规格变更。
简单数据分析：保存历史记录，做价格波动统计、规格盘点，发现异常自动打印告警。

不做浏览器模拟、不处理复杂 JS 解密，直接拿到结构化 JSON 结果，减少大量排错成本。

二、环境准备

只需要安装网络请求库，用于发起网络访问，解析返回结果，同时引入时间处理、csv 存储模块用于持久化监控记录。

pip install requests

三、完整实战代码

说明：下面代码中your_key、your_secret替换成你自己的凭证，product_id替换成目标商品编号。逻辑分为三块：获取商品原始数据、解析提取业务字段、定时循环监控、写入本地 CSV 留存历史，异常捕获保证程序不会单次失败直接崩溃。

# coding:utf-8
import requests
import time
import csv
from datetime import datetime

class ProductMonitor:
    def __init__(self, key, secret):
        self.key = key
        self.secret = secret
        self.base_url = "https://api‑gw.onebound.cn/jd/item_get_pro/"
        self.csv_file = "product_monitor_record.csv"
        self.init_csv()

    def init_csv(self):
        """初始化本地存储文件，不存在则写入表头"""
        header = ["crawl_time", "product_id", "title", "current_price",
                  "original_price", "stock", "shop_name", "sku_count"]
        try:
            with open(self.csv_file, "r", encoding="utf‑8‑sig") as f:
                pass
        except FileNotFoundError:
            with open(self.csv_file, "w", newline="", encoding="utf‑8‑sig") as f:
                writer = csv.writer(f)
                writer.writerow(header)

    def fetch_product_raw(self, product_id):
        """获取商品原始结构化数据"""
        params = {
            "key": self.key,
            "secret": self.secret,
            "num_iid": product_id,
            "cache": "no"   # 关闭缓存，获取最新快照
        }
        headers = {
            "Accept‑Encoding": "gzip",
            "Connection": "close"
        }
        try:
            resp = requests.get(self.base_url, params=params, headers=headers, timeout=20)
            return resp.json()
        except Exception as e:
            print(f"网络请求异常：{str(e)}")
            return None

    def parse_product_data(self, raw_json):
        """从原始返回结果解析业务需要字段"""
        if not raw_json:
            return None
        error_code = raw_json.get("error_code")
        if error_code != "0000":
            err_msg = raw_json.get("error", "未知错误")
            print(f"数据获取失败，错误信息：{err_msg}")
            return None
        item = raw_json.get("item", {})
        if not item:
            print("未匹配到商品主体信息")
            return None

        sku_list = item.get("skus", {}).get("sku", [])
        parse_data = {
            "crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
            "product_id": item.get("num_iid"),
            "title": item.get("title", ""),
            "current_price": float(item.get("price", 0)),
            "original_price": float(item.get("orginal_price", 0)),
            "stock": item.get("num", 0),
            "shop_name": item.get("seller_info", {}).get("shop_name", ""),
            "sku_count": len(sku_list),
            "skus_detail": sku_list
        }
        return parse_data

    def save_record(self, data):
        """把单次监控快照写入CSV做历史留存"""
        row = [
            data["crawl_time"],
            data["product_id"],
            data["title"],
            data["current_price"],
            data["original_price"],
            data["stock"],
            data["shop_name"],
            data["sku_count"]
        ]
        with open(self.csv_file, "a", newline="", encoding="utf‑8‑sig") as f:
            writer = csv.writer(f)
            writer.writerow(row)

    def alert_check(self, data, warn_price_threshold):
        """简单告警逻辑：价格低于阈值、库存为0触发提醒"""
        if data["current_price"] <= warn_price_threshold:
            print("‑" * 60)
            print(f"⚠️ 价格预警！商品：{data['title']}")
            print(f"当前售价 {data['current_price']}，低于设定阈值 {warn_price_threshold}")
            print("‑" * 60)
        if data["stock"] <= 0:
            print("‑" * 60)
            print(f"🔴 库存告警！商品 {data['title']} 当前已经无货")
            print("‑" * 60)

    def single_monitor_task(self, product_id, warn_price):
        """执行一次完整监控任务：拉取‑解析‑存储‑告警"""
        raw = self.fetch_product_raw(product_id)
        parsed = self.parse_product_data(raw)
        if parsed:
            self.save_record(parsed)
            self.alert_check(parsed, warn_price)
            print(f"[{parsed['crawl_time']}] 监控完成｜{parsed['title'][:35]}...｜现价:{parsed['current_price']}｜库存:{parsed['stock']}｜规格数:{parsed['sku_count']}")
        return parsed

if __name__ == "__main__":
    # ========== 配置区域，请修改为你的信息 ==========
    MY_KEY = "your_key"
    MY_SECRET = "your_secret"
    MONITOR_PRODUCT_ID = "10335871600"  # 商品编号
    PRICE_WARN_THRESHOLD = 120  # 低于该价格触发提醒
    MONITOR_INTERVAL = 60 * 30  # 监控间隔，单位秒，示例30分钟巡检一次
    # =============================================

    monitor = ProductMonitor(MY_KEY, MY_SECRET)
    print("===== 商品监控程序已启动 =====")
    print(f"监控商品ID:{MONITOR_PRODUCT_ID}，价格预警阈值:{PRICE_WARN_THRESHOLD}，轮询间隔:{MONITOR_INTERVAL}秒")
    while True:
        monitor.single_monitor_task(MONITOR_PRODUCT_ID, PRICE_WARN_THRESHOLD)
        time.sleep(MONITOR_INTERVAL)

四、代码功能说明

持久化存储：每次巡检把时间戳、价格、库存、店铺信息写入product_monitor_record.csv，Excel 可以直接打开查看历史轨迹。
异常容错：网络波动、商品下架会打印错误信息，不会直接让整个程序崩溃。
双重告警：价格低于预设阈值、库存归零，控制台输出醒目的提醒。
SKU 完整获取：返回结果中自带全部规格列表，可以扩展做规格级别的监控。

注意：轮询间隔不要设置过小，大促场景建议至少 10 分钟以上，避免高频访问造成限制。

五、拿到历史数据之后，怎么做数据分析

采集得到 CSV 历史记录文件之后，我们可以用 Pandas 做简单的统计分析，下面补充一小段数据分析脚本：

import pandas as pd

def analysis_history():
    df = pd.read_csv("product_monitor_record.csv", encoding="utf‑8‑sig")
    print("====历史数据概览====")
    print(f"总采集记录数：{len(df)}")
    print(f"价格最大值：{df['current_price'].max()}")
    print(f"价格最小值：{df['current_price'].min()}")
    print(f"平均售价：{round(df['current_price'].mean(),2)}")
    print("\n最近10条快照：")
    print(df.tail(10)[["crawl_time","current_price","stock"]])

if __name__ == "__main__":
    analysis_history()

执行后可以直接看到价格最高、最低、均值，快速判断商品的价格波动区间，辅助我们做采购、选品、竞品对标。如果需要更进一步，可以接入 Matplotlib 画出价格时序走势图。

六、常见踩坑点与优化方向

缓存干扰：代码中设置cache=no，保证拿到实时最新快照；不关闭缓存会返回历史旧数据，监控就失去意义。
商品下架情况：当商品已经删除、下架，返回结果会提示找不到商品，业务上可以增加逻辑，一旦连续多次返回不存在，记录到日志，停止监控。
多商品批量监控：可以维护一个商品 ID 列表，循环遍历列表，实现批量多款商品同时巡检。
告警渠道扩展：目前只是控制台打印，可扩展对接企业微信、邮件推送，做到无人值守通知。
规格粒度监控：当前示例取的商品主体价格，业务需要可以遍历skus_detail列表，针对每一个子规格单独记录价格库存。

七、小结

借助 Open‑Claw，我们不需要耗费大量精力处理网页逆向、JS 解密、验证码等麻烦工作，就能够稳定拿到完整结构化的商品信息。

整套流程分为：获取结构化原始数据 → 清洗解析业务字段 → 定时循环采集、落地存储快照 → 统计分析历史数据。对于商家、个人比价选品都非常实用。

万邦api博客

Nice to meet you, too!

api 数据挖掘