告别手工复制粘贴：搭建电商商品自动化监控与数据分析实践

适用场景：竞品盯价、库存巡检、商品信息跟踪、运营数据分析；语言 Python；适合电商运营、数据分析、后端开发阅读

前言

做电商运营或者竞品分析，很多人还在重复一件枯燥工作：打开商品页面，复制标题、现价、原价、库存、规格信息，粘贴到 Excel 表格，每天定时手动记录一遍。

商品一旦多起来，几十个上百个商品，人工处理不仅耗时巨大，还很容易出错，价格变动、库存清零、规格更新常常错过最佳的处理时机。促销大促期间，商品信息变化飞快，依靠人工盯守几乎不可能做到实时跟进。

最近接触到 OpenClaw 这套工具，它可以把商品页面的各类信息直接转为结构化数据，省去复杂页面解析、反爬处理，我们只需要聚焦业务逻辑，快速搭建一套属于自己的数据监控脚本，实现定时采集、本地存储、简单统计分析，把重复工作交给程序完成。

本文从实际落地角度完整演示，从环境准备、编写采集脚本、数据持久化存储，再到简单数据分析，完整跑通一套可用方案。

一、工具与环境准备

Python3.7 及以上版本
依赖库安装

pip install requests pandas sqlite3

requests：负责网络请求获取结构化商品数据
pandas：做数据清洗、统计、导出报表
sqlite3：轻量数据库，用来存放历史监控记录，无需额外安装服务

拿到 OpenClaw 凭证信息，包含两组密钥，用于身份校验，替换脚本内对应位置即可。

提示：不要把密钥硬编码直接提交到公开代码仓库，生产环境建议放到环境变量中。

二、核心采集脚本编写

我们的目标是拿到商品：商品编号、标题、现价、原价、库存、规格 SKU 列表、商品链接、图片地址、店铺名称等关键业务字段，做结构化解析。

# -*- coding:utf-8 -*-
import requests
import time
import json
import sqlite3
import pandas as pd

# --------------------------配置区域，请修改--------------------------
KEY = "你的凭证key"
SECRET = "你的凭证secret"
TARGET_GOODS_ID = "10335871600"  # 需要监控的商品编号
MONITOR_INTERVAL = 1800  # 监控周期，单位秒，这里设置30分钟采集一次
# ----------------------------------------------------------------

def fetch_goods_data(goods_id):
    """获取商品结构化原始数据"""
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": goods_id
    }
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        resp.raise_for_status()
        result = resp.json()
        return result
    except Exception as e:
        print(f"获取商品数据异常：{str(e)}")
        return None


def parse_goods_info(raw_json):
    """解析原始返回内容，提取业务需要字段"""
    if not raw_json or "item" not in raw_json:
        return None
    item = raw_json.get("item", {})
    if not item:
        return None

    data = {
        "num_iid": item.get("num_iid"),
        "title": item.get("title"),
        "price": float(item.get("price", 0)) if item.get("price") else 0,
        "orginal_price": float(item.get("orginal_price",0)) if item.get("orginal_price") else 0,
        "stock": item.get("num", 0),
        "detail_url": item.get("detail_url"),
        "pic_url": item.get("pic_url"),
        "brand": item.get("brand", ""),
        "shop_name": item.get("seller_info", {}).get("shop_name", ""),
        "capture_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    }

    # 提取多规格SKU信息
    sku_list = []
    skus_data = item.get("skus", {}).get("sku", [])
    for sku in skus_data:
        sku_item = {
            "sku_id": sku.get("sku_id"),
            "sku_price": sku.get("price"),
            "sku_original_price": sku.get("orginal_price"),
            "sku_properties": sku.get("properties_name"),
            "sku_stock": sku.get("quantity")
        }
        sku_list.append(sku_item)
    data["sku_info"] = sku_list
    return data

三、本地数据库存储：保存历史监控记录

想要做价格趋势、库存变化分析，必须把每一次采集结果落地保存，这里选用 sqlite，开箱即用。

def init_db():
    """初始化数据库表"""
    conn = sqlite3.connect("goods_monitor.db")
    cursor = conn.cursor()
    # 商品主记录表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_record (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        title TEXT,
        price REAL,
        orginal_price REAL,
        stock INTEGER,
        brand TEXT,
        shop_name TEXT,
        capture_time TEXT
    )
    ''')
    # SKU规格子表，存储多规格数据
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_sku (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        goods_record_id INTEGER,
        sku_id TEXT,
        sku_price REAL,
        sku_original_price REAL,
        sku_properties TEXT,
        sku_stock INTEGER
    )
    ''')
    conn.commit()
    conn.close()


def save_to_db(parsed_data):
    """将解析后的数据写入数据库"""
    if not parsed_data:
        return
    conn = sqlite3.connect("goods_monitor.db")
    cursor = conn.cursor()
    # 写入主表
    insert_sql = '''
    INSERT INTO goods_record(num_iid, title, price, orginal_price, stock, brand, shop_name, capture_time)
    VALUES (?,?,?,?,?,?,?,?)
    '''
    cursor.execute(insert_sql, (
        parsed_data["num_iid"],
        parsed_data["title"],
        parsed_data["price"],
        parsed_data["orginal_price"],
        parsed_data["stock"],
        parsed_data["brand"],
        parsed_data["shop_name"],
        parsed_data["capture_time"]
    ))
    record_id = cursor.lastrowid

    # 写入SKU子表
    for sku in parsed_data.get("sku_info", []):
        sku_sql = '''
        INSERT INTO goods_sku(goods_record_id,sku_id,sku_price,sku_original_price,sku_properties,sku_stock)
        VALUES (?,?,?,?,?,?)
        '''
        cursor.execute(sku_sql, (
            record_id,
            sku["sku_id"],
            float(sku["sku_price"]) if sku.get("sku_price") else 0,
            float(sku["sku_original_price"]) if sku.get("sku_original_price") else 0,
            sku["sku_properties"],
            sku["sku_stock"]
        ))
    conn.commit()
    conn.close()
    print(f"【{parsed_data['capture_time']}】商品 {parsed_data['num_iid']} 数据入库完成")

四、简单预警逻辑 + 定时循环监控

增加简单预警逻辑：当价格下跌超过阈值，或者库存变为 0 的时候打印提醒，实际项目可以扩展为邮件、企业微信消息推送。

def check_warning(data):
    """简单业务预警逻辑"""
    current_price = data.get("price", 0)
    stock = data.get("stock",0)
    threshold_rate = 0.8  # 低于原价80%触发降价提醒
    original = data.get("orginal_price", 0)

    if original > 0 and current_price <= original * threshold_rate:
        print(f"⚠️降价预警！商品：{data['title']}，现价{current_price}，原价{original}")
    if stock <= 0:
        print(f"⚠️库存预警！商品 {data['title']} 当前库存为0！")


def monitor_loop():
    """持续监控主循环"""
    init_db()
    print("商品监控程序已启动...")
    while True:
        raw = fetch_goods_data(TARGET_GOODS_ID)
        parsed = parse_goods_info(raw)
        if parsed:
            save_to_db(parsed)
            check_warning(parsed)
        time.sleep(MONITOR_INTERVAL)


if __name__ == "__main__":
    monitor_loop()

运行脚本之后，程序就会按照设定时间周期，自动拉取商品全部结构化信息，存入本地 sqlite 数据库，一旦触发降价、无库存条件就会打印告警信息。

五、数据分析：读取历史记录，导出报表

积累一段时间数据之后，我们可以读取数据库，做简单统计分析，导出 Excel 报表，方便运营复盘、看价格波动走势。新建 analysis.py 文件：

# analysis.py
import sqlite3
import pandas as pd

def export_history(goods_id):
    conn = sqlite3.connect("goods_monitor.db")
    df = pd.read_sql('''
        SELECT * FROM goods_record WHERE num_iid = ? ORDER BY capture_time
    ''', conn, params=(goods_id,))
    conn.close()
    if len(df) > 0:
        out_file = f"goods_{goods_id}_history.xlsx"
        df.to_excel(out_file, index=False)
        print(f"历史数据导出完成，文件:{out_file}")
        print("价格简单统计：")
        print(df[["price","orginal_price","stock"]].describe())
    else:
        print("暂无监控记录")

if __name__ == "__main__":
    export_history("10335871600")

执行后，会生成一份 Excel 文件，里面保存全部时间点采集的价格、库存、品牌店铺等记录，我们可以直接用 Excel 绘制价格折线图，直观看到一段时间商品价格波动变化。

六、实际业务拓展思路

批量监控多个商品：把商品 ID 放到列表或者配置文件，循环遍历采集，实现批量竞品巡检；
告警升级：把 print 打印告警改为企业微信、钉钉机器人推送，不用守着控制台；
数据清洗优化：SKU 多规格价格做单独监控，跟踪每个规格的价格、库存变化；
部署运行：Linux 服务器使用 nohup 后台运行脚本，实现 7×24 小时不间断监控；
可视化看板：读取 sqlite 数据接入 Grafana，直接做网页端可视化监控大盘。

七、踩坑经验总结

价格字段偶尔会存在偏差，属于平台页面动态渲染问题，做分析的时候建议多对比连续多次采集结果，不要单次结果直接判定；
密钥一定做好保管，公开代码不要明文上传；
设置合理采集间隔，不要高频请求，避免触发限制；
部分非自营店铺，部分店铺相关字段不会返回，代码做好判空，防止脚本报错中断；
商品下架时，返回内容会缺少主体信息，脚本做好空判断，保证循环不会崩溃。

写在最后

很多人一听到商品监控，就会觉得写页面解析、处理反爬会非常麻烦。借助 OpenClaw 把页面转为结构化数据，我们就不用再纠缠网页 DOM 结构、验证码、签名等复杂底层问题，开发者只需要把精力聚焦业务：存储、统计、预警、报表输出。

对于中小团队或者个人开发者，这套方案成本很低，几十行代码快速落地商品监控，把运营人员从复制粘贴表格这种重复劳动解放出来，把时间真正留给业务决策。

万邦api博客

Nice to meet you, too!

api 数据挖掘