用 SQLite 存监控历史：从 JSON 文件迁到单文件数据库

JSON 文件方案的三个瓶颈

很多小监控脚本是这么存的：latest.json 只保留最新一次结果，events.jsonl 追加补货、告警之类的事件。跑几天没问题，时间一长就会遇到三个问题。

• 查不动。想知道“100 元档最近 24 小时的库存曲线”，得把整个 jsonl 读进内存再过滤，文件到几十 MB 就开始卡。
• 没法聚合。“这个月检查了多少次、漏了几次”“补货集中在哪个时间段”，用文本文件只能自己写循环。
• 并发有风险。cron 在写、你手动在查，或者两个进程同时追加，轻则读到半行，重则互相覆盖。

这些正好是 SQLite 擅长的：单文件、零服务、Python 标准库自带 sqlite3，还支持索引、事务和聚合查询。

表结构：只留必要的字段

CREATE TABLE IF NOT EXISTS checks (
  id        INTEGER PRIMARY KEY AUTOINCREMENT,
  ts        TEXT    NOT NULL,          -- UTC ISO8601
  tier      INTEGER NOT NULL,          -- 500/200/100
  stock     INTEGER NOT NULL DEFAULT 0,
  has_stock INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_checks_ts ON checks(ts);
CREATE INDEX IF NOT EXISTS idx_checks_tier_ts ON checks(tier, ts);

CREATE TABLE IF NOT EXISTS events (
  id    INTEGER PRIMARY KEY AUTOINCREMENT,
  ts    TEXT    NOT NULL,
  kind  TEXT    NOT NULL,              -- restock / stock_up / error
  tier  INTEGER NOT NULL,
  stock INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts);

两个细节值得强调。第一，时间统一存 UTC 的 ISO8601 字符串，展示时再转本地时区，避免夏令时和跨机器时区把数据搅乱。第二，has_stock 单独存一列，不要用 stock > 0 反推——平台可能返回“有货但数量未知”的状态。

并发：WAL 加 busy_timeout

cron 写入和手动查询撞在一起时，默认的 rollback journal 会直接报 database is locked。加三行配置就能大幅缓解：

import sqlite3

conn = sqlite3.connect("/var/lib/monitor/monitor.db", timeout=10)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute("PRAGMA busy_timeout=5000")

WAL 允许“一写多读”并行，busy_timeout 让写锁冲突时先等 5 秒而不是立刻失败。监控这种低频写入场景，这套配置足够了。

写入：一次检查一个事务

from datetime import datetime, timezone

def record(conn, snapshot):
    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
    rows = [(ts, tier, s["stock"], int(s["has_stock"]))
            for tier, s in snapshot.items()]
    with conn:
        conn.executemany(
            "INSERT INTO checks(ts, tier, stock, has_stock) VALUES (?,?,?,?)",
            rows,
        )

with conn 会自动开事务并在异常时回滚，比手动 commit() 稳。注意别在循环里逐条 commit，每 10 分钟三次插入合成一个事务，写放大和锁竞争都会小很多。

三个马上能用的查询

1）某档位最近的库存变化

SELECT ts, stock FROM checks
WHERE tier = 100
ORDER BY ts DESC LIMIT 20;

2）今天有没有漏跑——正常每 10 分钟一次，一天应该是 144 次：

SELECT substr(ts, 1, 10) AS day, COUNT(*) AS n
FROM checks
GROUP BY day
ORDER BY day DESC LIMIT 7;

3）补货事件按小时分布，用来判断平台一般什么时候放量：

SELECT substr(ts, 12, 2) AS hour, COUNT(*) AS n
FROM events
WHERE kind = 'restock'
GROUP BY hour
ORDER BY n DESC;

从 JSONL 平滑迁移

不用停机。写一个一次性脚本，把历史事件读进来逐条插入，插入前先按 (ts, kind, tier) 查重，避免重复跑导致数据翻倍：

import json

with open("events.jsonl", encoding="utf-8") as f:
    for line in f:
        e = json.loads(line)
        conn.execute(
            """INSERT INTO events(ts, kind, tier, stock)
               SELECT ?,?,?,? WHERE NOT EXISTS (
                 SELECT 1 FROM events WHERE ts=? AND kind=? AND tier=?)""",
            (e["at"], e["kind"], e["tier"], e["stock"],
             e["at"], e["kind"], e["tier"]),
        )
conn.commit()

迁移完成后先并行跑一周：新数据写库，同时继续写 jsonl 兜底；确认查询结果对得上，再把 jsonl 关掉。

保留策略与备份

SQLite 不会自动瘦身。建议按天清理明细、保留聚合结果：

DELETE FROM checks WHERE ts < datetime('now', '-90 days');
VACUUM;

备份更简单，用 sqlite3 monitor.db ".backup /backup/monitor-$(date +%F).db" 就得到一致性快照，比复制正在写入的文件安全。

什么时候不该用

如果监控对象变成几十台机器、每秒都有数据，或者需要多台机器同时写同一个库，那就该上 Prometheus、VictoriaMetrics 这类时序库。SQLite 适合的是单机、低频、要长期留历史的场景——小脚本监控正好落在这个区间。

从 JSON 换到 SQLite，改动量通常只有几十行，但之后每一次“查一下上个月”的请求，都会感谢当初的自己。