1. 从手动盯价到脚本自动落库京东商品监控到底解决什么问题做电商运营或者选品调研最耗精力的往往不是分析而是「盯」。一个商品今天 89、明天 79、后天又回到 89库存从 200 掉到 0 再补货这些变化如果靠人工每天打开页面复制粘贴不仅效率低还特别容易漏掉关键节点。我试过用表格手动记录十几个竞品坚持不到一周就放弃了因为数据一多光是核对哪条记录对应哪个时间点就够头疼。OpenClaw 实战这套思路的核心是把「采集—落库—分析」拆成三个独立环节每个环节都用最轻量的工具完成。Python 负责发请求和调度SQLite 负责存历史快照pandas 负责把冷冰冰的记录变成能看懂的统计。你不需要部署 MySQL不需要写复杂的爬虫框架甚至不需要理解网页 DOM 结构只要拿到商品编号就能把标题、售价、原价、库存、店铺、发货地这些字段定时拉下来。这篇文章面向的是有基础 Python 环境、想快速搭一套本地监控脚本的读者。适合个人做竞品跟踪、小团队做选品复盘也适合想练手 requests SQLite pandas 组合的开发者。整篇会给出可直接复制的采集配置、建表语句、定时任务脚本以及数据校验和异常重试的验证动作。需要提前说明的是脚本仅用于技术学习实际使用时请遵守平台规则控制访问频率不要高频批量抓取避免对目标网站造成压力。核心检索词先摆出来OpenClaw 调用 Python 脚本抓取京东商品价格与库存落库 SQLite 并用 pandas 做趋势统计。下面从环境准备开始一步步把可运行的代码搭起来。2. TaoToken 前置准备API Key 与模型接入配置在写采集脚本之前先把调用链路里的凭证和接入配置理清楚。很多同学卡在第一步不是因为代码写错而是 Key 没配对、Base URL 填错、Model ID 写成了展示名。这里以 TaoToken 的接入方式为例把三件套Base URL、API Key、Model ID一次讲透。TaoToken 的 API 地址是https://taotoken.net/api注意这个地址不带任何查询参数直接作为请求的 base。API Key 需要到控制台创建路径是 console 页面下的 api-keys 管理。创建之后复制那串以sk-开头的字符串不要截图保存直接粘贴到环境变量里。如果你用的是 Claude Code 这类编码工具配置方式略有不同。Claude Code 走的是 Anthropic 兼容协议需要在 settings 里指定 base_url 和 api_key。下面给一份可复制的 JSON 配置片段路径和字段名保持和实际一致{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的实际Key, ANTHROPIC_MODEL: claude-sonnet-4-20250514 } }如果你用的是 Cline 或者带 MCP 的编辑器插件配置项通常写在cline_mcp_settings.json或者对应的 MCP 配置文件里。核心字段同样是三项Base URL 填https://taotoken.net/apiAPI Key 填控制台生成的 KeyModel ID 填你要调用的具体模型标识。Model ID 不是展示名称不能随便写「Claude 最新版」这种必须用平台文档里列出的准确 ID。对于 Codex 这类工具认证信息一般落在auth.json里。结构大致如下{ base_url: https://taotoken.net/api, api_key: sk-你的实际Key, model: gpt-4o }这里要提醒一个高频坑Base URL 末尾不要多加/v1或者/chat/completions很多 401 和 404 就是因为路径拼重复了。正确的做法是 base 只写到/api具体的端点由 SDK 或工具自己拼接。环境变量配置好之后可以用一个最小请求验证连通性。Python 里这样写import os import requests base_url os.getenv(ANTHROPIC_BASE_URL, https://taotoken.net/api) api_key os.getenv(ANTHROPIC_API_KEY) headers { Authorization: fBearer {api_key}, Content-Type: application/json } resp requests.post( f{base_url}/v1/messages, headersheaders, json{ model: claude-sonnet-4-20250514, max_tokens: 64, messages: [{role: user, content: ping}] }, timeout20 ) print(resp.status_code) print(resp.text[:200])如果返回 200 并且能看到内容说明 Key 和 Base URL 都没问题。如果返回 401优先检查 Key 是否复制完整、有没有多余空格如果返回 404检查 base 路径是不是多写了后缀。这一步过了再进入采集脚本的编写。3. 可复制配置requests 采集 SQLite 建表 定时任务脚本这一节是整篇的核心给出可以直接复制运行的完整脚本。结构上分成四块配置区、数据库初始化、采集与落库、定时循环与报表导出。每一块都标了注释方便你按需修改。先看配置区和数据库初始化。商品 ID 列表、采集间隔、数据库路径都放在最上面改起来直观。建表语句用CREATE TABLE IF NOT EXISTS重复运行不会报错。# -*- coding: utf-8 -*- import os import time import sqlite3 import requests import pandas as pd from datetime import datetime # -------------------------- 配置区域 --------------------------- API_KEY os.getenv(TAOTOKEN_API_KEY, your_key_here) BASE_URL https://taotoken.net/api MONITOR_ITEM_IDS [10335871600] FETCH_INTERVAL 600 # 采集间隔单位秒测试建议不低于 300 DB_PATH ./jd_monitor.db MAX_RETRY 3 # 单次请求最大重试次数 RETRY_BACKOFF 5 # 重试退避基数单位秒 # -------------------------------------------------------------- def init_db(): 初始化数据库存储每次采集快照 conn sqlite3.connect(DB_PATH) cur conn.cursor() cur.execute( CREATE TABLE IF NOT EXISTS item_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, num_iid TEXT, title TEXT, price REAL, original_price REAL, stock INTEGER, sales INTEGER, shop_name TEXT, location TEXT, fetch_time TIMESTAMP ) ) conn.commit() conn.close()建表字段和采集字段一一对应fetch_time用字符串存时间戳方便 pandas 直接解析。接下来是采集函数带重试和异常捕获def fetch_item_detail(num_iid): 获取单款商品详情带重试 headers { Accept-Encoding: gzip, Connection: close, Authorization: fBearer {API_KEY} } params { num_iid: num_iid, result_type: json, cache: no } for attempt in range(1, MAX_RETRY 1): try: resp requests.get( f{BASE_URL}/jd/item_get_pro, paramsparams, headersheaders, timeout20 ) resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: print(f请求异常 num_iid:{num_iid} 第{attempt}次 err:{e}) if attempt MAX_RETRY: time.sleep(RETRY_BACKOFF * attempt) return None重试逻辑用指数退避第一次等 5 秒第二次等 10 秒避免短时间内反复打同一个接口。落库函数负责把返回的 JSON 解析成字段并写入def save_snapshot(raw_data): 把单次采集结果写入数据库快照 if not raw_data or item not in raw_data: print(返回数据为空或缺少 item 字段跳过落库) return item raw_data.get(item, {}) conn sqlite3.connect(DB_PATH) cur conn.cursor() cur.execute( INSERT INTO item_history (num_iid, title, price, original_price, stock, sales, shop_name, location, fetch_time) VALUES (?,?,?,?,?,?,?,?,?) , ( item.get(num_iid), item.get(title), float(item.get(price, 0) or 0), float(item.get(orginal_price, 0) or 0), int(item.get(num, 0) or 0), int(item.get(sales, 0) or 0), item.get(nick, ), item.get(location, ), datetime.now().strftime(%Y-%m-%d %H:%M:%S) )) conn.commit() conn.close() price float(item.get(price, 0) or 0) stock int(item.get(num, 0) or 0) print(f【快照】{item.get(title)} | 当前价:{price} | 库存:{stock} | {datetime.now()}) if stock 0: print(f警告商品 {item.get(num_iid)} 库存归零)注意orginal_price这个字段名是接口返回的原始拼写不要想当然改成original_price否则取不到值。落库之后是报表导出和简易分析def export_analysis_csv(): 导出历史数据并输出 CSV conn sqlite3.connect(DB_PATH) df pd.read_sql(SELECT * FROM item_history ORDER BY fetch_time DESC, conn) conn.close() csv_name f./item_analysis_{datetime.now().strftime(%Y%m%d_%H%M)}.csv df.to_csv(csv_name, indexFalse, encodingutf_8_sig) print(f分析报表导出完成{csv_name}) return df def simple_analysis(df): 简易分析价格区间、库存变化、记录条数 if df.empty: print(暂无采集数据) return print(\n 简易分析报告 ) for item_id in df[num_iid].unique(): sub df[df[num_iid] item_id] min_p sub[price].min() max_p sub[price].max() latest sub.iloc[0] print(f商品ID:{item_id}) print(f记录条数:{len(sub)} | 历史最低:{min_p} | 历史最高:{max_p} f| 最新价格:{latest[price]} | 最新库存:{latest[stock]}) print(- * 40)最后是主循环把上面几个函数串起来def monitor_loop(): init_db() print(开始商品监控循环按 CtrlC 停止程序) try: while True: for item in MONITOR_ITEM_IDS: res fetch_item_detail(item) if res: save_snapshot(res) export_analysis_csv() time.sleep(FETCH_INTERVAL) except KeyboardInterrupt: print(\n收到停止信号准备导出最终报表...) df export_analysis_csv() simple_analysis(df) print(程序退出) if __name__ __main__: monitor_loop()运行之后控制台会持续输出快照信息每轮采集完自动导出一次 CSV。按 CtrlC 终止时会打印各商品的价格区间统计。这套脚本的采集间隔默认 600 秒测试阶段不要调得太小避免触发访问限制。4. 验证请求与成功结果数据校验和异常重试的实操动作脚本能跑起来只是第一步真正要确认的是「数据对不对、异常有没有被兜住」。这一节给出几个可执行的验证动作帮你判断采集链路是否健康。第一个动作是单次请求验证。在正式跑循环之前先单独调一次fetch_item_detail把返回的 JSON 打印出来重点看三个字段item.num_iid是否和传入的一致item.price是不是数字item.num是不是整数。如果price返回的是字符串落库时float()会报错需要加一层类型转换保护。下面这段可以单独跑if __name__ __main__: init_db() res fetch_item_detail(MONITOR_ITEM_IDS[0]) if res: item res.get(item, {}) print(num_iid:, item.get(num_iid)) print(title:, item.get(title)) print(price:, item.get(price), type(item.get(price))) print(stock:, item.get(num), type(item.get(num))) save_snapshot(res)第二个动作是数据库校验。落库之后用一条 SQL 确认记录真的写进去了并且字段没有错位conn sqlite3.connect(DB_PATH) cur conn.cursor() cur.execute(SELECT num_iid, title, price, stock, fetch_time FROM item_history ORDER BY id DESC LIMIT 5) for row in cur.fetchall(): print(row) conn.close()如果发现price全是 0大概率是字段名写错或者返回结构变了如果fetch_time为空检查插入语句的参数顺序。第三个动作是异常重试验证。把BASE_URL临时改成一个不存在的地址观察控制台是否按 5 秒、10 秒的间隔重试三次最后返回 None 而不是直接崩溃。这个动作能确认你的重试逻辑真的生效而不是写在代码里没被触发。第四个动作是 pandas 分析校验。导出 CSV 之后用 pandas 读回来检查fetch_time能不能被解析成 datetime价格列有没有异常值df pd.read_csv(./item_analysis_20260731_1522.csv) df[fetch_time] pd.to_datetime(df[fetch_time]) print(df[price].describe()) print(df.groupby(num_iid)[price].agg([min, max, count]))如果describe()出来的 min 是负数或者 max 大得离谱说明采集字段可能串了需要回到落库函数逐字段核对。这几个动作做完基本能确认「采集—落库—分析」三段链路是通的。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 对照实际跑脚本时报错信息往往比代码本身更让人头疼。这一节把几个高频错误和对应的排查方向列出来方便你对照定位。401 Unauthorized 是最常见的。出现这个错误先检查 API Key 有没有复制完整前后有没有多余空格或换行。然后确认请求头里的Authorization格式是不是Bearer sk-xxx少写Bearer或者多写冒号都会导致 401。如果 Key 确认没问题再看 Base URL 是不是写成了https://taotoken.net/api/带尾斜杠某些 SDK 对尾斜杠敏感去掉再试。local proxy failed 通常和网络环境有关。这个报错说明请求在到达目标地址之前就被本地代理拦截了。排查方向是检查系统环境变量里有没有HTTP_PROXY、HTTPS_PROXY这类配置如果有临时清掉再跑。另外确认代码里没有手动设置proxies参数。这个错误和 Key 无关纯粹是请求出口的问题。reading choices 这类报错一般出现在解析返回结构的时候。接口返回的 JSON 里如果没有item字段或者item下面没有price直接取就会抛 KeyError。解决办法是在落库函数里用.get()加默认值并且在解析前先判断item in raw_data。上面给的save_snapshot已经做了这层保护如果你自己改过代码记得保留这个判断。OAuth 相关报错通常出现在 Claude Code 或者带认证流程的工具里。如果你用的是 API Key 模式却触发了 OAuth 流程说明工具的认证方式选错了。检查配置文件里是不是同时存在api_key和oauth_token两个字段如果有删掉 OAuth 相关的只保留 API Key。另外确认ANTHROPIC_BASE_URL指向的是https://taotoken.net/api而不是某个 OAuth 授权地址。还有一个容易被忽略的坑SQLite 数据库文件被占用。如果你同时开了两个脚本实例写同一个jd_monitor.db会报database is locked。解决办法是确保同一时间只有一个写入进程或者把采集间隔拉长减少并发写入的概率。排查顺序建议是先看 HTTP 状态码再看返回体最后看本地数据库。状态码 401/404 属于配置问题状态码 200 但数据不对属于解析问题数据库报错属于并发或路径问题。按这个顺序走大部分问题都能定位到具体环节。6. 语义一致 CTA把脚本跑起来之后可以做什么脚本跑通之后你会发现真正的价值不在采集本身而在积累下来的历史数据。有了 SQLite 里的快照你可以做价格趋势图、库存波动分析、竞品调价频率统计甚至把 CSV 丢给 pandas 做更复杂的聚合。如果后续想让脚本具备「智能分析」能力比如自动总结价格波动原因、生成选品建议可以接入模型对话能力把统计结果作为上下文传进去。需要长期跑编码任务或者 Agent 类工作流的可以了解 Coding Plan适合把采集、分析、告警串成一条自动化链路。接入文档里有完整的端点和参数说明配置过程中遇到 401 或者路径问题对照文档里的示例再核一遍 Base URL 和 Key 的写法。最后留一个实用技巧把DB_PATH和MONITOR_ITEM_IDS放到环境变量或者单独的配置文件里不要硬编码在脚本中。这样换商品、换数据库路径的时候不用改代码也避免把敏感信息提交到公开仓库。采集间隔根据实际需要调整日常监控 600 秒起步测试阶段不要低于 300 秒。脚本按 CtrlC 退出时会自动导出最终报表这个习惯可以保留方便每次运行都有完整的数据快照。