#!/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()