TI-Status-Bot/test_apprise.py

#!/usr/bin/env python3
"""
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 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:
            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"🔗 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:
        result = apobj.notify(
            title="🧪 TI-Status Multi-Endpoint Test",
            body=test_message,
            body_format=apprise.NotifyFormat.MARKDOWN
        )
        
        if result:
            print("✅ Test erfolgreich! Nachricht wurde an alle Endpunkte gesendet.")
            return True
        else:
            print("❌ Test fehlgeschlagen. Keine Nachricht gesendet.")
            return False
            
    except Exception as e:
        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 direkten URLs"""
    
    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")
    
    test_urls = input("Gib deine Apprise URLs ein (kommagetrennt, oder Enter für Skip): ").strip()
    
    if not test_urls:
        print("⏭️ Test übersprungen")
        return
    
    urls = [url.strip() for url in test_urls.split(',')]
    
    try:
        apobj = apprise.Apprise()
        for url in urls:
            apobj.add(url)
        
        result = apobj.notify(
            title="🧪 Direkter Test",
            body="Direkter Apprise-Test erfolgreich!",
            body_format=apprise.NotifyFormat.MARKDOWN
        )
        
        if result:
            print("✅ Direkter Test erfolgreich!")
        else:
            print("❌ Direkter Test fehlgeschlagen")
            
    except Exception as e:
        print(f"❌ Fehler beim direkten Test: {e}")

if __name__ == "__main__":
    print("🧪 Erweiterter Apprise-Verbindungstest")
    print("=" * 50)
    
    # Zeige Konfiguration
    test_configuration()
    
    # 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 URLs korrekt sind")
    print("3. Teste das Hauptskript: python ti_status_checker.py")
    print("4. Aktiviere DEBUG_MODE=true für detaillierte Ausgaben")