企业官网建设流程全解析

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/api，API Key 填控制台生成的 Key，Model 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": f"Bearer {api_key}", "Content-Type": "application/json" } resp = requests.post( f"{base_url}/v1/messages", headers=headers, json={ "model": "claude-sonnet-4-20250514", "max_tokens": 64, "messages": [{"role": "user", "content": "ping"}] }, timeout=20 ) 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": f"Bearer {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", params=params, headers=headers, timeout=20 ) 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, index=False, encoding="utf_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("开始商品监控循环，按 Ctrl+C 停止程序") 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。按 Ctrl+C 终止时，会打印各商品的价格区间统计。这套脚本的采集间隔默认 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 秒。脚本按 Ctrl+C 退出时会自动导出最终报表，这个习惯可以保留，方便每次运行都有完整的数据快照。