Implementiere Multi-Endpoint Support und konfigurierbare Benachrichtigungsregeln (Version 3.0)
This commit is contained in:
107
README.md
107
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
|
||||
|
44
env.example
44
env.example
@ -7,4 +7,46 @@
|
||||
# APPRISE_URL=telegram://bottoken/ChatID
|
||||
# APPRISE_URL=discord://webhook_id
|
||||
|
||||
APPRISE_URL=https://mattermost.medisoftware.org/hooks/i67zgcgajifxxxtfwjxcxace7a
|
||||
# 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
|
183
test_apprise.py
183
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")
|
||||
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")
|
@ -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}")
|
||||
|
||||
|
Reference in New Issue
Block a user