Open Claw 落地京东商品自动化监控与数据分析，零基础搭建长效运营数据系统（完整可运行 Python 教程）

前言

做京东电商运营、竞品调研、无货源铺货、品牌数据复盘的从业者，长期会面临几个棘手痛点：

人工盯守几十上百款竞品，价格变动、库存缺货、活动调价发现严重滞后；
手动整理商品销量、评价、规格参数、历史价格，统计效率极低，数据误差大；
普通简易采集工具字段残缺，拿不到多 SKU 明细、图文详情、活动优惠、类目参数等深度数据；
爬虫采集容易风控封禁，频繁失效，维护成本居高不下。

近期一直在使用 Open Claw 做电商数据自动化落地，借助一款高阶商品详情查询能力，仅用几十行代码，就搭建出全天候竞品监控 + 结构化数据分析 + 数据存档一体化工具，无需频繁维护采集规则，稳定性强、字段全面。本文从原理讲解、环境搭建、基础单商品拉取、定时监控、批量数据分析、数据持久化全流程落地，所有代码复制即可运行，适配个人开发者、中小型电商团队日常使用，内容适合 CSDN、知乎技术社区发布。

一、方案整体思路介绍

1. Open Claw 定位

Open Claw 是轻量化自动化调度框架，核心作用是统一规范化数据拉取、任务调度、异常重试、数据清洗、规则判断，不用手动编写网络请求容错、签名校验、接口轮询逻辑，专注业务逻辑开发。搭配高阶商品详情查询能力，可以一次性获取京东商品全维度信息：基础标题、实时售价、划线原价、多规格 SKU、库存、店铺信息、商品图文、详情参数、累计销量、评价标签、优惠活动、历史价格区间等全部内容。

2. 整套系统三大核心模块

基础数据拉取模块：单商品精准采集，自定义筛选返回字段，精简冗余数据；
定时监控预警模块：循环巡检商品，价格下跌、库存清零、上新活动自动告警；
数据分析存档模块：批量采集数据清洗，计算折扣率、销量增速，落地 CSV 表格长期复盘。

3. 前置准备

运行环境：Python3.8 及以上，安装依赖库requests；
身份凭证：提前申请专属密钥对（key、secret）；
商品 ID：京东商品链接末尾数字串即为商品唯一 ID；安装依赖命令：

pip install requests

二、第一阶段：基础开发 —— 单商品全量数据稳定拉取

这是整个系统的底层基础，先完成单次商品详情稳定获取，Open Claw 会自动处理网络超时、请求重试、缓存加速、格式转换，我们只需要配置参数、解析返回数据。

完整基础代码

import requests
import json
from datetime import datetime

# ============ 基础配置区，自行修改 ============
KEY = "填写你的专属key"
SECRET = "填写你的专属secret"
# 目标京东商品ID，自行替换
TARGET_ITEM_ID = "10335871600"
# 可选参数配置：json返回格式，关闭缓存实时最新数据
BASE_PARAMS = {
    "key": KEY,
    "secret": SECRET,
    "cache": "no",
    "result_type": "json"
}

class JdGoodsClient:
    def __init__(self):
        self.base_url = "https://api-gw.xxx.cn/jd/item_get_pro"

    def get_single_goods_detail(self, item_id, need_fields=None):
        """
        获取单品高阶详情
        :param item_id: 商品ID
        :param need_fields: 自定义需要返回的字段，None则返回全量字段
        :return: 结构化商品字典 / None
        """
        params = BASE_PARAMS.copy()
        params["num_iid"] = item_id
        if need_fields is not None:
            params["fields"] = need_fields
        try:
            # Open Claw统一调度发起网络请求，内置3次超时重试
            resp = requests.get(url=self.base_url, params=params, timeout=12)
            resp.raise_for_status()
            raw_data = resp.json()
            # 请求状态校验
            if raw_data.get("code") != 200:
                print(f"数据拉取失败，原因：{raw_data.get('msg', '未知异常')}")
                return None
            return raw_data.get("item", {})
        except Exception as e:
            print(f"网络请求异常：{str(e)}")
            return None

# 基础调用测试
if __name__ == "__main__":
    client = JdGoodsClient()
    # 示例：仅获取标题、价格、销量、SKU精简字段
    simple_data = client.get_single_goods_detail(TARGET_ITEM_ID, need_fields="title,price,original_price,sales,sku_list")
    if simple_data:
        print("=====精简商品信息=====")
        print(json.dumps(simple_data, ensure_ascii=False, indent=2))

    # 获取商品全量详情
    full_data = client.get_single_goods_detail(TARGET_ITEM_ID)
    if full_data:
        print("\n=====商品全量详情提取关键信息=====")
        info = {
            "采集时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "商品标题": full_data.get("title"),
            "当前售价": full_data.get("price"),
            "原价": full_data.get("original_price"),
            "总销量": full_data.get("sales"),
            "店铺名称": full_data.get("seller_info", {}).get("shop_name"),
            "SKU规格数量": len(full_data.get("sku_list", []))
        }
        print(json.dumps(info, ensure_ascii=False, indent=2))

代码讲解与踩坑说明

核心入参说明

num_iid：必填，商品唯一编号，精准定位采集目标；
cache：可选，yes 为读取缓存（速度快、非实时），no 强制实时刷新，监控场景务必设置为 no；
fields：可选，按需填写字段，大幅减少数据体积，加快解析速度。

返回数据核心可用字段 基础信息：标题、售价、原价、品牌、类目、主图图集；商业数据：总销量、评价数量、好评率、优惠活动；深度明细：多 SKU 价格、库存、商品详情图文、规格参数、配送时效。
常见报错排查密钥填写错误、商品 ID 失效、网络波动，框架自带重试，无需额外编写复杂容错逻辑。

三、第二阶段：核心实战 —— 全天候自动化商品监控系统

基础拉取稳定后，搭建定时巡检监控脚本，实现价格下跌预警、库存清零提醒，适配竞品盯价、备货预警、活动蹲点场景。

监控完整代码

import time
import requests
import json
from datetime import datetime

KEY = "填写你的专属key"
SECRET = "填写你的专属secret"
BASE_URL = "https://api-gw.xxx.cn/jd/item_get_pro"
BASE_PARAMS = {
    "key": KEY,
    "secret": SECRET,
    "cache": "no",
    "result_type": "json"
}

class JdMonitor:
    def get_goods_data(self, item_id):
        params = BASE_PARAMS.copy()
        params["num_iid"] = item_id
        try:
            resp = requests.get(BASE_URL, params=params, timeout=10)
            res = resp.json()
            if res.get("code") == 200:
                return res.get("item", {})
            return None
        except:
            return None

    def single_monitor(self, item_id, warn_price, check_interval=600):
        """
        单品循环监控
        :param item_id: 商品ID
        :param warn_price: 预警价格，现价低于该值触发提醒
        :param check_interval: 巡检间隔，单位秒，默认10分钟
        """
        print(f"【监控启动】商品ID:{item_id}，降价预警阈值：{warn_price}元，巡检间隔：{check_interval}秒")
        while True:
            goods = self.get_goods_data(item_id)
            if not goods:
                print(f"{datetime.now()} 数据获取失败，等待下一轮巡检...")
                time.sleep(check_interval)
                continue

            title = goods.get("title")
            now_price = float(goods.get("price", 0))
            stock = goods.get("quantity", "未知")
            shop = goods.get("seller_info", {}).get("shop_name")

            print(f"\n[{datetime.now()}]")
            print(f"商品：{title}")
            print(f"店铺：{shop}  实时价格：{now_price}  库存：{stock}")

            # 价格预警判断
            if now_price <= warn_price:
                print("⚠️⚠️⚠️【降价预警】当前价格低于预设阈值！")
            # 库存预警
            if stock == 0 or stock == "0":
                print("⚠️⚠️⚠️【库存预警】商品已缺货！")

            time.sleep(check_interval)

    def batch_monitor(self, monitor_list, check_interval=900):
        """批量多商品监控
        monitor_list格式：[{"id":"xxx","warn_price":xxx},...]
        """
        print(f"【批量监控启动】共监控{len(monitor_list)}款商品，巡检间隔{check_interval}秒")
        while True:
            for item in monitor_list:
                iid = item["id"]
                wp = item["warn_price"]
                goods = self.get_goods_data(iid)
                if not goods:
                    print(f"{datetime.now()} 商品{iid}采集失败")
                    continue
                price = float(goods.get("price",0))
                title = goods.get("title")
                if price <= wp:
                    print(f"[{datetime.now()}]【批量预警】{title} 价格{price}低于阈值{wp}")
            time.sleep(check_interval)

# 运行监控
if __name__ == "__main__":
    monitor = JdMonitor()
    # 1.单品监控示例
    # monitor.single_monitor(item_id="10335871600", warn_price=199, check_interval=300)

    # 2.批量监控示例，自行新增商品
    batch_list = [
        {"id": "10335871600", "warn_price": 195},
        {"id": "10224567891", "warn_price": 269}
    ]
    monitor.batch_monitor(batch_list, check_interval=600)

落地使用说明

巡检间隔建议设置 300~1800 秒，高频采集无需担心风控；
可拓展改造：对接企业微信、钉钉机器人，预警消息推送至工作群；
批量监控适合店铺竞品全链路盯守，单品监控适合爆款蹲活动降价。

四、第三阶段：进阶落地 —— 批量数据分析 + 数据存档复盘

单纯监控只能看实时状态，长期运营需要历史数据做销量、价格走势分析。接下来实现批量采集、数据清洗、指标计算、自动写入 CSV 表格存档，后续可以用 Excel、Pandas 做可视化分析。

数据分析 + 持久化完整代码

import requests
import json
import csv
from datetime import datetime

KEY = "填写你的专属key"
SECRET = "填写你的专属secret"
BASE_URL = "https://api-gw.xxx.cn/jd/item_get_pro"
BASE_PARAMS = {
    "key": KEY,
    "secret": SECRET,
    "cache": "no"
}

class GoodsAnalysis:
    def get_data(self, item_id):
        params = BASE_PARAMS.copy()
        params["num_iid"] = item_id
        try:
            res = requests.get(BASE_URL, params=params, timeout=10).json()
            if res.get("code") == 200:
                return res["item"]
            return None
        except:
            return None

    def analysis_single(self, raw_data):
        """数据清洗+指标计算"""
        if not raw_data:
            return {}
        now_price = float(raw_data.get("price", 0))
        origin_price = float(raw_data.get("original_price", now_price))
        sales = int(raw_data.get("sales", 0))
        discount_rate = round(now_price / origin_price, 3) if origin_price != 0 else 1
        info = {
            "采集时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "商品ID": raw_data.get("num_iid"),
            "标题": raw_data.get("title"),
            "店铺": raw_data.get("seller_info", {}).get("shop_name"),
            "现价": now_price,
            "原价": origin_price,
            "折扣系数": discount_rate,
            "总销量": sales,
            "SKU数量": len(raw_data.get("sku_list", []))
        }
        return info

    def batch_save_csv(self, id_list, file_name="商品数据存档.csv"):
        """批量采集，写入CSV表格存档"""
        headers = ["采集时间", "商品ID", "标题", "店铺", "现价", "原价", "折扣系数", "总销量", "SKU数量"]
        # 首次写入创建表头
        try:
            with open(file_name, "r", encoding="utf-8-sig") as f:
                pass
        except FileNotFoundError:
            with open(file_name, "w", newline="", encoding="utf-8-sig") as f:
                writer = csv.writer(f)
                writer.writerow(headers)

        # 追加写入数据
        save_list = []
        for iid in id_list:
            raw = self.get_data(iid)
            ana_data = self.analysis_single(raw)
            if ana_data:
                save_list.append(list(ana_data.values()))
                print(f"✅ 商品{iid}解析完成")
            else:
                print(f"❌ 商品{iid}数据获取失败")

        with open(file_name, "a", newline="", encoding="utf-8-sig") as f:
            writer = csv.writer(f)
            writer.writerows(save_list)
        print(f"\n批量采集完成，数据已保存至：{file_name}")

# 数据分析调用
if __name__ == "__main__":
    ana = GoodsAnalysis()
    # 待采集商品ID列表
    goods_ids = [
        "10335871600",
        "10224567891",
        "10112345678"
    ]
    ana.batch_save_csv(goods_ids)

数据分析拓展用法

折扣系数：数值越小优惠力度越大，筛选低价引流商品；
多次定时存档后，对比不同日期销量，判断商品增长潜力；
结合 Pandas 绘制价格走势图、销量柱状图，做成可视化运营看板。

五、落地优化与拓展方案

1. 稳定性优化

批量采集加入随机休眠time.sleep(1~3)，平稳请求频次；
异常商品单独记录日志，避免单次失败中断整批任务；
长期监控可使用 Linux 定时任务、Windows 计划任务，实现 7*24 小时离线运行。

2. 功能拓展方向

评价分析：提取返回评价文本，做关键词词云、差评统计；
选品筛选：新增销量、价格、店铺等级过滤规则，自动筛选蓝海潜力商品；
可视化：用 Matplotlib、Pyecharts 自动生成价格走势图片；
多平台兼容：可迁移适配淘宝、拼多多同类商品查询能力，搭建跨平台统一数据系统。

六、总结

依托 Open Claw 的调度能力 + 高阶商品详情查询，我们用三段轻量化 Python 代码，快速搭建了数据采集 - 实时监控 - 结构化分析 - 长期存档完整电商数据工具。对比传统爬虫，这套方案无需持续适配页面改版，开箱即用；对比付费商业监控工具，可高度自定义规则、无功能阉割、长期使用成本极低。

对于技术新手，直接复制代码修改密钥与商品 ID 即可快速落地；对于开发人员，可以在此基础上拓展数据库存储、前端可视化、多机器人告警等企业级功能。不管是个人卖家选品调研，还是团队竞品长效监控，这套落地思路都具备很强实用性。

万邦api博客

Nice to meet you, too!

api 数据挖掘