157 lines
5.5 KiB
Python
157 lines
5.5 KiB
Python
"""Unified notification manager for setting up recipient-based notification routing.
|
|
|
|
This module provides helpers to initialize the unified notification system
|
|
based on the recipients configuration.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from .email_service import EmailService
|
|
from .logging_config import get_logger
|
|
from .notification_adapters import EmailNotificationAdapter, PushoverNotificationAdapter
|
|
from .notification_service import NotificationService
|
|
from .pushover_service import PushoverService
|
|
|
|
_LOGGER = get_logger(__name__)
|
|
|
|
|
|
def setup_notification_service(
|
|
config: dict[str, Any],
|
|
email_service: EmailService | None = None,
|
|
pushover_service: PushoverService | None = None,
|
|
) -> NotificationService | None:
|
|
"""Set up unified notification service from config.
|
|
|
|
Args:
|
|
config: Full configuration dictionary
|
|
email_service: Optional EmailService instance
|
|
pushover_service: Optional PushoverService instance
|
|
|
|
Returns:
|
|
NotificationService instance, or None if no recipients configured
|
|
|
|
"""
|
|
notifications_config = config.get("notifications", {})
|
|
recipients = notifications_config.get("recipients", [])
|
|
|
|
if not recipients:
|
|
_LOGGER.info("No notification recipients configured")
|
|
return None
|
|
|
|
notification_service = NotificationService()
|
|
|
|
# Process each recipient and their methods
|
|
for recipient in recipients:
|
|
recipient_name = recipient.get("name", "unknown")
|
|
methods = recipient.get("methods", [])
|
|
|
|
for method in methods:
|
|
method_type = method.get("type")
|
|
|
|
if method_type == "email":
|
|
if not email_service:
|
|
_LOGGER.warning(
|
|
"Email method configured for %s but email service not available",
|
|
recipient_name,
|
|
)
|
|
continue
|
|
|
|
email_address = method.get("address")
|
|
if not email_address:
|
|
_LOGGER.warning(
|
|
"Email method for %s missing address", recipient_name
|
|
)
|
|
continue
|
|
|
|
# Create a unique backend name for this recipient's email
|
|
backend_name = f"email_{recipient_name}"
|
|
|
|
# Check if we already have an email backend
|
|
if not notification_service.has_backend("email"):
|
|
# Create email adapter with all email recipients
|
|
email_recipients = []
|
|
for r in recipients:
|
|
for m in r.get("methods", []):
|
|
if m.get("type") == "email" and m.get("address"):
|
|
email_recipients.append(m.get("address"))
|
|
|
|
if email_recipients:
|
|
email_adapter = EmailNotificationAdapter(
|
|
email_service, email_recipients
|
|
)
|
|
notification_service.register_backend("email", email_adapter)
|
|
_LOGGER.info(
|
|
"Registered email backend with %d recipient(s)",
|
|
len(email_recipients),
|
|
)
|
|
|
|
elif method_type == "pushover":
|
|
if not pushover_service:
|
|
_LOGGER.warning(
|
|
"Pushover method configured for %s but pushover service not available",
|
|
recipient_name,
|
|
)
|
|
continue
|
|
|
|
priority = method.get("priority", 0)
|
|
|
|
# Check if we already have a pushover backend
|
|
if not notification_service.has_backend("pushover"):
|
|
# Pushover sends to user_key configured in pushover service
|
|
pushover_adapter = PushoverNotificationAdapter(
|
|
pushover_service, priority
|
|
)
|
|
notification_service.register_backend("pushover", pushover_adapter)
|
|
_LOGGER.info("Registered pushover backend with priority %d", priority)
|
|
|
|
if not notification_service.backends:
|
|
_LOGGER.warning("No notification backends could be configured")
|
|
return None
|
|
|
|
_LOGGER.info(
|
|
"Notification service configured with backends: %s",
|
|
list(notification_service.backends.keys()),
|
|
)
|
|
return notification_service
|
|
|
|
|
|
def get_enabled_backends(
|
|
notification_type: str, config: dict[str, Any]
|
|
) -> list[str] | None:
|
|
"""Get list of enabled backends for a notification type.
|
|
|
|
Args:
|
|
notification_type: "daily_report" or "error_alerts"
|
|
config: Full configuration dictionary
|
|
|
|
Returns:
|
|
List of backend names to use, or None for all backends
|
|
|
|
"""
|
|
notifications_config = config.get("notifications", {})
|
|
notification_config = notifications_config.get(notification_type, {})
|
|
|
|
if not notification_config.get("enabled", False):
|
|
return []
|
|
|
|
# Return None to indicate all backends should be used
|
|
# The NotificationService will send to all registered backends
|
|
return None
|
|
|
|
|
|
def get_notification_config(
|
|
notification_type: str, config: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""Get configuration for a specific notification type.
|
|
|
|
Args:
|
|
notification_type: "daily_report" or "error_alerts"
|
|
config: Full configuration dictionary
|
|
|
|
Returns:
|
|
Configuration dictionary for the notification type
|
|
|
|
"""
|
|
notifications_config = config.get("notifications", {})
|
|
return notifications_config.get(notification_type, {})
|