Python ile MQTT E-posta Uyarıları: Hız Sınırlı SMTP Bildirimleri

Python ile MQTT e-posta uyarıları, kimse bir panoya bakmıyorken bile doğru kişinin bir sorundan haberdar olmasını sağlamanın güvenilir bir yoludur. Bu eğitimde alarm ve telemetri topic'lerine abone olan, eşikleri histerezisle değerlendiren ve gürültülü bir sensörün gelen kutusunu asla doldurmaması için bekleme süreleri ve özetlerle SMTP e-postaları gönderen küçük bir servis geliştireceksiniz. Ayrıca bir broker'ın yerleşik uyarı kurallarının nereye oturduğunu ve özel bir betiğin hâlâ nerede daha iyi bir araç olduğunu göreceksiniz.

İyi Bir Uyarı E-postası Servisinin İhtiyaçları

on_message içinden e-posta göndermek beş satır sürer. İnsanların görmezden gelmeyi öğrenmediği e-postalar göndermek ise biraz daha fazla tasarım gerektirir. Kullanışlı bir servis şunlara sahiptir:

Topic Düzeni

Servis, public MQTT test broker üzerinde bir test öneki altındaki iki tür topic'i dinler:

connectmq/demo/mail-6630/<device>/telemetry   {"temperature": 72.5, "pressure": 2.1}
connectmq/demo/mail-6630/<device>/alarm       {"severity": "critical", "text": "Door open"}

Telemetri, betikte tanımlanan eşiklere göre kontrol edilir. Alarm mesajları ise cihazın kendi başına karar verdiği olaylardır ve bekleme süresine tabi olarak olduğu gibi iletilir. broker.connectmq.com üzerindeki public alan herkes tarafından okunabilir; bu yüzden kendi rastgele önekinizi seçin ve gerçek adresleri ve verileri test payload'larının dışında tutun.

SMTP Kimlik Bilgilerini Hazırlayın

Kimliği doğrulanmış gönderimleri kabul eden bir SMTP sunucusuna ihtiyacınız var. Tipik seçenekler şirketinizin posta sunucusu, bir işlemsel e-posta sağlayıcısı veya uygulama parolalı kişisel bir posta kutusudur. Şunları toplayın:

• Sunucu ve port: genellikle STARTTLS ile 587 veya örtük TLS ile 465
• Kullanıcı adı ve parola (hesap iki faktörlü kimlik doğrulama kullanıyorsa uygulamaya özel parola)
• Sunucunun göndermeye izin verdiği bir gönderici adresi

Bunları asla betikte değil, ortam değişkenlerinde saklayın.

Python ile MQTT E-posta Uyarıları Servisi

Tek üçüncü taraf bağımlılığı kurun:

pip install "paho-mqtt>=2.0"

Aşağıdakini mqtt_email_alerts.py olarak kaydedin:

import json
import os
import queue
import smtplib
import ssl
import threading
import time
import uuid
from email.message import EmailMessage

import paho.mqtt.client as mqtt

# --- MQTT ayarları ---
BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"mail-alerts-{uuid.uuid4().hex[:8]}")
PREFIX = os.environ.get("TOPIC_PREFIX", "connectmq/demo/mail-6630")

# --- SMTP ayarları ---
SMTP_HOST = os.environ["SMTP_HOST"]
SMTP_PORT = int(os.environ.get("SMTP_PORT", "587"))
SMTP_USER = os.environ["SMTP_USER"]
SMTP_PASS = os.environ["SMTP_PASS"]
MAIL_FROM = os.environ.get("MAIL_FROM", SMTP_USER)
MAIL_TO = [a.strip() for a in os.environ["MAIL_TO"].split(",")]
COOLDOWN = int(os.environ.get("COOLDOWN_MINUTES", "15")) * 60

# metrik -> (üzerinde tetikle, altında temizle)
THRESHOLDS = {
    "temperature": (80.0, 75.0),
    "pressure": (3.0, 2.7),
}

outbox: "queue.Queue[EmailMessage]" = queue.Queue()
active = {}      # (cihaz, metrik) -> alarm durumundayken True
last_sent = {}   # anahtar -> son e-postanın zaman damgası


def build_email(subject: str, body: str) -> EmailMessage:
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = MAIL_FROM
    msg["To"] = ", ".join(MAIL_TO)
    msg.set_content(body)
    return msg


def queue_email(key, subject, body, force=False):
    now = time.time()
    if not force and now - last_sent.get(key, 0) < COOLDOWN:
        print(f"Cooldown active for {key}, not emailing")
        return
    last_sent[key] = now
    outbox.put(build_email(subject, body))


def smtp_worker():
    context = ssl.create_default_context()
    while True:
        msg = outbox.get()
        for attempt in range(1, 4):
            try:
                if SMTP_PORT == 465:
                    server = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, context=context, timeout=20)
                else:
                    server = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=20)
                    server.starttls(context=context)
                with server:
                    server.login(SMTP_USER, SMTP_PASS)
                    server.send_message(msg)
                print(f"Sent: {msg['Subject']}")
                break
            except (smtplib.SMTPException, OSError) as exc:
                print(f"SMTP attempt {attempt} failed: {exc}")
                time.sleep(10 * attempt)


def check_thresholds(device: str, data: dict):
    for metric, (trigger, clear) in THRESHOLDS.items():
        value = data.get(metric)
        if not isinstance(value, (int, float)):
            continue
        key = (device, metric)
        if not active.get(key) and value > trigger:
            active[key] = True
            queue_email(key, f"[ALERT] {device} {metric} = {value}",
                        f"{device} reported {metric} = {value}, above the limit of {trigger}.\n"
                        f"Topic: {PREFIX}/{device}/telemetry\nTime: {time.strftime('%Y-%m-%d %H:%M:%S %Z')}")
        elif active.get(key) and value < clear:
            active[key] = False
            # çözüm bildirimini bekleme süresi içinde olsa bile her zaman gönder
            queue_email(key, f"[RESOLVED] {device} {metric} = {value}",
                        f"{device} {metric} is back below {clear} (now {value}).", force=True)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"MQTT connect failed: {reason_code}")
        return
    client.subscribe(f"{PREFIX}/+/telemetry", qos=0)
    client.subscribe(f"{PREFIX}/+/alarm", qos=1)
    print("Subscribed to telemetry and alarm topics")


def on_message(client, userdata, msg):
    device, kind = msg.topic.split("/")[-2:]
    try:
        data = json.loads(msg.payload)
    except (json.JSONDecodeError, UnicodeDecodeError):
        print(f"Ignoring non-JSON payload on {msg.topic}")
        return
    if not isinstance(data, dict):
        return
    if kind == "telemetry":
        check_thresholds(device, data)
    elif kind == "alarm" and not msg.retain:
        severity = str(data.get("severity", "alarm")).upper()
        queue_email((device, "alarm"), f"[{severity}] {device}: {data.get('text', 'alarm')}",
                    json.dumps(data, indent=2))


def main():
    threading.Thread(target=smtp_worker, daemon=True).start()
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)
    client.connect(BROKER, PORT, keepalive=60)
    client.loop_forever(retry_first_connection=True)


if __name__ == "__main__":
    main()

Mantık Nasıl Çalışır?

Tüm MQTT callback'leri paho'nun ağ iş parçacığında çalışır; bu yüzden active ve last_sent yalnızca tek bir iş parçacığından değiştirilir ve kilit gerektirmez. SMTP worker'ı posta sunucusuna olan ağ bağlantısını yönetir ve artan gecikmelerle üç kereye kadar yeniden dener. Posta sunucunuz yavaşsa veya kısa süreliğine erişilemiyorsa MQTT akmaya devam eder ve e-postalar yalnızca gecikmeli ulaşır.

En değerli kısım histerezistir. 80 civarında dolaşan bir sıcaklık, aksi halde çizgiyi her iki yönde geçtikçe birkaç saniyede bir uyarı üretirdi. Tetikleme 80'de ve temizleme 75'te olduğunda, değer yükseldiğinde bir uyarı ve gerçekten düzeldiğinde bir çözüm bildirimi alırsınız.

Uçtan Uca Test Edin

export SMTP_HOST=smtp.example.com SMTP_PORT=587
export SMTP_USER=alerts@example.com SMTP_PASS='app-password'
export MAIL_TO=you@example.com
python mqtt_email_alerts.py

Sınırın üzerinde bir değer, ardından düzelmiş bir değer, ardından bir cihaz alarmı yayınlayın:

mosquitto_pub -h broker.connectmq.com -p 1883 -i mail-pub-6630 \
  -t connectmq/demo/mail-6630/oven2/telemetry -m '{"temperature": 84.1}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i mail-pub-6630 \
  -t connectmq/demo/mail-6630/oven2/telemetry -m '{"temperature": 72.0}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i mail-pub-6630 \
  -t connectmq/demo/mail-6630/oven2/alarm -m '{"severity": "critical", "text": "Door open"}'

Üç e-posta almalısınız: bir uyarı, bir çözüm bildirimi ve kritik alarm. 84.1 değerini birkaç kez daha yayınlayın ve başka hiçbir şey gelmediğine dikkat edin; durum makinesi alarmın zaten etkin olduğunu bilir.

Gürültülü Sahalar İçin Özet E-postalar

Bekleme süreleri tek bir cihazın gelen kutusunu doldurmasını engeller, ancak elli cihazlı bir saha bir elektrik dalgalanması sırasında yine de elli ayrı e-posta üretebilir. Çok sayıda uyarı genellikle birlikte geliyorsa anlık e-postalardan özet (digest) e-postalara geçin. Fikir basittir: her e-postayı doğrudan giden kutusuna koymak yerine bir listeye bir satır ekleyin; worker her birkaç dakikada bir, olan her şeyi cihaza göre gruplayarak özetleyen tek bir mesaj göndersin. Kritik alarmlar özeti atlayarak yine saniyeler içinde ulaşabilir.

Özet e-postalar uyarı yorgunluğunu da görünür kılar. Aynı üç sensör her özette görünüyorsa eşikler yanlıştır veya sensörlerin bakıma ihtiyacı vardır; bu da posta istemcisine bir filtre kuralı eklemekle değil, ekiple konuşarak çözülmesi gereken bir konudur. Eşik sözlüğünü düzenli olarak gözden geçirin ve geliştirici olmayan kişilerin sınırları ayarlaması gerektiğinde onu küçük bir YAML veya JSON dosyasına taşıyın; böylece değişiklikler kod düzenlemeyi gerektirmez.

Yerleşik Uyarı Kuralları ve Python Betiği Karşılaştırması

ConnectMQ, portalda eşik tabanlı uyarı kuralları ve bir uyarı günlüğü içerir. Bunlar, broker'ınız genelinde bir değerin sınırı ne zaman aştığını görmenin, dağıtılacak hiçbir kod olmadan en hızlı yoludur. Portalın dışında kalan davranışlara ihtiyaç duyduğunuzda özel bir betik yine de değerlidir:

Yaygın bir kurulum ikisini birlikte kullanır: devreye alma ve operasyon sırasında görünürlük için portaldaki kurallar, insanların gelen kutusuna ulaşması gereken bildirimler için ise bunun gibi bir betik.

Sorun Giderme Kontrol Listesi

1. SMTPAuthenticationError. Hesabın bir uygulama parolasına ihtiyacı var veya hesap için SMTP gönderimi devre dışı.
2. 587 veya 465 portunda bağlantı zaman aşımları. Bazı bulut sağlayıcıları ve ağlar giden SMTP'yi engeller. Diğer portu veya bir işlemsel e-posta sağlayıcısını deneyin.
3. E-postalar spam klasörüne düşüyor. Gönderici alan adında SMTP sunucunuz için SPF ve DKIM kayıtları yok. Kontrol ettiğiniz bir alan adından gönderin ve bunları yapılandırın.
4. Bir alarm hiç çözülmüyor. Değer hiçbir zaman temizleme seviyesinin altına düşmüyor veya cihaz telemetri göndermeyi bıraktı. N dakika boyunca hiç mesaj gelmediğinde bir "eski veri" uyarısı eklemeyi düşünün.
5. Yeniden başlatmalardan sonra yinelenen uyarılar. Durum bellekte tutuluyor. Uzun süre çalışan kurulumlar için active değerini küçük bir JSON dosyasında veya SQLite'ta kalıcı hale getirin.

Üretim Kurulumu

Servisi, ConnectMQ portalında üretilen kullanıcı adı, parola ve client ID ile 8883 portunda kimlik bilgileri ve TLS kullanan private bir broker'a karşı çalıştırın. Yönetilen bir MQTT broker üzerinde broker tarafı sizin için işletilir; böylece işiniz bu tek süreci systemd veya bir container yeniden başlatma politikasıyla ayakta tutmaya indirgenir. E-posta yerine sohbeti tercih ediyorsanız MQTT Telegram botu aynı yapıyı kullanır; her ölçümün denetim kaydına ihtiyacınız varsa bunu MQTT mesajlarını CSV'ye kaydetme ile birleştirin. Uzun süre çalışan servislerde paho'nun yeniden bağlanma davranışı için paho-mqtt yeniden bağlanma yönetimi yazısına, Python dışındaki kütüphane seçenekleri için ise MQTT client kütüphaneleri rehberine bakın.

ConnectMQ ile Başlayın

Uyarı servisiniz için dakikalar içinde private bir broker kurun. app.connectmq.com adresinden kaydolun, kredi kartı gerektirmeyen ücretsiz Serverless planını seçin, bir broker oluşturun ve üretilen kimlik bilgilerini yukarıdaki ortam değişkenlerinde kullanın. Ardından portalda bir eşik uyarı kuralı ekleyin ve onu Python uyarılarınızla yan yana karşılaştırın.

Sık Sorulan Sorular

paho-mqtt doğrudan e-posta gönderebilir mi?

Hayır. paho-mqtt yalnızca MQTT konuşur. Gönderimi Python'un standart smtplib ve email modülleri üstlenir; betik ise bir mesaj bir koşulla eşleştiğinde bir e-postayı kuyruğa alarak ikisini birbirine bağlar.

Yüzlerce uyarı e-postası almaktan nasıl kaçınırım?

Histerezisi (ayrı tetikleme ve temizleme seviyeleri) cihaz başına bir bekleme süresiyle birleştirin. Betik bir sınır aşıldığında tek bir uyarı gönderir, koşul sürdüğü sürece sessiz kalır ve koşul ortadan kalktığında tek bir çözüm bildirimi gönderir.

Alarm topic'leri için QoS 1 kullanmalı mıyım?

Evet. QoS 1, abone bağlı olduğu sürece her alarmın ona en az bir kez teslim edilmesini sağlar; bekleme süresi de ara sıra oluşan yinelenen mesajları absorbe eder.

Ücretsiz ConnectMQ planı bunun için yeterli mi?

Evet. Ücretsiz Serverless planı 25 bağlantıya kadar ve QoS 1'e kadar destek sunar; bu da bir uyarı servisini ve küçük bir cihaz grubunu karşılar.