如何实现SQLite到Python服务的实时告警：当视图distance达50时触发

核心问题分析 #

你之前的AFTER UPDATE触发器返回旧数据，大概率是因为视图的计算列（minutes_unseen/distance）动态依赖当前时间，触发器直接查询视图时，可能因事务未提交或未实时计算时间差值导致拿到非最新结果；另外若触发器误用OLD行数据而非NEW行数据，也会出现数值不准确的情况。

以下是三个无需轮询的实时告警可行方案：

方案1：在触发器中实时计算阈值并调用Python告警逻辑 #

直接在data表的更新触发器中，基于NEW.last_seen（更新后的最新值）计算当前的minutes_unseen和distance，判断触发阈值后调用Python自定义函数执行告警。

步骤1：注册Python自定义告警函数

在Python连接SQLite时，注册一个可被SQLite触发器调用的函数：

import sqlite3
import time

def send_alert(sensor_id, current_distance):
    # 替换为你的实际告警逻辑（如发邮件、推消息到MQ等）
    print(f"[ALERT] Sensor {sensor_id} distance reached {current_distance} (≥50) at {time.strftime('%Y-%m-%d %H:%M:%S')}")

# 连接数据库，开启WAL模式（优化多线程读写）
conn = sqlite3.connect('sensor_data.db', check_same_thread=False)
# 注册自定义函数：函数名、参数个数、回调函数
conn.create_function('trigger_alert', 2, send_alert)

步骤2：创建触发器

在SQLite中创建AFTER UPDATE触发器，实时计算并判断阈值：

CREATE TRIGGER IF NOT EXISTS sensor_distance_alert
AFTER UPDATE OF last_seen ON data
BEGIN
    -- 计算当前未检测时长（分钟）和对应的distance
    SELECT trigger_alert(
        NEW.id,
        -- 替换为你的distance计算逻辑，示例为线性增长到100
        MIN(100, (strftime('%s', 'now') - strftime('%s', NEW.last_seen)) / 60 * 2)
    )
    -- 只有当distance≥50时触发告警
    WHERE MIN(100, (strftime('%s', 'now') - strftime('%s', NEW.last_seen)) / 60 * 2) >= 50;
END;

注意事项

• 确保last_seen的格式能被strftime('%s', ...)正确解析（推荐ISO格式：YYYY-MM-DD HH:MM:SS）。
• 告警逻辑如果耗时较长，建议在send_alert中启动异步任务（如用threading或asyncio），避免阻塞写入线程。

方案2：将distance计算逻辑移至写入环节，监控存储列 #

如果不想在触发器中做复杂计算，可以在每次更新last_seen时，直接计算并更新data表的distance列，然后触发器监控该列的变化。

步骤1：修改写入逻辑

每次更新传感器数据时，同步计算distance：

def update_sensor_data(sensor_id, new_last_seen):
    # 计算当前distance
    calc_sql = """
    SELECT MIN(100, (strftime('%s', 'now') - strftime('%s', ?)) / 60 * 2)
    """
    distance = conn.execute(calc_sql, (new_last_seen,)).fetchone()[0]
    
    # 更新data表
    update_sql = """
    UPDATE data SET last_seen = ?, distance = ? WHERE id = ?
    """
    conn.execute(update_sql, (new_last_seen, distance, sensor_id))
    conn.commit()

步骤2：创建简化的触发器

CREATE TRIGGER IF NOT EXISTS sensor_distance_alert
AFTER UPDATE OF distance ON data
BEGIN
    SELECT trigger_alert(NEW.id, NEW.distance) WHERE NEW.distance >= 50;
END;

这个方案的优势是计算逻辑更清晰，触发器无需重复计算，减少出错概率。

方案3：利用SQLite WAL日志监听（进阶） #

如果需要监控所有可能触发阈值的情况（包括未更新last_seen但时间推移导致distance自然增长到50的场景），可以开启SQLite的WAL模式，通过监听WAL日志的变化，仅检查被修改的传感器行是否达到阈值：

1. 开启WAL模式：conn.execute('PRAGMA journal_mode=WAL')
2. 用Python的sqlite3或轻量工具监听WAL日志，当有写入操作时，针对修改的传感器ID实时计算当前distance并判断阈值。

关键注意点 #

• 多线程读写时，务必开启SQLite的WAL模式（PRAGMA journal_mode=WAL），这是SQLite支持多线程读单线程写的前提。
• 触发器中的计算逻辑要和data_view中的逻辑完全一致，避免出现视图显示和触发器判断不一致的情况。
• 若使用自定义函数，确保SQLite连接的check_same_thread=False，但需严格遵守单线程写入的约束，避免多线程写入冲突。

内容的提问来源于stack exchange，提问作者dev_99_kind