适用场景竞品盯价、库存巡检、商品信息跟踪、运营数据分析语言 Python适合电商运营、数据分析、后端开发阅读前言做电商运营或者竞品分析很多人还在重复一件枯燥工作打开商品页面复制标题、现价、原价、库存、规格信息粘贴到 Excel 表格每天定时手动记录一遍。商品一旦多起来几十个上百个商品人工处理不仅耗时巨大还很容易出错价格变动、库存清零、规格更新常常错过最佳的处理时机。促销大促期间商品信息变化飞快依靠人工盯守几乎不可能做到实时跟进。最近接触到 OpenClaw 这套工具它可以把商品页面的各类信息直接转为结构化数据省去复杂页面解析、反爬处理我们只需要聚焦业务逻辑快速搭建一套属于自己的数据监控脚本实现定时采集、本地存储、简单统计分析把重复工作交给程序完成。本文从实际落地角度完整演示从环境准备、编写采集脚本、数据持久化存储再到简单数据分析完整跑通一套可用方案。一、工具与环境准备Python3.7 及以上版本依赖库安装pip install requests pandas sqlite3requests负责网络请求获取结构化商品数据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, paramsparams, headersheaders, timeout20) 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 fgoods_{goods_id}_history.xlsx df.to_excel(out_file, indexFalse) 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 结构、验证码、签名等复杂底层问题开发者只需要把精力聚焦业务存储、统计、预警、报表输出。对于中小团队或者个人开发者这套方案成本很低几十行代码快速落地商品监控把运营人员从复制粘贴表格这种重复劳动解放出来把时间真正留给业务决策。