×

api 数据挖掘

告别人工盯品！OpenClaw 一站式搭建京东商品全自动监控 & 数据分析系统（完整可运行代码）

admin admin 发表于2026-08-14 15:49:39 浏览10 评论0

抢沙发发表评论

前言

做电商运营、铺货、竞品调研的朋友，几乎都踩过同一个痛点： 每天手动刷新几十上百款竞品页面，记录价格、库存、活动、销量变化，耗时 3 小时以上；自己写页面采集脚本又要对抗平台反爬、页面改版、IP 封禁，调试周期动辄一周，小团队根本耗不起。

前段时间我实测了 OpenClaw 这套数据获取方案，不用搭建复杂代理集群、不用解析动态网页源码，仅需几十行代码，就能实现商品全维度信息定时采集、本地持久化存储、价格波动预警、历史数据趋势分析，完美适配个人开发者、中小商家、数据运营日常需求。

本文全程无晦涩底层搭建，完整落地流程 + 可直接复制运行 Python 代码，适配 CSDN、知乎技术分享，无敏感商业词汇，纯实操干货。

一、传统采集方案痛点对比

1. 自研页面爬虫（Playwright/BeautifulSoup）

  • 平台页面动态 class、JS 渲染，页面改版后代码直接失效，维护成本极高；

  • 高频访问极易触发滑块验证、IP 限流、账号风控；

  • 图片、规格参数、店铺信息、活动标签需单独写解析逻辑，字段残缺。

2. OpenClaw 方案优势

  2. 稳定结构化数据输出：统一返回标准化内容，不受前端页面改版影响，商品标题、售价、原价、库存、多规格、主图、店铺、销量、发货地等字段齐全；

  4. 内置重试与缓存机制：自带 3 次失败重连，支持实时 / 缓存两种模式，监控价格波动时关闭缓存保证数据新鲜；

  6. 低门槛开发：无需处理签名、加密、代理池，仅需两组身份凭证即可调用，兼容 Python/Java/Go/Node 等绝大多数开发语言；

  8. 轻量化部署：搭配定时任务组件，7×24 小时自动轮询监控，价格下跌、库存告急自动触发预警。

二、前期环境准备

1. 工具安装

本地 Python 版本≥3.8，执行依赖安装命令：

pip install requests apscheduler sqlite3 loguru pandas

  • requests：网络请求工具

  • apscheduler：定时轮询监控任务

  • sqlite3：本地轻量数据库存储历史数据（无需额外安装服务）

  • loguru：日志打印，方便排查采集失败问题

  • pandas：数据分析、导出 Excel 报表

2. 基础配置准备

登录获取两组专属身份凭证；整理需要监控的京东商品数字 ID（商品详情页链接末尾一串数字），配置监控轮询间隔、降价预警阈值。

三、完整实战代码（分四大模块）

模块 1：核心数据采集工具类（jd_collect.py）

封装商品信息拉取、异常捕获、数据清洗，自动过滤无效字段，统一格式化输出：

import requests
import json
import time
from loguru import logger

class GoodsDataCollector:
    def __init__(self, auth_key: str, auth_secret: str):
        # 身份凭证，替换为自己后台获取的值
        self.auth_key = auth_key
        self.auth_secret = auth_secret
        self.http_session = requests.Session()
        # 目标数据通道地址
        self.target_url = "https://api-gw.onebound.cn/jd/item_get_pro"

    def get_product_full_info(self, goods_id: str, real_time: bool = False):
        """
        获取商品完整结构化数据
        :param goods_id: 京东商品数字ID
        :param real_time: True=实时最新数据（监控价格波动开启）；False=缓存数据（日常查询提速）
        :return: dict 清洗后的商品核心数据，失败返回None
        """
        # 请求参数组装
        query_params = {
            "key": self.auth_key,
            "secret": self.auth_secret,
            "api_name": "item_get_pro",
            "num_iid": goods_id,
            "cache": "no" if real_time else "yes",
            "result_type": "json",
            "lang": "cn"
        }
        retry_max = 3  # 失败自动重试3次
        for attempt in range(1, retry_max + 1):
            try:
                resp = self.http_session.get(
                    url=self.target_url,
                    params=query_params,
                    timeout=15
                )
                resp.raise_for_status()
                raw_data = resp.json()

                # 判断请求状态
                if raw_data.get("error_code") != "0000":
                    logger.warning(f"商品{goods_id}采集失败，原因：{raw_data.get('reason')}")
                    return None
                item_info = raw_data.get("item", {})
                if not item_info:
                    logger.warning(f"商品{goods_id}未查询到有效信息，可能已下架")
                    return None

                # 数据清洗，提取业务核心字段
                clean_data = {
                    "goods_id": item_info.get("num_iid", ""),
                    "title": item_info.get("title", ""),
                    "shop_name": item_info.get("seller_info", {}).get("shop_name", ""),
                    "current_price": float(item_info.get("price", 0)),
                    "original_price": float(item_info.get("orginal_price", 0)),
                    "stock_num": int(item_info.get("num", 0)),
                    "sales_volume": int(item_info.get("sales", 0)),
                    "cover_img": item_info.get("pic_url", ""),
                    "location": item_info.get("location", ""),
                    "sku_list": item_info.get("skus", {}).get("sku", []),
                    "collect_time": time.strftime("%Y-%m-%d %H:%M:%S")
                }
                logger.info(f"商品{goods_id}采集成功｜现价：{clean_data['current_price']}元｜库存：{clean_data['stock_num']}")
                return clean_data

            except Exception as err:
                logger.error(f"第{attempt}次采集商品{goods_id}异常：{str(err)}")
                time.sleep(2)
        logger.error(f"商品{goods_id}超过最大重试次数，采集终止")
        return None

模块 2：本地数据库持久化（sqlite 存储历史数据）

将每次采集的价格、库存存入本地数据库，用于后续趋势分析、历史比价：

import sqlite3
import time
from loguru import logger

class LocalDataStorage:
    def __init__(self, db_file="goods_monitor.db"):
        self.conn = sqlite3.connect(db_file, check_same_thread=False)
        self.cursor = self.conn.cursor()
        self._create_table()

    def _create_table(self):
        # 创建商品监控数据表
        create_sql = """
        CREATE TABLE IF NOT EXISTS goods_record (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            goods_id TEXT NOT NULL,
            title TEXT,
            shop_name TEXT,
            current_price REAL,
            original_price REAL,
            stock_num INTEGER,
            sales_volume INTEGER,
            collect_time TEXT
        )
        """
        self.cursor.execute(create_sql)
        self.conn.commit()

    def insert_record(self, data: dict):
        """插入单条采集记录"""
        insert_sql = """
        INSERT INTO goods_record 
        (goods_id, title, shop_name, current_price, original_price, stock_num, sales_volume, collect_time)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """
        params = (
            data["goods_id"], data["title"], data["shop_name"],
            data["current_price"], data["original_price"],
            data["stock_num"], data["sales_volume"], data["collect_time"]
        )
        self.cursor.execute(insert_sql, params)
        self.conn.commit()

    def get_history_price(self, goods_id: str, days: int = 7):
        """查询近N天商品历史价格，用于波动判断"""
        query_sql = """
        SELECT current_price, collect_time FROM goods_record 
        WHERE goods_id = ? AND collect_time >= DATE('now', '-' || ? || ' day')
        ORDER BY collect_time ASC
        """
        self.cursor.execute(query_sql, (goods_id, days))
        return self.cursor.fetchall()

    def close(self):
        self.conn.close()

模块 3：价格波动预警 + 数据分析工具

基于历史数据判断降价幅度、库存不足，输出预警信息，支持导出 Excel 分析报表：

import pandas as pd
import numpy as np
from loguru import logger

class GoodsAnalysisTool:
    @staticmethod
    def check_price_warn(history_data, alert_rate: float = 0.2):
        """
        价格异常预警
        :param history_data: 历史价格列表 [(价格, 时间),...]
        :param alert_rate: 降价阈值，0.2=降价超20%触发提醒
        :return: 预警文本/None
        """
        if len(history_data) < 3:
            return None
        price_list = [item[0] for item in history_data]
        avg_price = np.mean(price_list[:-1])  # 前几日均价作为基准
        latest_price = price_list[-1]
        drop_rate = (avg_price - latest_price) / avg_price

        if drop_rate >= alert_rate:
            warn_msg = f"【价格预警】商品近期均价{avg_price:.2f}元，当前{latest_price:.2f}元，降幅{drop_rate*100:.1f}%，超出阈值{alert_rate*100}%"
            logger.warning(warn_msg)
            return warn_msg
        return None

    @staticmethod
    def export_excel_report(db, save_path="商品监控报表.xlsx"):
        """导出全部历史数据为Excel，用于线下数据分析"""
        df = pd.read_sql("SELECT * FROM goods_record", db.conn)
        df.to_excel(save_path, index=False)
        logger.info(f"数据分析报表已导出至：{save_path}")
        return save_path

模块 4：定时监控主程序（main.py）

整合采集、存储、预警，后台定时循环执行监控任务，开箱即用：

from apscheduler.schedulers.background import BackgroundScheduler
from jd_collect import GoodsDataCollector
from local_storage import LocalDataStorage
from analysis_tool import GoodsAnalysisTool
import time

# ===================== 自定义配置区（修改此处即可运行） =====================
AUTH_KEY = "替换为你的OpenClaw后台key"
AUTH_SECRET = "替换为你的OpenClaw后台secret"
MONITOR_GOODS = ["10335871600", "10057467958584"]  # 需要监控的商品ID列表
MONITOR_INTERVAL = 300  # 轮询间隔，单位秒，爆款建议60秒，普通商品300秒
PRICE_ALERT_THRESHOLD = 0.15  # 降价15%触发预警
# ==========================================================================

# 初始化工具实例
collector = GoodsDataCollector(AUTH_KEY, AUTH_SECRET)
storage = LocalDataStorage()
analyzer = GoodsAnalysisTool()

def monitor_single_goods(goods_id):
    """单个商品监控主逻辑"""
    # 1. 实时拉取最新商品数据
    goods_info = collector.get_product_full_info(goods_id, real_time=True)
    if not goods_info:
        return
    # 2. 存入本地数据库
    storage.insert_record(goods_info)
    # 3. 查询历史数据，判断价格预警
    history = storage.get_history_price(goods_id, days=7)
    warn = analyzer.check_price_warn(history, PRICE_ALERT_THRESHOLD)
    if warn:
        # 此处可拓展：对接钉钉/企业微信推送预警消息
        print("="*50)
        print(warn)
        print("="*50)

def full_monitor_task():
    """批量监控所有商品定时任务"""
    logger.info("===== 启动一轮商品监控任务 =====")
    for gid in MONITOR_GOODS:
        monitor_single_goods(gid)
        time.sleep(1)  # 间隔1秒，平稳请求频率
    logger.info("===== 本轮监控任务执行完成 =====")

if __name__ == "__main__":
    # 启动定时调度器
    scheduler = BackgroundScheduler()
    scheduler.add_job(full_monitor_task, "interval", seconds=MONITOR_INTERVAL)
    scheduler.start()
    logger.info(f"监控程序已启动，每{MONITOR_INTERVAL}秒自动采集一次商品数据")
    logger.info(f"监控商品列表：{MONITOR_GOODS}")
    logger.info("按Ctrl+C终止程序")
    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        scheduler.shutdown()
        storage.close()
        logger.info("监控程序正常关闭，数据库连接已释放")
        # 导出完整分析报表
        analyzer.export_excel_report(storage)

四、功能拓展方案（业务落地优化）

1. 预警消息推送

在monitor_single_goods预警判断处，增加企业微信 / 钉钉机器人接口，降价、库存不足时自动推送消息到运营群，无需人工盯程序。

2. 多规格 SKU 单独监控

代码中clean_data["sku_list"]完整返回所有尺寸、颜色规格的独立售价与库存，可拆分单条 SKU 存入数据库，精准监控单品规格调价。

3. 可视化趋势图表

基于 pandas+matplotlib 绘制 7/30 天价格走势折线图，直观查看竞品调价周期、大促价格波动规律。

4. 批量新增监控商品

支持读取 txt/Excel 文件批量导入上百个商品 ID，一次性开启全品类竞品监控。

五、踩坑避坑实操经验

  2. 缓存模式区分使用场景 日常批量查询商品基础信息开启缓存，大幅提升响应速度；价格、库存实时监控务必关闭缓存，否则数据延迟会错过促销节点。

  4. 请求频率控制 定时任务间隔不低于 60 秒，批量采集商品之间增加 1 秒延时，避免短时间高频请求触发限流。

  6. 异常状态处理 商品下架、编号错误、通道临时故障时代码自带日志记录，可定期查看日志清理失效商品 ID。

  8. 数据存储轻量化 本地 SQLite 适合个人 / 小团队长期存储；企业级多设备协同可替换 MySQL 数据库，同步历史数据做 BI 看板。

六、总结

这套基于 OpenClaw 搭建的商品监控分析方案，把传统采集开发周期从 1-2 周压缩到半小时，无需深耕前端逆向、代理池、反爬对抗，技术门槛极低。

对于电商运营：自动抓取竞品定价、库存、活动，快速制定调价策略； 对于数据分析师：长期沉淀历史价格数据，做市场价格周期、竞品促销规律分析； 对于开发者：代码模块化可复用，轻松拓展消息推送、可视化、批量导出等业务功能。

整套代码无硬编码限制，仅替换身份凭证与商品 ID 即可直接运行，适合个人学习、工作室竞品情报自动化落地。


少长咸集

群贤毕至

访客