Категории

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

Скрипт-диспетчер для рассылки алертов по датчикам в Telegram и емаил

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

Вот улучшенная версия alert_dispatcher.py, которая умеет работать с несколькими каналами.

Требования

• Python 3.6+
• Библиотеки: pymysql, requests (для Telegram)
• Локальный SMTP (для email)
• Telegram-бот (создаётся через @BotFather)

Настройка Telegram

1. Создайте бота через @BotFather — получите токен вида 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11.
2. Напишите боту любое сообщение (например, /start).
3. Получите chat_id — можно через запрос:

   curl "https://api.telegram.org/bot<ТОКЕН>/getUpdates"

   В ответе будет "chat":{"id":123456789,...}.

Код скрипта

Код скрипта

#!/home/mazzick/venv/bin/python3
"""
Универсальный диспетчер уведомлений с поддержкой email и Telegram.
Читает сообщения из zigbee2mqtt/_alerts и маршрутизирует их.
"""

import pymysql
import smtplib
import requests
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"

# Telegram
TELEGRAM_BOT_TOKEN = "123456:ABC-DEF..."  # ← ЗАМЕНИ!
TELEGRAM_CHAT_ID = "123456789"             # ← ЗАМЕНИ!

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

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 []

            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"❌ Email error: {e}", file=sys.stderr)

def send_telegram(text):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {
        "chat_id": TELEGRAM_CHAT_ID,
        "text": text,
        "parse_mode": "HTML"
    }
    try:
        resp = requests.post(url, json=payload, timeout=10)
        if resp.status_code == 200:
            print(f"✅ Telegram: {text[:30]}...")
        else:
            print(f"❌ Telegram error: {resp.text}", file=sys.stderr)
    except Exception as e:
        print(f"❌ Telegram exception: {e}", file=sys.stderr)

def dispatch_alert(payload):
    try:
        data = json.loads(payload)
    except (json.JSONDecodeError, TypeError):
        print(f"⚠️ Неверный JSON: {payload}", file=sys.stderr)
        return

    channel = data.get("channel")
    if not channel:
        print(f"⚠️ Не указан канал: {payload}", file=sys.stderr)
        return

    if channel == "email":
        subject = data.get("subject", "Уведомление IoT")
        body = data.get("body", "Нет текста")
        to = data.get("to")
        send_email(subject, body, to)

    elif channel == "telegram":
        text = data.get("text", "[пустое сообщение]")
        send_telegram(text)

    else:
        print(f"❓ Неизвестный канал: {channel}", file=sys.stderr)

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

    for payload in alerts:
        dispatch_alert(payload)

if __name__ == "__main__":
    main()

Что дальше?

• Добавить поддержку pushover, discord, signal-cli.
• Сделать историю уведомлений (не удалять, а помечать).
• Добавить повторные попытки при ошибке сети.

Теперь моя система уведомлений стала по-настоящему универсальной — и при этом осталась простой, надёжной и полностью под моим контролем.

---

P.S. Код открыт для доработки — делитесь своими улучшениями в комментариях!

Безопасность: как не засветить токены и пароли

Когда вы пишете скрипт, который отправляет email или Telegram-сообщения, рано или поздно приходится хранить секретные данные: пароли от базы, токены ботов, SMTP-ключи. И главная ошибка — прописать их прямо в коде:

# ПЛОХО! Никогда так не делайте в реальных проектах
TELEGRAM_BOT_TOKEN = "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11"

Почему это плохо?

• Если вы случайно опубликуете код в GitHub — токен станет публичным.
• Любой, кто получит доступ к файлу (через бэкап, ошибку веб-сервера, общий хостинг), сможет угнать вашего бота.
• Вы не сможете легко менять настройки без правки кода.

Решение: файл .env

Создайте рядом со скриптом простой текстовый файл .env:

# .env
TELEGRAM_BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11
TELEGRAM_CHAT_ID=987654321
DB_PASSWORD=1212121212
EMAIL_TO=9447333@gmail.com

А в самом скрипте читайте эти значения через переменные окружения. Вот минимальная реализация без сторонних библиотек:

import os

# Загружаем .env, если он есть
env_path = os.path.join(os.path.dirname(__file__), '.env')
if os.path.exists(env_path):
    with open(env_path) as f:
        for line in f:
            if line.strip() and not line.startswith('#'):
                key, value = line.strip().split('=', 1)
                os.environ[key] = value

# Используем
TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')

Не забудьте про .gitignore!

Чтобы случайно не отправить секреты в репозиторий, добавьте в файл .gitignore:

.env
*.log
__pycache__/

Теперь Git будет игнорировать .env, а ваши токены останутся только на вашем сервере.

Что делать, если токен уже засветился?

1. Срочно зайдите в @BotFather.
2. Отправьте команду /revoke.
3. Выберите бота — Telegram выдаст новый токен, а старый станет недействительным.
4. Обновите токен в своём .env.

Эти простые шаги уберегут вас от спама, потери бота и утечки служебной информации.

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

---