Python 监听 数据库表并触发通知

一、核心需求

当前项目需要监听 students 表。当学生出现三科满分时，先不真正对接邮箱或短信，只在终端打印一条模拟通知，让开发阶段可以确认触发逻辑是否生效。

监听脚本文件：

student_full_score_monitor.py

启动命令：

poetry run python student_full_score_monitor.py

脚本启动后会每隔几秒查询一次数据库。如果发现数学、英语、历史都是 100 分的学生，就打印：

[模拟通知] 2026-06-22 14:00:00 发现全满分学生: ID=1001 姓名=新同学 数学=100 英语=100 历史=100 总分=300

二、查询全满分学生

核心查询函数如下：

def find_full_score_students(conn: sqlite3.Connection) -> list[dict]:
    """查找三科都是 100 分的学生。"""
    cursor = conn.execute(
        """
        SELECT id, name, age, math, english, history, total
        FROM students
        WHERE math = 100 AND english = 100 AND history = 100
        ORDER BY id
        """
    )
    return [dict(row) for row in cursor.fetchall()]

这段代码做了两件事：

1. 从 students 表查找三科都是 100 分的学生。

2. 把 SQLite 查询结果转成 dict 列表，方便后面拼接通知内容。

因为查询条件是：

WHERE math = 100 AND english = 100 AND history = 100

所以项目里建立了联合索引：

CREATE INDEX IF NOT EXISTS idx_students_full_score
ON students (math, english, history);

这样监听脚本频繁查询时，可以尽量走索引。

三、模拟通知函数

通知函数目前没有真正发邮件或短信，而是打印到终端：

def mock_notify(student: dict) -> None:
    """模拟邮箱或短信通知，后续可替换为真实发送函数。"""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(
        "[模拟通知] "
        f"{now} 发现全满分学生: "
        f"ID={student['id']} 姓名={student['name']} "
        f"数学={student['math']} 英语={student['english']} 历史={student['history']} "
        f"总分={student['total']}"
    )

说明：

1. datetime.now() 用于记录触发时间。

2. student 里包含学生 ID、姓名、成绩和总分。

3. 后续如果接入邮箱或短信，只需要把 print() 换成真实发送逻辑。

例如后续可以替换成：

send_email(subject="发现全满分学生", content=message)

或者：

send_sms(phone="13800000000", content=message)

四、轮询监听逻辑

当前监听逻辑如下：

def monitor(interval: float) -> None:
    """轮询监听 students 表中的全满分记录。"""
    notified_ids: set[int] = set()
    print(f"开始监听 students 表，全满分触发模拟通知。轮询间隔: {interval} 秒")
    print("按 Ctrl+C 停止监听。")

    while True:
        try:
            with connect_students() as conn:
                ensure_student_indexes(conn)
                students = find_full_score_students(conn)
        except FileNotFoundError as err:
            print(f"数据库不存在: {err}")
            time.sleep(interval)
            continue

        for student in students:
            student_id = student["id"]
            if student_id in notified_ids:
                continue
            mock_notify(student)
            notified_ids.add(student_id)

        time.sleep(interval)

整体流程：

启动监听
  ↓
连接 student.db
  ↓
确保索引存在
  ↓
查询全满分学生
  ↓
未通知过则打印模拟通知
  ↓
等待 interval 秒
  ↓
继续下一轮

五、为什么要用 notified_ids

notified_ids 用来记录本次脚本运行中已经通知过的学生 ID：

notified_ids: set[int] = set()

如果 A 学生第一次满分：

查询到 A
  ↓
A 不在 notified_ids
  ↓
通知 A
  ↓
把 A 的 id 加入 notified_ids

如果下一轮仍然查询到 A：

查询到 A
  ↓
A 已经在 notified_ids
  ↓
跳过，不重复通知

这个设计可以避免同一个学生在同一次监听进程中反复刷屏。

六、真实系统里的异常通知规则

真实系统里监听的可能不是学生成绩，而是模块异常、接口失败率、磁盘空间、任务积压、参数超阈值等。

这种场景一般不会只用内存里的 set，而是建一张告警记录表：

CREATE TABLE alert_records (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    alert_key TEXT NOT NULL,
    status TEXT NOT NULL,
    first_triggered_at TEXT NOT NULL,
    last_seen_at TEXT NOT NULL,
    last_notified_at TEXT NOT NULL,
    notify_count INTEGER NOT NULL,
    recovered_at TEXT
);

字段说明：

alert_key           告警唯一标识，例如 module_a:param_a
status              active / recovered / acknowledged / closed
first_triggered_at  第一次异常时间
last_seen_at        最近一次仍然异常的时间
last_notified_at    最近一次通知时间
notify_count        通知次数
recovered_at        恢复时间

这种结构可以解决两个关键问题：

1. 怎么判断异常是否还没解决。

2. 怎么判断是否需要再次通知。

判断逻辑可以写成：

当前检测仍异常
并且 alert_records 里 status = active
  ↓
说明还没解决

当前检测恢复正常
并且 alert_records 里 status = active
  ↓
标记 recovered

十、当前 demo 和真实系统的区别

当前 demo：

内存 set 去重
脚本重启后状态丢失
只打印模拟通知
适合学习轮询和触发逻辑

真实系统：

数据库保存告警记录
支持重启后继续去重
支持持续异常重复提醒
支持恢复通知
支持确认、关闭、升级