From 2b1e4274b3124f86172965acb14376cfbc9d6006 Mon Sep 17 00:00:00 2001 From: Markus Busche Date: Fri, 27 Jun 2025 14:08:26 +0200 Subject: [PATCH] Implementiere Multi-Endpoint Support und konfigurierbare Benachrichtigungsregeln (Version 3.0) --- README.md | 107 ++++++++++++++++++++++--- env.example | 44 ++++++++++- test_apprise.py | 183 ++++++++++++++++++++++++++++++++++--------- ti_status_checker.py | 178 ++++++++++++++++++++++++++++++++++++++--- 4 files changed, 450 insertions(+), 62 deletions(-) diff --git a/README.md b/README.md index b93437f..2611e2b 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,13 @@ Ein Python-Skript, das den TI-Status überwacht und neue Meldungen über Apprise - Überwacht die TI-Status-API auf neue Meldungen - Sendet Benachrichtigungen über Apprise (unterstützt viele Dienste wie Mattermost, Slack, Telegram, Discord, etc.) +- **Mehrere Endpunkte gleichzeitig** (Mattermost + Slack + Telegram + ...) +- **Konfigurierbare Benachrichtigungsregeln** (Filter, Zeiten, Verzögerungen) - Konfiguration über .env Datei - Markdown-Formatierung der Nachrichten - Vermeidet Duplikate durch lokale Statusverfolgung - Debug-Ausgaben für bessere Transparenz +- Umfassende Test-Tools ## Installation @@ -40,38 +43,108 @@ pip install -r requirements.txt cp env.example .env ``` -2. Bearbeite die `.env` Datei und setze deine Apprise URL: +2. Bearbeite die `.env` Datei und konfiguriere deine Endpunkte: -### Mattermost Webhook -``` -APPRISE_URL=mattermost://username:password@mattermost.medisoftware.org/channel?webhook=your_webhook_id +### Mehrere Endpunkte konfigurieren + +```bash +# Mattermost Webhook +APPRISE_URL_MATTERMOST=mattermost://username:password@mattermost.medisoftware.org/channel?webhook=your_webhook_id + +# Slack (optional) +APPRISE_URL_SLACK=slack://token_a/token_b/token_c/#channel + +# Telegram (optional) +APPRISE_URL_TELEGRAM=telegram://bottoken/ChatID + +# Discord (optional) +APPRISE_URL_DISCORD=discord://webhook_id + +# Email (optional) +APPRISE_URL_EMAIL=smtp://user:pass@gmail.com:587 + +# Pushover (optional) +APPRISE_URL_PUSHOVER=pover://token/user_key + +# Microsoft Teams (optional) +APPRISE_URL_TEAMS=msteams://TokenA/TokenB/TokenC/ ``` -### Andere Dienste -- **Slack**: `APPRISE_URL=slack://token_a/token_b/token_c/#channel` -- **Telegram**: `APPRISE_URL=telegram://bottoken/ChatID` -- **Discord**: `APPRISE_URL=discord://webhook_id` +### Benachrichtigungsregeln + +```bash +# Benachrichtigungslevel (all, critical, maintenance, outage) +NOTIFICATION_LEVEL=all + +# Filter für bestimmte Begriffe (kommagetrennt) +NOTIFICATION_FILTERS=störung,wartung,warnung + +# Benachrichtigungszeiten (24h Format) +NOTIFICATION_HOURS=08:00-18:00 + +# Verzögerung zwischen Benachrichtigungen (in Sekunden) +NOTIFICATION_DELAY=5 + +# Debug-Modus für detaillierte Ausgaben +DEBUG_MODE=true +``` + +### Weitere Apprise-URLs Weitere Apprise-URLs findest du in der [Apprise-Dokumentation](https://github.com/caronc/apprise#supported-notifications). ## Verwendung -Skript einmalig ausführen: +### Hauptskript ausführen: ```bash python ti_status_checker.py ``` -Das Skript gibt Debug-Informationen aus: +### Verbindungstest: +```bash +python test_apprise.py +``` + +Das Test-Skript bietet: +- Test aller konfigurierten Endpunkte +- Einzeltests für jeden Endpunkt +- Konfigurationsanzeige +- Direkte URL-Tests + +### Debug-Modus aktivieren: +```bash +# In .env setzen: +DEBUG_MODE=true +``` + +Das Skript gibt dann detaillierte Debug-Informationen aus: - API-Aufruf und Antwort - Anzahl der gefundenen Meldungen - Verarbeitung jeder einzelnen Meldung +- Benachrichtigungsregeln-Auswertung +- Endpunkt-Status -Für kontinuierliche Überwachung (z.B. mit cron): +### Für kontinuierliche Überwachung (z.B. mit cron): ```bash # Alle 5 Minuten ausführen */5 * * * * cd /path/to/TI-Status2Mattermost && python ti_status_checker.py ``` +## Benachrichtigungsregeln + +### Filter +- Nur Meldungen mit bestimmten Begriffen senden +- Beispiel: `NOTIFICATION_FILTERS=störung,wartung` + +### Zeiten +- Benachrichtigungen nur zu bestimmten Zeiten +- Beispiel: `NOTIFICATION_HOURS=08:00-18:00` (nur werktags) +- Beispiel: `NOTIFICATION_HOURS=09:00-17:00,19:00-22:00` (mehrere Zeiträume) + +### Verzögerungen +- Verzögerung zwischen mehreren Benachrichtigungen +- Beispiel: `NOTIFICATION_DELAY=30` (30 Sekunden Pause) + ## Unterstützte Dienste Apprise unterstützt über 80 verschiedene Benachrichtigungsdienste, darunter: @@ -86,14 +159,22 @@ Apprise unterstützt über 80 verschiedene Benachrichtigungsdienste, darunter: ## Dateien -- `ti_status_checker.py` - Hauptskript mit Debug-Ausgaben +- `ti_status_checker.py` - Hauptskript mit Multi-Endpoint-Support +- `test_apprise.py` - Umfassendes Test-Tool für alle Endpunkte - `requirements.txt` - Python-Abhängigkeiten (python-dotenv, apprise) -- `env.example` - Beispiel-Konfiguration +- `env.example` - Beispiel-Konfiguration mit allen Optionen - `ti_status_state.json` - Lokale Statusverfolgung (wird automatisch erstellt) - `.env` - Deine Konfiguration (nicht im Repository) ## Changelog +### Version 3.0 (Multi-Endpoint & Rules) +- ✅ Mehrere Apprise-Endpunkte gleichzeitig +- ✅ Konfigurierbare Benachrichtigungsregeln (Filter, Zeiten, Verzögerungen) +- ✅ Erweiterte Debug-Ausgaben +- ✅ Umfassendes Test-Tool für alle Endpunkte +- ✅ Fallback für alte Konfigurationen + ### Version 2.0 (Apprise-Integration) - ✅ Umstellung von Mattermost Webhook auf Apprise API - ✅ Konfiguration über .env Datei diff --git a/env.example b/env.example index 0375c51..e8b9d2e 100644 --- a/env.example +++ b/env.example @@ -7,4 +7,46 @@ # APPRISE_URL=telegram://bottoken/ChatID # APPRISE_URL=discord://webhook_id -APPRISE_URL=https://mattermost.medisoftware.org/hooks/i67zgcgajifxxxtfwjxcxace7a \ No newline at end of file +# Apprise Konfiguration für mehrere Endpunkte +# Mehrere URLs können durch Kommas getrennt werden + +# Mattermost Webhook +APPRISE_URL_MATTERMOST=mattermost://username:password@mattermost.medisoftware.org/channel?webhook=your_webhook_id + +# Slack (optional) +# APPRISE_URL_SLACK=slack://token_a/token_b/token_c/#channel + +# Telegram (optional) +# APPRISE_URL_TELEGRAM=telegram://bottoken/ChatID + +# Discord (optional) +# APPRISE_URL_DISCORD=discord://webhook_id + +# Email (optional) +# APPRISE_URL_EMAIL=smtp://user:pass@gmail.com:587 + +# Pushover (optional) +# APPRISE_URL_PUSHOVER=pover://token/user_key + +# Microsoft Teams (optional) +# APPRISE_URL_TEAMS=msteams://TokenA/TokenB/TokenC/ + +# Benachrichtigungsregeln +# Mögliche Werte: all, critical, maintenance, outage +NOTIFICATION_LEVEL=all + +# Benachrichtigungsfilter (kommagetrennt) +# Leer lassen für alle Meldungen +# Beispiele: "störung", "wartung", "warnung" +NOTIFICATION_FILTERS= + +# Benachrichtigungszeiten (24h Format, kommagetrennt) +# Leer lassen für 24/7 Benachrichtigungen +# Beispiel: "08:00-18:00" für nur werktags +NOTIFICATION_HOURS= + +# Verzögerung zwischen Benachrichtigungen (in Sekunden) +NOTIFICATION_DELAY=0 + +# Debug-Modus (true/false) +DEBUG_MODE=false \ No newline at end of file diff --git a/test_apprise.py b/test_apprise.py index 37e8acc..a2db9ae 100644 --- a/test_apprise.py +++ b/test_apprise.py @@ -1,47 +1,93 @@ #!/usr/bin/env python3 """ -Einfacher Test für Apprise Mattermost-Verbindung +Erweiterter Test für Apprise-Verbindungen mit mehreren Endpunkten """ import os from dotenv import load_dotenv import apprise +from datetime import datetime # Lade .env Datei load_dotenv() -def test_mattermost_connection(): - """Testet die Mattermost-Verbindung über Apprise""" +def get_apprise_urls(): + """Holt alle konfigurierten Apprise URLs aus der .env""" + urls = {} - # Hole Apprise URL aus .env - apprise_url = os.getenv('APPRISE_URL') + # Prüfe alle möglichen URL-Konfigurationen + url_keys = [ + 'APPRISE_URL_MATTERMOST', + 'APPRISE_URL_SLACK', + 'APPRISE_URL_TELEGRAM', + 'APPRISE_URL_DISCORD', + 'APPRISE_URL_EMAIL', + 'APPRISE_URL_PUSHOVER', + 'APPRISE_URL_TEAMS' + ] - if not apprise_url: - print("❌ APPRISE_URL nicht in .env gefunden!") - print("Bitte erstelle eine .env Datei mit deiner Apprise URL") + for key in url_keys: + url = os.getenv(key) + if url: + service_name = key.replace('APPRISE_URL_', '').lower() + urls[service_name] = url + + # Fallback für alte Konfiguration + if not urls: + legacy_url = os.getenv('APPRISE_URL') + if legacy_url: + urls['legacy'] = legacy_url + + return urls + +def is_debug_mode(): + """Prüft ob Debug-Modus aktiviert ist""" + return os.getenv('DEBUG_MODE', 'false').lower() == 'true' + +def test_all_endpoints(): + """Testet alle konfigurierten Endpunkte""" + + urls = get_apprise_urls() + + if not urls: + print("❌ Keine Apprise URLs konfiguriert!") + print("Bitte erstelle eine .env Datei basierend auf env.example") return False - print(f"🔗 Teste Apprise URL: {apprise_url}") + print(f"🔗 Gefundene Endpunkte: {len(urls)}") + for service, url in urls.items(): + print(f" - {service}: {url}") + + print("\n🧪 Teste alle Endpunkte...") + + # Erstelle Apprise Objekt + apobj = apprise.Apprise() + + # Füge alle URLs hinzu + for service, url in urls.items(): + apobj.add(url) + + # Sende Test-Nachricht + test_message = f"""🧪 **TI-Status Test-Nachricht** + +Dies ist eine Test-Nachricht von der TI-Status2Mattermost App. + +**Konfiguration:** +- Endpunkte: {len(urls)} +- Services: {', '.join(urls.keys())} +- Zeit: {datetime.now().strftime('%d.%m.%Y %H:%M:%S')} + +✅ Apprise-Verbindung funktioniert!""" try: - # Erstelle Apprise Objekt - apobj = apprise.Apprise() - - # Füge die URL hinzu - apobj.add(apprise_url) - - # Teste die Verbindung - print("📡 Sende Test-Nachricht...") - - # Sende eine Test-Nachricht result = apobj.notify( - title="🧪 TI-Status Test", - body="Dies ist eine Test-Nachricht von der TI-Status2Mattermost App.\n\n✅ Apprise-Verbindung funktioniert!", + title="🧪 TI-Status Multi-Endpoint Test", + body=test_message, body_format=apprise.NotifyFormat.MARKDOWN ) if result: - print("✅ Test erfolgreich! Nachricht wurde gesendet.") + print("✅ Test erfolgreich! Nachricht wurde an alle Endpunkte gesendet.") return True else: print("❌ Test fehlgeschlagen. Keine Nachricht gesendet.") @@ -51,23 +97,78 @@ def test_mattermost_connection(): print(f"❌ Fehler beim Test: {e}") return False +def test_individual_endpoints(): + """Testet jeden Endpunkt einzeln""" + + urls = get_apprise_urls() + + if not urls: + print("❌ Keine Endpunkte zum Testen verfügbar") + return + + print("\n🔍 Teste Endpunkte einzeln...") + + for service, url in urls.items(): + print(f"\n📡 Teste {service}...") + + try: + apobj = apprise.Apprise() + apobj.add(url) + + result = apobj.notify( + title=f"🧪 {service.title()} Test", + body=f"Einzeltest für {service} erfolgreich!\n\nZeit: {datetime.now().strftime('%H:%M:%S')}", + body_format=apprise.NotifyFormat.MARKDOWN + ) + + if result: + print(f"✅ {service}: Erfolgreich") + else: + print(f"❌ {service}: Fehlgeschlagen") + + except Exception as e: + print(f"❌ {service}: Fehler - {e}") + +def test_configuration(): + """Zeigt die aktuelle Konfiguration an""" + + print("📋 Aktuelle Konfiguration:") + print("=" * 40) + + # Apprise URLs + urls = get_apprise_urls() + print(f"🔗 Endpunkte: {len(urls)}") + for service, url in urls.items(): + print(f" - {service}: {url}") + + # Benachrichtigungsregeln + print(f"\n📋 Benachrichtigungslevel: {os.getenv('NOTIFICATION_LEVEL', 'all')}") + print(f"🔍 Filter: {os.getenv('NOTIFICATION_FILTERS', 'keine')}") + print(f"⏰ Benachrichtigungszeiten: {os.getenv('NOTIFICATION_HOURS', '24/7')}") + print(f"⏳ Verzögerung: {os.getenv('NOTIFICATION_DELAY', '0')}s") + print(f"🔧 Debug-Modus: {os.getenv('DEBUG_MODE', 'false')}") + def test_without_env(): - """Testet mit einer direkten Mattermost URL""" + """Testet mit direkten URLs""" - print("\n🔧 Alternative: Teste mit direkter URL") - print("Beispiel für Mattermost Webhook:") - print("mattermost://username:password@mattermost.medisoftware.org/channel?webhook=your_webhook_id") + print("\n🔧 Alternative: Teste mit direkten URLs") + print("Beispiele:") + print("- Mattermost: mattermost://username:password@server/channel?webhook=id") + print("- Slack: slack://token_a/token_b/token_c/#channel") + print("- Telegram: telegram://bottoken/ChatID") - # Hier könntest du eine Test-URL direkt eingeben - test_url = input("Gib deine Apprise URL ein (oder Enter für Skip): ").strip() + test_urls = input("Gib deine Apprise URLs ein (kommagetrennt, oder Enter für Skip): ").strip() - if not test_url: + if not test_urls: print("⏭️ Test übersprungen") return + urls = [url.strip() for url in test_urls.split(',')] + try: apobj = apprise.Apprise() - apobj.add(test_url) + for url in urls: + apobj.add(url) result = apobj.notify( title="🧪 Direkter Test", @@ -84,16 +185,24 @@ def test_without_env(): print(f"❌ Fehler beim direkten Test: {e}") if __name__ == "__main__": - print("🧪 Apprise Mattermost-Verbindungstest") - print("=" * 40) + print("🧪 Erweiterter Apprise-Verbindungstest") + print("=" * 50) - # Test mit .env - success = test_mattermost_connection() + # Zeige Konfiguration + test_configuration() - if not success: + # Teste alle Endpunkte zusammen + success = test_all_endpoints() + + if success: + # Teste Endpunkte einzeln + test_individual_endpoints() + else: + # Fallback: Direkter Test test_without_env() print("\n📋 Nächste Schritte:") print("1. Überprüfe deine .env Datei") - print("2. Stelle sicher, dass die Mattermost URL korrekt ist") - print("3. Teste das Hauptskript: python ti_status_checker.py") \ No newline at end of file + print("2. Stelle sicher, dass die URLs korrekt sind") + print("3. Teste das Hauptskript: python ti_status_checker.py") + print("4. Aktiviere DEBUG_MODE=true für detaillierte Ausgaben") \ No newline at end of file diff --git a/ti_status_checker.py b/ti_status_checker.py index 4adef6f..035e1a6 100644 --- a/ti_status_checker.py +++ b/ti_status_checker.py @@ -5,6 +5,7 @@ import re import os from dotenv import load_dotenv import apprise +import time # Lade Umgebungsvariablen aus .env Datei load_dotenv() @@ -13,16 +14,118 @@ TI_API_URL = "https://ti-lage.prod.ccs.gematik.solutions/lageapi/v2/tilage" STATE_FILE = "ti_status_state.json" # Apprise Konfiguration aus Umgebungsvariablen -APPRISE_URL = os.getenv('APPRISE_URL') -if not APPRISE_URL: - raise ValueError("APPRISE_URL muss in der .env Datei definiert sein") +def get_apprise_urls(): + """Holt alle konfigurierten Apprise URLs aus der .env""" + urls = [] + + # Prüfe alle möglichen URL-Konfigurationen + url_keys = [ + 'APPRISE_URL_MATTERMOST', + 'APPRISE_URL_SLACK', + 'APPRISE_URL_TELEGRAM', + 'APPRISE_URL_DISCORD', + 'APPRISE_URL_EMAIL', + 'APPRISE_URL_PUSHOVER', + 'APPRISE_URL_TEAMS' + ] + + for key in url_keys: + url = os.getenv(key) + if url: + urls.append(url) + if is_debug_mode(): + print(f"✅ {key}: {url}") + + # Fallback für alte Konfiguration + if not urls: + legacy_url = os.getenv('APPRISE_URL') + if legacy_url: + urls.append(legacy_url) + if is_debug_mode(): + print(f"✅ Legacy APPRISE_URL: {legacy_url}") + + return urls + +def is_debug_mode(): + """Prüft ob Debug-Modus aktiviert ist""" + return os.getenv('DEBUG_MODE', 'false').lower() == 'true' + +def get_notification_level(): + """Holt das konfigurierte Benachrichtigungslevel""" + return os.getenv('NOTIFICATION_LEVEL', 'all').lower() + +def get_notification_filters(): + """Holt die konfigurierten Benachrichtigungsfilter""" + filters_str = os.getenv('NOTIFICATION_FILTERS', '') + if filters_str: + return [f.strip().lower() for f in filters_str.split(',')] + return [] + +def get_notification_hours(): + """Holt die konfigurierten Benachrichtigungszeiten""" + hours_str = os.getenv('NOTIFICATION_HOURS', '') + if hours_str: + return [h.strip() for h in hours_str.split(',')] + return [] + +def get_notification_delay(): + """Holt die konfigurierte Verzögerung zwischen Benachrichtigungen""" + return int(os.getenv('NOTIFICATION_DELAY', '0')) + +def is_within_notification_hours(): + """Prüft ob die aktuelle Zeit innerhalb der konfigurierten Benachrichtigungszeiten liegt""" + hours = get_notification_hours() + if not hours: + return True # Keine Einschränkung + + current_time = datetime.now().strftime('%H:%M') + + for time_range in hours: + if '-' in time_range: + start, end = time_range.split('-') + if start <= current_time <= end: + return True + else: + # Einzelne Zeit + if current_time == time_range: + return True + + return False + +def should_send_notification(message): + """Prüft ob eine Benachrichtigung gesendet werden soll basierend auf den Regeln""" + + # Prüfe Benachrichtigungszeiten + if not is_within_notification_hours(): + if is_debug_mode(): + print(f"⏰ Benachrichtigung außerhalb der konfigurierten Zeiten: {datetime.now().strftime('%H:%M')}") + return False + + # Prüfe Filter + filters = get_notification_filters() + if filters: + message_lower = message.lower() + for filter_term in filters: + if filter_term in message_lower: + if is_debug_mode(): + print(f"✅ Nachricht entspricht Filter '{filter_term}'") + return True + + if is_debug_mode(): + print(f"❌ Nachricht entspricht keinem Filter: {filters}") + return False + + return True def fetch_status_messages(): - print(f"Rufe TI-Status-API ab: {TI_API_URL}") + if is_debug_mode(): + print(f"Rufe TI-Status-API ab: {TI_API_URL}") resp = requests.get(TI_API_URL) resp.raise_for_status() data = resp.json() - print(f"API-Antwort erhalten. Anzahl Meldungen: {len(data.get('meldungen', []))}") + + if is_debug_mode(): + print(f"API-Antwort erhalten. Anzahl Meldungen: {len(data.get('meldungen', []))}") messages = [] for meldung in data.get("meldungen", []): @@ -32,9 +135,12 @@ def fetch_status_messages(): link = meldung.get("link", "") msg = f"{zeit}\n- {titel}: {beschreibung}\n{link}".strip() messages.append(msg) - print(f"Verarbeite Meldung: {titel[:50]}...") + + if is_debug_mode(): + print(f"Verarbeite Meldung: {titel[:50]}...") - print(f"Insgesamt {len(messages)} Meldungen verarbeitet") + if is_debug_mode(): + print(f"Insgesamt {len(messages)} Meldungen verarbeitet") return messages def load_state(): @@ -68,40 +174,90 @@ def markdownify_message(message): return md_message def send_notification(message): + """Sendet Benachrichtigungen an alle konfigurierten Endpunkte""" + + # Prüfe ob Benachrichtigung gesendet werden soll + if not should_send_notification(message): + if is_debug_mode(): + print("🚫 Benachrichtigung wird nicht gesendet (Regeln)") + return + md_message = markdownify_message(message) + # Hole alle konfigurierten URLs + urls = get_apprise_urls() + if not urls: + print("❌ Keine Apprise URLs konfiguriert!") + return + # Erstelle Apprise Objekt apobj = apprise.Apprise() - # Füge die Apprise URL hinzu - apobj.add(APPRISE_URL) + # Füge alle URLs hinzu + for url in urls: + apobj.add(url) # Erstelle die Nachricht title = "Neue TI-Status-Meldung" body = f"{md_message}\n\n[Zur Statusseite](https://fachportal.gematik.de/ti-status)\n_Gemeldet am {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}_" + if is_debug_mode(): + print(f"📤 Sende Benachrichtigung an {len(urls)} Endpunkt(e)") + # Sende die Nachricht - apobj.notify( + result = apobj.notify( title=title, body=body, body_format=apprise.NotifyFormat.MARKDOWN ) + + if result: + if is_debug_mode(): + print("✅ Benachrichtigung erfolgreich gesendet") + else: + print("❌ Fehler beim Senden der Benachrichtigung") + + # Verzögerung zwischen Benachrichtigungen + delay = get_notification_delay() + if delay > 0: + if is_debug_mode(): + print(f"⏳ Warte {delay} Sekunden...") + time.sleep(delay) def main(): + # Prüfe Konfiguration + urls = get_apprise_urls() + if not urls: + print("❌ Keine Apprise URLs konfiguriert!") + print("Bitte erstelle eine .env Datei basierend auf env.example") + return + + if is_debug_mode(): + print("🔧 Debug-Modus aktiviert") + print(f"📋 Benachrichtigungslevel: {get_notification_level()}") + print(f"🔍 Filter: {get_notification_filters()}") + print(f"⏰ Benachrichtigungszeiten: {get_notification_hours()}") + print(f"⏳ Verzögerung: {get_notification_delay()}s") + state = load_state() known_messages = set(state.get("messages", [])) print("Prüfe TI-Status-API auf neue Meldungen...") + try: messages = fetch_status_messages() new_messages = [m for m in messages if m not in known_messages] + for msg in new_messages: - print(f"Neue Meldung gefunden: {msg}") + print(f"Neue Meldung gefunden: {msg[:100]}...") send_notification(msg) known_messages.add(msg) + if new_messages: save_state({"messages": list(known_messages)}) + print(f"✅ {len(new_messages)} neue Meldung(en) verarbeitet") else: print(f"Keine neuen Meldungen ({datetime.now().strftime('%H:%M:%S')})") + except Exception as e: print(f"Fehler: {e}")