dify-1.8.1/api/services/notification_service.py

"""Notification service for sending alerts through various channels."""

import logging
from typing import Any

import httpx

from extensions.ext_mail import mail
from libs.email_i18n import get_email_i18n_service

logger = logging.getLogger(__name__)


class NotificationService:
    """Unified notification service for alerts."""

    @staticmethod
    def send_email(to_emails: list[str], subject: str, content: str) -> dict[str, Any]:
        """Send email notification.

        Args:
            to_emails: List of recipient email addresses
            subject: Email subject
            content: HTML email content

        Returns:
            dict with status: 'success' or 'failed', and optional 'error' message
        """
        if not mail.is_inited():
            return {"status": "failed", "error": "Email service not initialized"}

        try:
            email_service = get_email_i18n_service()
            for email in to_emails:
                email_service.send_raw_email(to=[email], subject=subject, html_content=content)

            logger.info(f"Alert email sent successfully to {to_emails}")
            return {"status": "success"}
        except Exception as e:
            logger.exception(f"Failed to send alert email to {to_emails}")
            return {"status": "failed", "error": str(e)}

    @staticmethod
    def send_dingtalk(webhook_urls: list[str], message: str) -> dict[str, Any]:
        """Send DingTalk notification.

        Args:
            webhook_urls: List of DingTalk webhook URLs
            message: Message text

        Returns:
            dict with status: 'success' or 'failed', and optional 'error' message
        """
        try:
            for webhook_url in webhook_urls:
                response = httpx.post(
                    webhook_url,
                    json={"msgtype": "text", "text": {"content": message}},
                    timeout=5.0,
                )
                if response.status_code != 200:
                    raise Exception(f"DingTalk notification failed: {response.text}")

            logger.info(f"DingTalk notification sent successfully to {len(webhook_urls)} webhooks")
            return {"status": "success"}
        except Exception as e:
            logger.exception("Failed to send DingTalk notification")
            return {"status": "failed", "error": str(e)}

    @staticmethod
    def send_wecom(webhook_urls: list[str], message: str) -> dict[str, Any]:
        """Send WeCom (Enterprise WeChat) notification.

        Args:
            webhook_urls: List of WeCom webhook URLs
            message: Message text

        Returns:
            dict with status: 'success' or 'failed', and optional 'error' message
        """
        try:
            for webhook_url in webhook_urls:
                response = httpx.post(
                    webhook_url,
                    json={"msgtype": "text", "text": {"content": message}},
                    timeout=5.0,
                )
                if response.status_code != 200:
                    raise Exception(f"WeCom notification failed: {response.text}")

            logger.info(f"WeCom notification sent successfully to {len(webhook_urls)} webhooks")
            return {"status": "success"}
        except Exception as e:
            logger.exception("Failed to send WeCom notification")
            return {"status": "failed", "error": str(e)}

    @classmethod
    def send_alert_notification(
        cls,
        channels: dict[str, list[str]],
        app_name: str,
        metric_name: str,
        metric_value: float,
        threshold_value: float,
        operator: str,
    ) -> dict[str, dict[str, Any]]:
        """Send alert notification through multiple channels.

        Args:
            channels: Dictionary of notification channels {'email': [...], 'dingtalk': [...], 'wecom': [...]}
            app_name: Name of the application
            metric_name: Name of the metric that triggered the alert
            metric_value: Current value of the metric
            threshold_value: Threshold that was exceeded
            operator: Comparison operator (>, <, etc.)

        Returns:
            Dictionary with status for each channel
        """
        results = {}

        # Format metric value based on type
        if "response_time" in metric_name:
            metric_display = f"{metric_value:.2f} ms"
            threshold_display = f"{threshold_value:.2f} ms"
        elif "error_rate" in metric_name:
            metric_display = f"{metric_value:.2f}%"
            threshold_display = f"{threshold_value:.2f}%"
        else:
            metric_display = f"{metric_value:.2f}"
            threshold_display = f"{threshold_value:.2f}"

        # Build message
        text_message = (
            f"🚨 告警通知\n\n"
            f"应用名称: {app_name}\n"
            f"告警指标: {metric_name}\n"
            f"当前值: {metric_display}\n"
            f"阈值: {operator} {threshold_display}\n"
            f"触发时间: {cls._get_current_time_str()}"
        )

        html_message = f"""
        <html>
        <body style="font-family: Arial, sans-serif; padding: 20px;">
            <h2 style="color: #ff4d4f;">🚨 告警通知</h2>
            <div style="background-color: #f5f5f5; padding: 15px; border-radius: 5px;">
                <p><strong>应用名称:</strong> {app_name}</p>
                <p><strong>告警指标:</strong> {metric_name}</p>
                <p><strong>当前值:</strong> <span style="color: #ff4d4f;">{metric_display}</span></p>
                <p><strong>阈值:</strong> {operator} {threshold_display}</p>
                <p><strong>触发时间:</strong> {cls._get_current_time_str()}</p>
            </div>
            <p style="margin-top: 20px; color: #666;">
                请及时检查应用运行状态并采取必要措施。
            </p>
        </body>
        </html>
        """

        # Send email notifications
        if "email" in channels and channels["email"]:
            results["email"] = cls.send_email(
                to_emails=channels["email"], subject=f"【告警】{app_name} - {metric_name}", content=html_message
            )

        # Send DingTalk notifications
        if "dingtalk" in channels and channels["dingtalk"]:
            results["dingtalk"] = cls.send_dingtalk(webhook_urls=channels["dingtalk"], message=text_message)

        # Send WeCom notifications
        if "wecom" in channels and channels["wecom"]:
            results["wecom"] = cls.send_wecom(webhook_urls=channels["wecom"], message=text_message)

        return results

    @staticmethod
    def _get_current_time_str() -> str:
        """Get current time as formatted string."""
        from datetime import datetime, timezone

        return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")