SQLite 状态管理
当重复性任务需要记住它已经完成的工作——比如已经告警过的项目、已处理的最后游标——从而避免在下一个 tick 中重复通知或重复抓取时，请使用此技能。一个标准库中的 SQLite 文件即可提供持久化、带文件锁的状态，无需服务器，也无需额外依赖。
何时调用
- 用户说：“不要重复告警同一个东西”、“跨运行去重”、“记住最后 ID”、“从中断处恢复”。
- 对话中的代码是一个 cron/计划脚本，当前每次运行都会重新处理所有内容。
何时不应调用
- 状态很小且仅在单次运行中短暂存在——此时普通的集合/字典就足够了。
- 多台机器需要并发共享状态——请使用真正的服务器数据库，而不是单个文件。
具体示例
用户输入：
我的新闻爬虫每小时都会重复发送相同的标题。让它只发送新的。
输出：
from state import filter_new, mark_seen
db = "news.db"
fresh = filter_new(db, [a["url"] for a in articles]) # 只保留未见的 URL
for url in fresh:
send_alert(url)
mark_seen(db, *fresh) # 记录它们，供下次运行使用
应用模式
- 使用稳定且唯一的键来标识每个项目（URL、ID、内容哈希），而不是数组位置。
- 使用
filter_new决定要执行的操作，仅在操作成功后才调用mark_seen。 - 将进度存储为命名游标（
set_cursor/get_cursor），以便可恢复的爬虫能够从中断处重新开始。 - 启用 WAL 模式，实现耐久写入和并发读取；保持单一写入者。
- 备份或版本化管理单个
.db文件——它是整个任务的记忆。
参考：assets/state.py。
来源
提炼自作者在其自动化项目中的生产实践。v1.0.0。 另见：[[pipeline-orchestrator]]、[[cron-dispatch]]、[[webhook-receiver]]。
→ 使用 Trawlkit 构建完整可运行的机器人。