Notifications

Notifications and Alerts - C-Suite

Este documento descreve o sistema de notificações e alertas centralizado para o ecossistema C-Suite.

Visão Geral

O módulo common/common_notifications.py fornece um sistema centralizado de notificações suportando múltiplos canais (email, Slack, webhooks) com diferentes níveis de prioridade.

Características

Configuração

Variáveis de Ambiente

# Habilitar notificações
NOTIFICATIONS_ENABLED=true

# Email (SMTP)
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@gmail.com
SMTP_PASSWORD=your-password
SMTP_FROM=noreply@csuite.local
SMTP_USE_TLS=true

# Slack
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL
SLACK_CHANNEL=#alerts
SLACK_DEFAULT_CHANNEL=#alerts

# Webhook
WEBHOOK_URL=https://your-webhook-endpoint.com/notify
WEBHOOK_SECRET=your-secret-key

# Default recipient
NOTIFICATION_DEFAULT_EMAIL=admin@csuite.local

Uso

Enviar Notificação Básica

from common.common_notifications import send_notifications import send_notification, NotificationChannel, NotificationPriority

# Email
send_notification(
    title="Backup Completed",
    message="Database backup completed successfully",
    priority=NotificationPriority.MEDIUM,
    channel=NotificationChannel.EMAIL,
    recipient="admin@csuite.local"
)

# Slack
send_notification(
    title="High CPU Usage",
    message="CPU usage is above 80%",
    priority=NotificationPriority.HIGH,
    channel=NotificationChannel.SLACK,
    recipient="#alerts"
)

Enviar Alertas

from common.common_notifications import send_alert, send_critical_alert

# Alerta de alta prioridade (Slack)
send_alert(
    title="Service Down",
    message="csuite-api service is not responding",
    recipient="#ops-team",
    metadata={"service": "csuite-api", "status_code": 503}
)

# Alerta crítico (múltiplos canais)
send_critical_alert(
    title="Database Connection Lost",
    message="Cannot connect to primary database",
    metadata={"database": "csuite", "error": "Connection timeout"}
)

Usando Notification Object

from common.common_notifications import Notification, NotificationService

notification = Notification(
    title="Policy Updated",
    message="Policy 'min_margin' was updated",
    priority=NotificationPriority.MEDIUM,
    channel=NotificationChannel.EMAIL,
    recipient="admin@csuite.local",
    metadata={
        "policy_id": "min_margin",
        "updated_by": "user123",
        "old_value": 0.10,
        "new_value": 0.15
    }
)

service = NotificationService()
service.send(notification)

Canais Suportados

Email

Envia notificações via SMTP.

send_notification(
    title="Test Email",
    message="This is a test",
    channel=NotificationChannel.EMAIL,
    recipient="user@example.com"
)

Slack

Envia notificações para canais do Slack via webhook.

send_notification(
    title="Test Slack",
    message="This is a test",
    channel=NotificationChannel.SLACK,
    recipient="#alerts"
)

Webhook

Envia notificações para um endpoint HTTP customizado.

send_notification(
    title="Test Webhook",
    message="This is a test",
    channel=NotificationChannel.WEBHOOK,
    metadata={"custom_field": "value"}
)

Console

Para desenvolvimento/testing, imprime no console.

send_notification(
    title="Test Console",
    message="This is a test",
    channel=NotificationChannel.CONSOLE
)

Níveis de Prioridade

Integração com Outros Módulos

Com Audit Log

from common.common_audit import audit_log
from common.common_notifications import send_alert, NotificationPriority

def update_policy(policy_id: str, new_value: float, user_id: str):
    # Atualizar política
    # ...

    # Log de auditoria
    audit_log(
        action="policy_updated",
        user_id=user_id,
        resource_type="policy",
        resource_id=policy_id
    )

    # Notificar administradores
    send_alert(
        title="Policy Updated",
        message=f"Policy {policy_id} was updated to {new_value}",
        metadata={"policy_id": policy_id, "new_value": new_value}
    )

Com Health Checks

from common.common_health import HealthChecker
from common.common_notifications import send_critical_alert

def check_service_health():
    checker = HealthChecker()
    status = checker.check_dependency("database:csuite")

    if status.status == "unhealthy":
        send_critical_alert(
            title="Database Unhealthy",
            message=f"Database health check failed: {status.message}",
            metadata={"dependency": "database:csuite", "error": status.message}
        )

Exemplos de Uso

Backup Failed

from common.common_notifications import send_critical_alert

try:
    backup_database()
except Exception as e:
    send_critical_alert(
        title="Database Backup Failed",
        message=f"Backup failed: {str(e)}",
        metadata={"error": str(e), "timestamp": datetime.utcnow().isoformat()}
    )

Rate Limit Exceeded

from common.common_notifications import send_alert

if rate_limit_exceeded:
    send_alert(
        title="Rate Limit Exceeded",
        message=f"User {user_id} exceeded rate limit",
        metadata={"user_id": user_id, "limit": 100, "requests": 150}
    )

Policy Violation

from common.common_notifications import send_notification, NotificationPriority

if policy_violated:
    send_notification(
        title="Policy Violation",
        message=f"Decision violated policy: {policy_name}",
        priority=NotificationPriority.HIGH,
        channel=NotificationChannel.SLACK,
        metadata={"policy": policy_name, "decision_id": decision_id}
    )

Troubleshooting

Email não está sendo enviado

  1. Verifique configurações SMTP
  2. Verifique credenciais
  3. Verifique firewall/portas
  4. Teste com SMTP_USE_TLS=false primeiro

Slack não está funcionando

  1. Verifique webhook URL
  2. Verifique formato do payload
  3. Verifique permissões do webhook

Notificações desabilitadas

Se NOTIFICATIONS_ENABLED=false, todas as notificações serão ignoradas (apenas logadas).

Dependências

Opcionais (instaladas automaticamente se disponíveis):

Referências

🔊 Text-to-Speech

1.0x
1.0
Pronto para reproduzir