Open Claw 实战教程：5 分钟搭建京东商品自动化监控与数据分析系统

admin 发表于2026-08-02 09:50:47 浏览7 评论0

前言

做电商运营、竞品调研、货源选品、铺货运维的朋友，几乎都会遇到同一个痛点：手动盯商品价格、库存、促销活动、规格变动效率极低；批量整理竞品数据、做价格走势复盘、销量口碑分析时，复制粘贴整理数据耗时耗力，人工统计还容易出错。

很多开发者尝试爬虫抓取页面数据，却频繁遭遇反爬封禁、页面改版失效、数据残缺等问题，调试与维护成本居高不下。最近一直在使用 Open Claw 工具完成电商数据自动化采集，依托其内置的高级商品详情采集能力，无需复杂逆向、不用频繁维护爬虫规则，就能稳定拿到京东商品全维度结构化数据，快速落地实时监控 + 长效数据分析两套常用业务，本文结合落地代码，完整分享从工具接入、实时监控脚本、数据分析整理全流程实操。

一、工具核心能力介绍

本次用到的高级详情采集能力，相比普通基础采集，最大优势是数据覆盖面更广、实时性更强、结构化程度更高：

基础信息全覆盖：商品标题、品牌、类目、主图、轮播图、长详情图文、商品参数、发货配送、店铺资质、自营 / 第三方店铺类型一键提取；
动态变动数据：实时售价、原价、优惠券、满减活动、多规格 SKU 单价与库存、近 30 天销量、库存余量；
口碑评价维度：好评率、总评论数、追评数量、差评标签，可用于竞品口碑打分；
灵活配置：支持关闭缓存获取毫秒级实时数据，可自由切换 JSON、格式化文本等数据格式，适配数据库存储、表格导出、可视化图表等场景。

调用逻辑十分简单，仅需要 2 个身份凭证 + 目标商品 ID，即可发起数据拉取，GET/POST 两种请求方式全部兼容，Python、Java、JS 等主流编程语言都能快速接入。

核心入参说明

表格

参数	是否必填	作用说明
身份凭证 1	是	个人专属调用密钥，平台后台注册后直接获取
身份凭证 2	是	安全校验密钥，和凭证 1 搭配使用，保障调用安全
商品 ID	是	京东商品详情页末尾数字编号，定位目标商品
cache	可选	yes 默认缓存加速；no 关闭缓存，用于价格、库存实时监控
result_type	可选	默认 JSON，日常数据分析可选择格式化阅读文本

返回核心可用字段（监控 & 分析高频使用）

基础：title 标题、shop_name 店铺名、pic_url 主图、goods_attr 商品参数
价格促销：price 现价、org_price 原价、coupon 优惠券列表
规格库存：sku 多规格集合、sku_price 单品价格、sku_stock 单品库存
销量口碑：sales 累计销量、goodRateShow 好评率、commentCount 总评论量

二、环境准备

本次全程使用 Python 实现，代码轻量化、无复杂依赖，只需要安装网络请求库：

pip install requests

运行环境：Python3.7 及以上，Windows、Mac、Linux 服务器均可部署，服务器部署可实现 7×24 小时不间断监控。

三、基础调用：单商品结构化数据获取

先写基础封装函数，完成单次商品数据拉取、异常重试、数据解析，为后续监控、数据分析打下基础。

import requests
import time
import json

class JDGoodsCollector:
    def __init__(self, key, secret):
        # 初始化身份凭证，替换为自己后台获取的凭证
        self.key = key
        self.secret = secret
        self.base_url = "https://api-gw.onebound.cn/jd/item_get_pro"
        self.session = requests.Session()  # 长连接优化请求速度

    def get_goods_detail(self, item_id, real_time=False):
        """
        拉取商品详情
        :param item_id: 京东商品ID
        :param real_time: 是否实时数据，True关闭缓存，用于监控价格库存变动
        :return: 结构化商品字典/None
        """
        params = {
            "key": self.key,
            "secret": self.secret,
            "api_name": "jd.item_get_pro",
            "num_iid": item_id,
            "cache": "no" if real_time else "yes",
            "result_type": "json"
        }
        retry_times = 3  # 3次失败重试
        for i in range(retry_times):
            try:
                resp = self.session.get(self.base_url, params=params, timeout=12)
                resp.raise_for_status()
                data = resp.json()
                if data.get("code") == 200:
                    return data.get("item", {})
                else:
                    print(f"请求异常：{data.get('msg', '未知错误')}")
            except Exception as e:
                print(f"第{i+1}次请求失败：{str(e)}，1秒后重试")
                time.sleep(1)
        print("多次请求失败，放弃本次采集")
        return None

# 测试基础调用
if __name__ == "__main__":
    # 替换为你的个人凭证
    KEY = "你的专属key"
    SECRET = "你的专属secret"
    collector = JDGoodsCollector(KEY, SECRET)
    target_item = "100012345678"  # 替换目标京东商品ID
    goods_info = collector.get_goods_detail(target_item, real_time=False)
    if goods_info:
        print("====商品基础信息====")
        print(f"商品标题：{goods_info.get('title')}")
        print(f"当前售价：{goods_info.get('price')} 元")
        print(f"原价：{goods_info.get('org_price')} 元")
        print(f"店铺名称：{goods_info.get('shop_name')}")
        print(f"累计销量：{goods_info.get('sales')}")
        print(f"好评率：{goods_info.get('goodRateShow')}")

运行后即可一键打印商品核心信息，代码做了超时、网络异常重试，稳定性远高于爬虫脚本。

四、实战 1：搭建商品实时价格 & 库存监控脚本

需求场景

竞品降价预警、自有商品库存告警、大促期间价格变动捕捉，一旦价格下调、库存大幅减少，立刻打印提醒，可拓展对接微信、钉钉推送消息。实现思路：循环定时采集实时数据，对比上一轮采集数据，出现变动触发告警。

import requests
import time
import json

class JDMonitor:
    def __init__(self, key, secret):
        self.collector = JDGoodsCollector(key, secret)
        # 历史数据缓存，存储上一轮采集结果
        self.history_data = {}

    def monitor_single_goods(self, item_id, interval=300):
        """
        单商品循环监控
        :param item_id: 商品ID
        :param interval: 采集间隔，单位秒，默认5分钟采集一次
        """
        print(f"====开始监控商品{item_id}，采集间隔{interval}秒====")
        while True:
            now_data = self.collector.get_goods_detail(item_id, real_time=True)
            if not now_data:
                time.sleep(interval)
                continue
            # 首次采集，仅缓存数据不对比
            if item_id not in self.history_data:
                self.history_data[item_id] = now_data
                print(f"初始化缓存完成，当前价格：{now_data.get('price')}元")
                time.sleep(interval)
                continue

            old_data = self.history_data[item_id]
            alert_list = []
            # 价格变动检测
            old_price = float(old_data.get("price", 0))
            new_price = float(now_data.get("price", 0))
            if new_price < old_price:
                alert_list.append(f"【降价提醒】原价{old_price} → 现价{new_price}")
            elif new_price > old_price:
                alert_list.append(f"【涨价提醒】原价{old_price} → 现价{new_price}")

            # 库存变动检测
            old_stock = old_data.get("sku", [])
            new_stock = now_data.get("sku", [])
            if len(new_stock) > 0 and str(new_stock[0].get("skuStock")) != str(old_stock[0].get("skuStock")):
                alert_list.append(f"【库存变动】当前库存：{new_stock[0].get('skuStock')}")

            # 输出告警
            if alert_list:
                print("\n=====触发告警=====")
                print(f"商品：{now_data.get('title')}")
                for alert in alert_list:
                    print(alert)
                print("==================\n")

            # 更新历史缓存
            self.history_data[item_id] = now_data
            time.sleep(interval)

# 监控运行入口
if __name__ == "__main__":
    KEY = "你的专属key"
    SECRET = "你的专属secret"
    monitor = JDMonitor(KEY, SECRET)
    # 监控目标商品，间隔300秒（5分钟）采集一次
    monitor.monitor_single_goods("100012345678", interval=300)

拓展优化方向

多商品批量监控：将商品 ID 存入列表，循环遍历采集；
消息推送：对接钉钉机器人、企业微信接口，变动推送至工作群；
阈值告警：设定降价幅度阈值，仅降价超过 5% 才提醒，过滤小幅波动。

五、实战 2：批量采集 + 长效数据分析整理脚本

日常竞品分析、选品调研，需要批量抓取多款商品数据，整理为表格结构化数据，用于价格对比、口碑排序、热度筛选，下面代码实现批量采集 + 数据整理，可直接导出 CSV 表格。

import requests
import time
import csv

class JDAnalysisTool:
    def __init__(self, key, secret):
        self.collector = JDGoodsCollector(key, secret)

    def batch_collect(self, id_list):
        """批量采集多款商品数据"""
        result_list = []
        print(f"开始批量采集{len(id_list)}个商品数据")
        for idx, item_id in enumerate(id_list):
            print(f"正在采集第{idx+1}/{len(id_list)}：{item_id}")
            info = self.collector.get_goods_detail(item_id, real_time=False)
            if info:
                item_dict = {
                    "商品ID": item_id,
                    "商品标题": info.get("title"),
                    "现价": info.get("price"),
                    "原价": info.get("org_price"),
                    "店铺": info.get("shop_name"),
                    "销量": info.get("sales"),
                    "好评率": info.get("goodRateShow"),
                    "总评论": info.get("commentCount")
                }
                result_list.append(item_dict)
            time.sleep(1)  # 防频繁请求限流
        return result_list

    def export_csv(self, data_list, save_name="jd_goods_analysis.csv"):
        """导出csv表格，Excel可直接打开"""
        if not data_list:
            print("无数据可导出")
            return
        headers = list(data_list[0].keys())
        with open(save_name, "w", encoding="utf-8-sig", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=headers)
            writer.writeheader()
            writer.writerows(data_list)
        print(f"数据已导出至：{save_name}")

    def sort_hot(self, data_list):
        """数据分析：按销量降序排序，筛选热门商品"""
        return sorted(data_list, key=lambda x: int(x.get("销量", 0)) if x.get("销量") else 0, reverse=True)

# 数据分析运行入口
if __name__ == "__main__":
    KEY = "你的专属key"
    SECRET = "你的专属secret"
    analysis = JDAnalysisTool(KEY, SECRET)
    # 批量竞品ID列表
    goods_ids = [
        "100012345678",
        "100023456789",
        "100034567890"
    ]
    data = analysis.batch_collect(goods_ids)
    analysis.export_csv(data)
    hot_data = analysis.sort_hot(data)
    print("\n====热门商品排序（销量从高到低）====")
    for item in hot_data:
        print(f"{item['商品标题']} | 销量：{item['销量']} | 价格：{item['现价']}")

数据分析拓展用法

价格区间筛选：增加函数过滤低价、高价竞品，锁定对标价位商品；
口碑筛选：过滤好评率低于 90% 的劣质竞品；
入库存储：将数据存入 MySQL、Sqlite，长期存储后绘制价格走势折线图，复盘大促价格规律。

六、落地常见问题与优化技巧

调用限流：个人日常使用间隔 1~3 秒，企业批量采集用异步队列错峰请求，不要短时间高频循环；
监控实时性：日常监控 5~10 分钟采集一次足够，618、双 11 大促可缩短至 2~3 分钟；
轻量化部署：脚本放到云服务器后台常驻运行，无需本地电脑一直开机；
冗余过滤：不需要详情图片、长参数时，可自行裁剪解析字段，减少内存占用。

七、总结

借助 Open Claw 的高级商品详情采集能力，不用钻研复杂网页逆向、不用反复维护爬虫规则，几十行代码就能快速落地稳定可靠的商品监控系统与数据分析工具。

个人从业者：用来盯竞品、选货源、薅促销活动；
电商团队：搭建自动化数据中台，批量竞品监控、价格调研、铺货素材整理；整套代码轻量化易修改，新手复制替换凭证与商品 ID 即可直接运行，长期使用可以根据业务叠加消息推送、可视化图表、数据库存储等拓展功能。

万邦api博客

Nice to meet you, too!

api 数据挖掘