Категории

• Главная
• 1С
• 2040
• CG
• IOT умный дом
• Joomla
• Linux
• macOS
• Windows
• Безопасность
• Железо
• Защита данных
• Простотак
• Сети LAN
• Создание сайтов
• Электрика

Описание скрипта для рассылки уведомлений по сценарию

15.01.2026 | коды из категории: IOT умный дом

Основа системы — небольшой Python-скрипт, который раз в несколько минут проверяет базу данных и отправляет всё, что попало в «служебный» топик. Вот его ключевые части:

1. Чтение и удаление сообщений

Скрипт ищет все записи с топиком zigbee2mqtt/_alerts/email, забирает их содержимое и сразу удаляет из таблицы — чтобы не отправлять одно и то же письмо много раз.

SELECT id, value FROM sensor_data WHERE topic = 'zigbee2mqtt/_alerts/email';
DELETE FROM sensor_data WHERE id IN (...);

2. Гибкий формат сообщений

Скрипт понимает два формата:

• Простой текст — например, "Протечка в ванной!". В этом случае тема письма будет стандартной: «Уведомление из IoT».
• JSON — позволяет задать тему, тело и даже получателя:

  {
    "subject": "🚨 Авария!",
    "body": "Датчик воды сработал в 03:14",
    "to": "admin@example.com"
  }

3. Отправка через локальный SMTP

Скрипт использует smtplib и подключается к localhost:25 — то есть предполагается, что у вас на сервере настроен sendmail, Postfix или другой MTA. Это самый простой способ для домашнего сервера.

Если вы хотите использовать Gmail, Mailgun или другой внешний SMTP — достаточно заменить блок подключения (добавить логин/пароль, порт 587 и TLS). Но для начала локального SMTP более чем достаточно.

4. Запуск по расписанию

Сам скрипт ничего не слушает — он идемпотентный: запустили → обработал → вышел. Поэтому его легко вставить в cron:

# Проверять уведомления каждые 2 минуты
*/2 * * * * /home/user/alert_dispatcher.py >> /var/log/iot_alerts.log 2>&1

Альтернатива: не удалять, а помечать

Если вам важна история уведомлений, добавьте в таблицу колонку processed TINYINT DEFAULT 0 и замените DELETE на:

UPDATE sensor_data SET processed = 1 WHERE id IN (...);

А в SELECT добавьте условие WHERE ... AND processed = 0. Так вы сможете смотреть лог всех сработавших алертов — даже старых.

Полный код скрипта

Вот готовый к использованию вариант (не забудьте подставить свой email и пароль от БД!):

#!/home/mazzick/venv/bin/python3
"""
Универсальный диспетчер уведомлений.
Читает "служебные" сообщения из MySQL и отправляет их на email.
Требуется таблица sensor_data с колонками: id (INT AI), topic (VARCHAR), value (TEXT).
"""

import pymysql
import smtplib
import json
import sys
from email.message import EmailMessage

# === НАСТРОЙКИ ===
DB_CONFIG = {
    'host': 'localhost',
    'user': 'iot_user',
    'password': 'your_password_here',  # ← ЗАМЕНИ!
    'database': 'iot_db',
    'charset': 'utf8mb4'
}

ALERT_TOPIC = 'zigbee2mqtt/_alerts/email'
EMAIL_TO_DEFAULT = "your_email@example.com"  # ← ЗАМЕНИ!
SMTP_HOST = "localhost"

# =================

def fetch_and_delete_alerts():
    """Получает все сообщения из топика уведомлений и удаляет их."""
    conn = pymysql.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cursor:
            # Получаем все записи
            cursor.execute("""
                SELECT id, value FROM sensor_data
                WHERE topic = %s
                ORDER BY id ASC
            """, (ALERT_TOPIC,))
            rows = cursor.fetchall()

            if not rows:
                return []

            # Безопасное удаление по списку ID (параметризованный запрос)
            ids = [row[0] for row in rows]
            format_strings = ','.join(['%s'] * len(ids))
            cursor.execute(
                f"DELETE FROM sensor_data WHERE id IN ({format_strings})",
                ids
            )
            conn.commit()
            print(f"📨 Обработаны уведомления с ID: {ids}", file=sys.stderr)
            return [row[1] for row in rows]

    finally:
        conn.close()

def send_email(subject, body, to=None):
    msg = EmailMessage()
    msg.set_content(body)
    msg["Subject"] = subject
    msg["From"] = "iot-alert@local"
    msg["To"] = to or EMAIL_TO_DEFAULT

    try:
        with smtplib.SMTP(SMTP_HOST) as s:
            s.send_message(msg)
        print(f"✅ Email отправлен: {subject}")
    except Exception as e:
        print(f"❌ Ошибка SMTP: {e}", file=sys.stderr)

def main():
    alerts = fetch_and_delete_alerts()
    if not alerts:
        print("📭 Нет новых уведомлений")
        return

    for payload in alerts:
        try:
            data = json.loads(payload)
            subject = data.get("subject", "Уведомление IoT")
            body = data.get("body", payload)
            to = data.get("to")
        except (json.JSONDecodeError, TypeError):
            subject = "🚨 Уведомление из IoT"
            body = str(payload)
            to = None

        send_email(subject, body, to)

if __name__ == "__main__":
    main()

← Назад к списку

---