breakpilot-pwa/voice-service/bqas/notifier.py

#!/usr/bin/env python3
"""
BQAS Notifier - Benachrichtigungsmodul fuer BQAS Test-Ergebnisse

Unterstuetzt verschiedene Benachrichtigungsmethoden:
- macOS Desktop-Benachrichtigungen
- Log-Datei
- Slack Webhook (optional)
- E-Mail (optional)
"""

import argparse
import json
import os
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
from dataclasses import dataclass, asdict


@dataclass
class NotificationConfig:
    """Konfiguration fuer Benachrichtigungen."""

    # Allgemein
    enabled: bool = True
    log_file: str = "/var/log/bqas/notifications.log"

    # macOS Desktop
    desktop_enabled: bool = True
    desktop_sound_success: str = "Glass"
    desktop_sound_failure: str = "Basso"

    # Slack (optional)
    slack_enabled: bool = False
    slack_webhook_url: Optional[str] = None
    slack_channel: str = "#bqas-alerts"

    # E-Mail (optional)
    email_enabled: bool = False
    email_recipient: Optional[str] = None
    email_sender: str = "bqas@localhost"

    @classmethod
    def from_env(cls) -> "NotificationConfig":
        """Erstellt Config aus Umgebungsvariablen."""
        return cls(
            enabled=os.getenv("BQAS_NOTIFY_ENABLED", "true").lower() == "true",
            log_file=os.getenv("BQAS_LOG_FILE", "/var/log/bqas/notifications.log"),
            desktop_enabled=os.getenv("BQAS_NOTIFY_DESKTOP", "true").lower() == "true",
            slack_enabled=os.getenv("BQAS_NOTIFY_SLACK", "false").lower() == "true",
            slack_webhook_url=os.getenv("BQAS_SLACK_WEBHOOK"),
            slack_channel=os.getenv("BQAS_SLACK_CHANNEL", "#bqas-alerts"),
            email_enabled=os.getenv("BQAS_NOTIFY_EMAIL", "false").lower() == "true",
            email_recipient=os.getenv("BQAS_EMAIL_RECIPIENT"),
        )


@dataclass
class Notification:
    """Eine Benachrichtigung."""

    status: str  # "success", "failure", "warning"
    message: str
    details: Optional[str] = None
    timestamp: str = ""
    source: str = "bqas"

    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.now().isoformat()


class BQASNotifier:
    """Haupt-Notifier-Klasse fuer BQAS."""

    def __init__(self, config: Optional[NotificationConfig] = None):
        self.config = config or NotificationConfig.from_env()

    def notify(self, notification: Notification) -> bool:
        """Sendet eine Benachrichtigung ueber alle aktivierten Kanaele."""
        if not self.config.enabled:
            return False

        success = True

        # Log-Datei (immer)
        self._log_notification(notification)

        # Desktop (macOS)
        if self.config.desktop_enabled:
            if not self._send_desktop(notification):
                success = False

        # Slack
        if self.config.slack_enabled and self.config.slack_webhook_url:
            if not self._send_slack(notification):
                success = False

        # E-Mail
        if self.config.email_enabled and self.config.email_recipient:
            if not self._send_email(notification):
                success = False

        return success

    def _log_notification(self, notification: Notification) -> None:
        """Schreibt Benachrichtigung in Log-Datei."""
        try:
            log_path = Path(self.config.log_file)
            log_path.parent.mkdir(parents=True, exist_ok=True)

            log_entry = {
                **asdict(notification),
                "logged_at": datetime.now().isoformat(),
            }

            with open(log_path, "a") as f:
                f.write(json.dumps(log_entry) + "\n")
        except Exception as e:
            print(f"Fehler beim Logging: {e}", file=sys.stderr)

    def _send_desktop(self, notification: Notification) -> bool:
        """Sendet macOS Desktop-Benachrichtigung."""
        try:
            title = self._get_title(notification.status)
            sound = (
                self.config.desktop_sound_failure
                if notification.status == "failure"
                else self.config.desktop_sound_success
            )

            script = f'display notification "{notification.message}" with title "{title}" sound name "{sound}"'

            subprocess.run(
                ["osascript", "-e", script], capture_output=True, timeout=5
            )
            return True
        except Exception as e:
            print(f"Desktop-Benachrichtigung fehlgeschlagen: {e}", file=sys.stderr)
            return False

    def _send_slack(self, notification: Notification) -> bool:
        """Sendet Slack-Benachrichtigung."""
        try:
            import urllib.request

            emoji = self._get_emoji(notification.status)
            color = self._get_color(notification.status)

            payload = {
                "channel": self.config.slack_channel,
                "attachments": [
                    {
                        "color": color,
                        "title": f"{emoji} BQAS {notification.status.upper()}",
                        "text": notification.message,
                        "fields": [
                            {
                                "title": "Details",
                                "value": notification.details or "Keine Details",
                                "short": False,
                            },
                            {
                                "title": "Zeitpunkt",
                                "value": notification.timestamp,
                                "short": True,
                            },
                        ],
                    }
                ],
            }

            req = urllib.request.Request(
                self.config.slack_webhook_url,
                data=json.dumps(payload).encode("utf-8"),
                headers={"Content-Type": "application/json"},
            )

            with urllib.request.urlopen(req, timeout=10) as response:
                return response.status == 200
        except Exception as e:
            print(f"Slack-Benachrichtigung fehlgeschlagen: {e}", file=sys.stderr)
            return False

    def _send_email(self, notification: Notification) -> bool:
        """Sendet E-Mail-Benachrichtigung (via sendmail)."""
        try:
            subject = f"[BQAS] {notification.status.upper()}: {notification.message}"
            body = f"""
BQAS Test-Ergebnis
==================

Status: {notification.status.upper()}
Nachricht: {notification.message}
Details: {notification.details or 'Keine'}
Zeitpunkt: {notification.timestamp}

---
BQAS - Breakpilot Quality Assurance System
"""

            msg = f"Subject: {subject}\nFrom: {self.config.email_sender}\nTo: {self.config.email_recipient}\n\n{body}"

            process = subprocess.Popen(
                ["/usr/sbin/sendmail", "-t"],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            process.communicate(msg.encode("utf-8"), timeout=30)

            return process.returncode == 0
        except Exception as e:
            print(f"E-Mail-Benachrichtigung fehlgeschlagen: {e}", file=sys.stderr)
            return False

    @staticmethod
    def _get_title(status: str) -> str:
        """Gibt Titel basierend auf Status zurueck."""
        titles = {
            "success": "BQAS Erfolgreich",
            "failure": "BQAS Fehlgeschlagen",
            "warning": "BQAS Warnung",
        }
        return titles.get(status, "BQAS")

    @staticmethod
    def _get_emoji(status: str) -> str:
        """Gibt Emoji basierend auf Status zurueck."""
        emojis = {
            "success": ":white_check_mark:",
            "failure": ":x:",
            "warning": ":warning:",
        }
        return emojis.get(status, ":information_source:")

    @staticmethod
    def _get_color(status: str) -> str:
        """Gibt Slack-Farbe basierend auf Status zurueck."""
        colors = {
            "success": "good",
            "failure": "danger",
            "warning": "warning",
        }
        return colors.get(status, "#808080")


def main():
    """CLI-Einstiegspunkt."""
    parser = argparse.ArgumentParser(description="BQAS Notifier")
    parser.add_argument(
        "--status",
        choices=["success", "failure", "warning"],
        required=True,
        help="Status der Benachrichtigung",
    )
    parser.add_argument(
        "--message",
        required=True,
        help="Benachrichtigungstext",
    )
    parser.add_argument(
        "--details",
        default=None,
        help="Zusaetzliche Details",
    )
    parser.add_argument(
        "--desktop-only",
        action="store_true",
        help="Nur Desktop-Benachrichtigung senden",
    )

    args = parser.parse_args()

    # Konfiguration laden
    config = NotificationConfig.from_env()

    # Bei --desktop-only andere Kanaele deaktivieren
    if args.desktop_only:
        config.slack_enabled = False
        config.email_enabled = False

    # Benachrichtigung erstellen und senden
    notifier = BQASNotifier(config)
    notification = Notification(
        status=args.status,
        message=args.message,
        details=args.details,
    )

    success = notifier.notify(notification)
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()