#!/usr/bin/env python3 """ GENESIS SLACK NOTIFICATION BRIDGE ================================= Sends Genesis status updates and alerts to Slack. Configuration: Set SLACK_WEBHOOK_URL environment variable or add webhook_url to config/secrets.env Usage: from monitoring.slack_bridge import SlackBridge bridge = SlackBridge() bridge.send_status("Genesis is operational") """ import json import os import urllib.request import urllib.error from datetime import datetime from pathlib import Path from typing import Dict, Optional, Any class SlackBridge: """Sends notifications to Slack via webhook.""" def __init__(self, webhook_url: str = None): """ Initialize Slack bridge. Args: webhook_url: Slack webhook URL. If not provided, reads from SLACK_WEBHOOK_URL env var or config/secrets.env """ self.webhook_url = webhook_url or self._load_webhook_url() self.default_channel = "#genesis-alerts" def _load_webhook_url(self) -> Optional[str]: """Load webhook URL from environment or config file.""" # Check environment variable url = os.environ.get("SLACK_WEBHOOK_URL") if url: return url # Check config file secrets_path = Path(__file__).parent.parent.parent / "config" / "secrets.env" if secrets_path.exists(): with open(secrets_path) as f: for line in f: if line.startswith("SLACK_WEBHOOK_URL="): return line.split("=", 1)[1].strip() return None def send_message( self, text: str, channel: str = None, username: str = "Genesis", icon_emoji: str = ":robot_face:", attachments: list = None ) -> bool: """ Send a message to Slack. Args: text: Message text channel: Override default channel username: Bot username icon_emoji: Bot icon attachments: Rich message attachments Returns: True if sent successfully """ if not self.webhook_url: print("[SlackBridge] No webhook URL configured") return False payload = { "text": text, "username": username, "icon_emoji": icon_emoji } if channel: payload["channel"] = channel if attachments: payload["attachments"] = attachments try: data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( self.webhook_url, data=data, headers={"Content-Type": "application/json"} ) with urllib.request.urlopen(req, timeout=10) as resp: return resp.status == 200 except Exception as e: print(f"[SlackBridge] Error sending message: {e}") return False def send_status(self, status: str, details: Dict = None) -> bool: """ Send a status update. Args: status: Status message details: Optional details dict Returns: True if sent """ timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") text = f"*Genesis Status Update* ({timestamp})\n{status}" attachments = None if details: fields = [ {"title": k, "value": str(v), "short": True} for k, v in details.items() ] attachments = [{ "color": "good", "fields": fields }] return self.send_message(text, attachments=attachments) def send_alert( self, title: str, message: str, severity: str = "warning", details: Dict = None ) -> bool: """ Send an alert notification. Args: title: Alert title message: Alert message severity: "info", "warning", "error", "critical" details: Additional details Returns: True if sent """ colors = { "info": "#36a64f", "warning": "#ffcc00", "error": "#ff6600", "critical": "#ff0000" } icons = { "info": ":information_source:", "warning": ":warning:", "error": ":x:", "critical": ":rotating_light:" } timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") attachment = { "color": colors.get(severity, "#cccccc"), "title": f"{icons.get(severity, '')} {title}", "text": message, "footer": f"Genesis Alert System | {timestamp}", } if details: attachment["fields"] = [ {"title": k, "value": str(v), "short": True} for k, v in details.items() ] return self.send_message( text=f"*{title}*", attachments=[attachment], icon_emoji=icons.get(severity, ":robot_face:") ) def send_task_completion( self, task_id: str, task_title: str, success: bool, duration: float = None ) -> bool: """ Send task completion notification. Args: task_id: Task ID task_title: Task title success: Whether task succeeded duration: Execution duration in seconds Returns: True if sent """ status_icon = ":white_check_mark:" if success else ":x:" status_text = "COMPLETED" if success else "FAILED" color = "good" if success else "danger" details = {"Task ID": task_id} if duration: details["Duration"] = f"{duration:.2f}s" attachment = { "color": color, "title": f"{status_icon} Task {status_text}: {task_title}", "fields": [ {"title": k, "value": str(v), "short": True} for k, v in details.items() ], "footer": f"Genesis Hyperdrive | {datetime.now().strftime('%H:%M:%S')}" } return self.send_message( text=f"Task {task_id} {status_text.lower()}", attachments=[attachment] ) def send_budget_alert( self, remaining: float, threshold: float = 50.0 ) -> bool: """ Send budget alert when remaining budget is low. Args: remaining: Remaining budget in dollars threshold: Alert threshold Returns: True if sent """ if remaining > threshold: return False severity = "critical" if remaining < 10 else "warning" return self.send_alert( title="Budget Alert", message=f"Genesis budget is running low: ${remaining:.2f} remaining", severity=severity, details={ "Remaining": f"${remaining:.2f}", "Threshold": f"${threshold:.2f}", "Action": "Review and replenish credits" } ) def test_slack_bridge(): """Test the Slack bridge (requires webhook URL).""" bridge = SlackBridge() if not bridge.webhook_url: print("No webhook URL configured. Set SLACK_WEBHOOK_URL env var.") return print("Testing Slack Bridge...") # Test status result = bridge.send_status( "Genesis Status Dashboard Test", {"Mode": "Hyperdrive", "Phase": 2, "Tasks": "10/10"} ) print(f"Status message: {'Sent' if result else 'Failed'}") # Test alert result = bridge.send_alert( "Test Alert", "This is a test alert from Genesis", severity="info" ) print(f"Alert message: {'Sent' if result else 'Failed'}") if __name__ == "__main__": test_slack_bridge()