如何实现读取手机通知并触发SMTP定时邮件告警？

一、核心方案选型 #

优先用Python实现，生态库齐全，能快速搞定通知监听、邮件发送、数据库存储和任务调度需求。手机与电脑的联动靠网络推送（而非SQL），SQL仅用于临时存储通知。

二、分步实现方案 #

1. 手机通知联动电脑 #

两种可行方案：

• 方案A（无开发成本）：用现成工具（如Pushbullet/IFTTT），设置规则：当目标游戏推送包含指定关键词的通知时，向电脑上的Python服务发送HTTP POST请求。
• 方案B（自定义）：开发轻量Android监听APP，通过系统NotificationListenerService捕获游戏通知，直接推送到电脑的Python后端。

2. 临时存储（SQLite） #

用轻量SQLite数据库存储通知，自动清理超过10分钟的记录，避免冗余数据。

3. 邮件告警逻辑 #

检测到关键词后，循环30次发送邮件，每次间隔5-10秒（随机避免被反垃圾拦截）。

4. 休眠唤醒任务调度 #

通过Windows任务计划程序，设置每20分钟唤醒电脑执行脚本，检查近期通知。

三、代码示例 #

1. Python后端服务（接收通知+实时告警） #

from flask import Flask, request, jsonify
import sqlite3
import time
import smtplib
from email.mime.text import MIMEText
import random

app = Flask(__name__)

# 初始化SQLite数据库
def init_db():
    conn = sqlite3.connect('notifications.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS notifications
                 (content TEXT, timestamp INTEGER)''')
    conn.commit()
    conn.close()

# 清理10分钟前的旧通知
def cleanup_old_notifications():
    conn = sqlite3.connect('notifications.db')
    c = conn.cursor()
    ten_min_ago = int(time.time()) - 600
    c.execute('DELETE FROM notifications WHERE timestamp < ?', (ten_min_ago,))
    conn.commit()
    conn.close()

# 发送告警邮件
def send_alert():
    smtp_server = 'smtp.example.com'
    smtp_port = 587
    sender = 'your_email@example.com'
    password = 'your_app_password'  # 用邮箱APP密码，别用原密码
    receiver = 'target_email@example.com'

    msg = MIMEText('游戏遭到攻击，请立即处理！')
    msg['Subject'] = '[紧急] 游戏攻击告警'
    msg['From'] = sender
    msg['To'] = receiver

    try:
        with smtplib.SMTP(smtp_server, smtp_port) as server:
            server.starttls()
            server.login(sender, password)
            server.send_message(msg)
    except Exception as e:
        print(f'邮件发送失败: {str(e)}')

# 接收手机推送的通知
@app.route('/push-notification', methods=['POST'])
def receive_notification():
    data = request.get_json()
    content = data.get('content', '')
    timestamp = int(time.time())

    # 存储通知
    conn = sqlite3.connect('notifications.db')
    c = conn.cursor()
    c.execute('INSERT INTO notifications VALUES (?, ?)', (content, timestamp))
    conn.commit()
    conn.close()

    # 检测关键词并触发告警
    if "you're being attacked" in content.lower():
        for _ in range(30):
            send_alert()
            time.sleep(random.randint(5, 10))

    return jsonify({'status': 'success'})

if __name__ == '__main__':
    init_db()
    cleanup_old_notifications()
    # 监听所有网卡，让手机能访问
    app.run(host='0.0.0.0', port=5000, debug=False)

2. 任务计划检查脚本（休眠唤醒后执行） #

单独写一个脚本，供Windows任务计划调用：

import sqlite3
import time
import smtplib
from email.mime.text import MIMEText
import random

def cleanup_old_notifications():
    conn = sqlite3.connect('notifications.db')
    c = conn.cursor()
    ten_min_ago = int(time.time()) - 600
    c.execute('DELETE FROM notifications WHERE timestamp < ?', (ten_min_ago,))
    conn.commit()
    conn.close()

def send_alert():
    smtp_server = 'smtp.example.com'
    smtp_port = 587
    sender = 'your_email@example.com'
    password = 'your_app_password'
    receiver = 'target_email@example.com'

    msg = MIMEText('游戏遭到攻击，请立即处理！')
    msg['Subject'] = '[紧急] 游戏攻击告警'
    msg['From'] = sender
    msg['To'] = receiver

    try:
        with smtplib.SMTP(smtp_server, smtp_port) as server:
            server.starttls()
            server.login(sender, password)
            server.send_message(msg)
    except Exception as e:
        print(f'邮件发送失败: {str(e)}')

def check_recent_alerts():
    conn = sqlite3.connect('notifications.db')
    c = conn.cursor()
    ten_min_ago = int(time.time()) - 600
    c.execute('SELECT content FROM notifications WHERE timestamp >= ? AND content LIKE ?',
              (ten_min_ago, "%you're being attacked%"))
    results = c.fetchall()
    conn.close()

    if results:
        for _ in range(30):
            send_alert()
            time.sleep(random.randint(5, 10))

if __name__ == '__main__':
    cleanup_old_notifications()
    check_recent_alerts()

3. Android通知监听核心代码（Kotlin） #

如果自定义APP，核心逻辑如下：

class GameNotificationListener : NotificationListenerService() {
    override fun onNotificationPosted(sbn: StatusBarNotification) {
        super.onNotificationPosted(sbn)
        // 替换为目标游戏的包名
        if (sbn.packageName == "com.your.game.package") {
            val content = sbn.notification.extras.getString(android.app.Notification.EXTRA_TEXT) ?: ""
            if (content.contains("you're being attacked", ignoreCase = true)) {
                // 发送POST到电脑后端
                val url = "http://你的电脑IP:5000/push-notification"
                val requestBody = JSONObject().put("content", content).toString()
                val client = OkHttpClient()
                val body = RequestBody.create(MediaType.parse("application/json"), requestBody)
                val req = Request.Builder().url(url).post(body).build()
                client.newCall(req).enqueue(object : Callback {
                    override fun onFailure(call: Call, e: IOException) {}
                    override fun onResponse(call: Call, response: Response) {}
                })
            }
        }
    }
}

四、Windows任务计划设置 #

1. 创建基本任务，触发器选择「每天」，设置每20分钟重复一次，持续时间选「无限期」
2. 操作选「启动程序」，程序路径填Python安装路径（如C:\Python39\python.exe），添加参数填检查脚本的完整路径（如D:\scripts\check_alerts.py）
3. 进入「条件」选项卡，勾选「唤醒计算机运行此任务」
4. 「设置」选项卡中，可勾选任务失败后的重试规则

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