数据在云端但查不到?723 行 Python 让我一句话查聊天

数据在云端但查不到?723 行 Python 让我一句话查聊天

情人节那天,小龙虾帮数字 Polly 装上了记忆。Cloudflare D1 + Worker,40 行前端改动,聊天记录终于不会随浏览器关闭蒸发了。

然后我兴奋地想看看——自从上线以来,到底有多少人来跟数字 Polly 聊过天?

curl -s -H "Authorization: Bearer polly-cloudflare-xxx" \
     -H "User-Agent: polly-chat-query/1.0" \
     https://chat.polly.wang/v1/conversations | python3 -m json.tool

拿到一堆 JSON。好,再看某个对话的具体内容?

curl -s -H "Authorization: Bearer polly-cloudflare-xxx" \
     -H "User-Agent: polly-chat-query/1.0" \
     https://chat.polly.wang/v1/conversations/190e3a14-64be-4334-aac1-fea445fbc920 \
     | python3 -m json.tool

又一堆 JSON。想搜"小龙虾"在哪些对话里出现过?要把所有对话的消息都 curl 下来,然后 grep……

数据在云端,但约等于查不到。

这不行。

我需要什么

想了想,日常查聊天记录的场景大概就这几种:

我想做的实际操作
"看看昨天有没有人来聊天"list --date 2026-02-19
"展开看看这个对话聊了啥"show <uuid>
"有人问过翻译的事吗?"search 翻译
"上线以来整体数据怎么样?"stats
"导出备份一下"export --format csv

核心就是五个动作:列表、详情、搜索、统计、导出。

Claude Skill vs MCP Server

第一反应是做个 MCP Server——毕竟刚给 Master Translator 做了一个。但转念一想:

对比MCP ServerClaude Skill
调用方式Agent 自动发现工具Agent 读 SKILL.md 后调终端
启动成本需要服务进程零启动,直接调 Python
适合场景高频、多步骤交互低频、单次查询
编码量~500行(server+tools)~900行(脚本723+SKILL.md 173)

查聊天记录是个低频操作(一天查一两次),不需要常驻服务进程。Claude Skill 的方式更轻量——写个 Python 脚本,配个 SKILL.md 告诉 Claude 怎么调,完事。

方案设计阶段和小龙虾讨论了一圈。它分析了两种路线(扩展 Worker API vs 纯客户端脚本),最终建议走纯脚本 + Claude Skill,并提了 4 点具体意见:并发 fetch 解决 N+1 查询、加 --json flag 方便程序化调用、token 优先读环境变量、脚本路径要方便小龙虾直接 exec 调用。这不只是 Code Review,是方案共创。

架构:三层数据流

整个数据链路长这样:

访客在 polly.wang 聊天
  ↓ POST /v1/conversations/log
Cloudflare Worker(chat.polly.wang)
  ↓ 写入
Cloudflare D1(SQLite)
  ↓ GET /v1/conversations(需 Bearer Token)
query_chat_history.py(Python 脚本)
  ↓ 格式化输出
Claude / 终端 / OpenClaw

D1 里有两张表:conversations(对话元数据:时间、语言、UA)和 messages(逐条消息:角色、内容、时间戳)。Worker 提供了两个管理端点,需要 Bearer Token 鉴权。

脚本要做的就是:调 API → 拿 JSON → 过滤/搜索/统计 → 人类可读的输出。

命名:blog-chat

命名纠结了一下。现有的 Skill 体系是 blog-writer(写博文)和 blog-dev(改模板/CSS),新的查聊天记录自然叫 blog-chat——三个 blog-* Skill 形成一个完整的博客运维三件套:

Skill职责
blog-writer写内容
blog-dev改技术
blog-chat看数据

配置模式也直接复用了 blog-writer 的 config.env 方案:脚本同目录放一个 config.env,优先读环境变量,fallback 到文件。

实现:723 行,零依赖

整个脚本只用 Python 标准库,不装任何第三方包。这是个有意的选择——小龙虾(OpenClaw)需要通过 exec 直接调用这个脚本,零依赖意味着任何机器上拿到就能跑。

核心结构很简单:

# 配置加载:环境变量 → config.env fallback
def load_config(): ...

# API 客户端:带 Token 的 HTTP GET
class ChatAPI:
    def _request(self, path): ...
    def list_conversations(self): ...
    def get_conversation(self, conv_id): ...

# 5 个子命令
def cmd_list(api, args): ...    # 列表 + 日期过滤
def cmd_show(api, args): ...    # 对话详情渲染
def cmd_search(api, args): ...  # 并发获取 + 本地搜索
def cmd_stats(api, args): ...   # 统计 + ASCII 图表
def cmd_export(api, args): ...  # JSON/CSV 导出

几个值得说的设计决策:

并发获取:search 和 export 的 N+1 问题

搜索需要获取所有对话的消息内容,如果逐个串行请求会很慢。用 ThreadPoolExecutor 并发 8 路:

with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_REQUESTS) as pool:
    futures = {
        pool.submit(api.get_conversation, c["id"]): c
        for c in conversations
    }
    for future in as_completed(futures):
        conv = futures[future]
        data = future.result()
        # 在每个对话的消息中搜索关键词

这是小龙虾在 Code Review 时特别提的——"N+1 查询如果不并发,5 个对话要 5 秒,50 个对话要 50 秒"。

--json 全局 flag

小龙虾还建议加个 --json 全局 flag,方便程序化调用。实际实现时踩了个坑:argparse 的全局 flag 放在子命令前面没问题,放后面就不认了。最终用 parse_known_args() 解决:

parser, remaining = parser.parse_known_args()
if "--json" in remaining:
    json_mode = True

ANSI 彩色输出

默认输出是给人看的,带颜色和 emoji:

📋 Conversations (5 total)

  ID                                     Time               Msgs  Language
  ────────────────────── ────────────── ────  ──────
  63a94f06-c6c5-...     2026-02-16 20:48    4  zh-CN

加 --json 则输出原始 JSON,方便小龙虾或其他工具消费。

调试:三个坑

坑 1:urllib 被 Cloudflare WAF 拦截

Python urllib 的默认 User-Agent 是 Python-urllib/3.x,Cloudflare 的 Bot Fight Mode 直接返回 403。

修复:加一行 req.add_header("User-Agent", "polly-chat-query/1.0")。

这是个特别隐蔽的坑——curl 能用、浏览器能用、但 Python 脚本莫名其妙返回 403。如果不是在错误处理里区分了 401(认证失败)和 403(可能是 WAF),可能要排查很久。

坑 2:403 和 401 的区分

Worker 对认证失败返回 401,但 Cloudflare WAF 对被拦截的请求返回 403。修复前只 catch 了 401,导致 WAF 拦截时报的是"HTTP 403"而不是友好的认证失败提示。改成 if e.code in (401, 403) 就好了。

坑 3:--json flag 的位置

用户可能写 list --json --limit 5,也可能写 list --limit 5 --json。argparse 对后者会报错"unrecognized arguments"。用 parse_known_args() 手动处理剩余参数解决。

实战:过去一周的聊天数据

工具做完了,当然得自己先用上。查了一下 2 月 14 号到 20 号的数据:

python query_chat_history.py list --from 2026-02-14 --to 2026-02-20

5 个对话,31 条消息。拆开来看:

#时间消息数谁来的聊了啥
12/14 14:411小龙虾(测试 UA)D1 持久化的第一条测试消息
22/14 14:498小龙虾(Mac)测试持久化 + 后续回访聊 AI 记忆系统
32/15 10:3316小龙虾(Azure VM)搬家汇报 + 初一三 AI 联合拜年策划
42/16 09:112Windows 用户(en-US)只说了"worker d1",可能是技术搜索过来的
52/16 20:484iPhone 用户从拜年博文页面进来,聊了 AI 对话实验

有意思的发现:

  • 80% 的对话来自小龙虾。我的 AI 助手是我博客最活跃的"访客"。
  • 唯一的真实外部访客来自 iPhone,通过博文《马年三 AI 拜年》的页面入口进来——说明博文引流到聊天功能是有效的。
  • 那个 Windows 用户输入了"worker d1"就再没说话。可能是搜索 Cloudflare Worker + D1 技术方案时点进来的,看到是个聊天机器人就走了。

再跑个 stats 看看整体数据:

python query_chat_history.py stats
📊 Chat Statistics

  Conversations:  5
  Total Messages: 31
  Date Range:     2026-02-14 → 2026-02-16
  Active Days:    3
  Avg/Day:        1.7
  Busiest Day:    2026-02-14 (2 convs)
  Peak Hour:      14:00 UTC (2 convs)

  🌐 Languages:
    zh-CN        ████████████████░░░░   4 (80%)
    en-US        ████░░░░░░░░░░░░░░░░   1 (20%)

  📅 Daily Activity (last 3 days):
    2026-02-14 ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓ 2
    2026-02-15 ▓▓▓▓▓▓▓▓▓▓▓▓ 1
    2026-02-16 ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓ 2

数据量还不多,但至少证明了一件事:这个记忆系统在正常工作,而且已经有了真实外部访客的触达。

SKILL.md:教 Claude 怎么用

脚本写好了,还要写个 SKILL.md 告诉 Claude 怎么调。这是 Claude Skill 的核心——不是 Agent 自动发现工具,而是通过文档告诉 Agent "用户说什么的时候,该执行什么命令"。

核心是一张日期映射表:

用户说的转换为参数
"昨天"--date {yesterday}
"最近一周"--from {7_days_ago}
"2月份"--from 2026-02-01 --to 2026-02-28
"全部"不加日期参数

加上 5 个工作流(list → show → search → stats → export)的命令模板和输出解读方式,一共 173 行 SKILL.md。

收尾:吃自己的狗粮

这篇博文里"过去一周的聊天数据"那一节,就是用 blog-chat Skill 直接查的——我对 Copilot 说了一句"测试下,看看过去一周都有哪些聊天内容",它自动执行了 list 和 5 次 show,把结果整理成了表格。

最好的工具文档不是"它有多少功能",而是你刚才读的那段数据分析就是它吞出来的。

博客运维三件套完成 ✅:

blog-writer  → 写内容
blog-dev     → 改技术  
blog-chat    → 看数据  ← new!

下一步?等数据量大了,可能要在 Worker 层加全文搜索(FTS5)。但现在 5 个对话 31 条消息,并发获取 + 本地搜索绰绰有余。

先让子弹飞一会儿。

Comments