# =================================================================== # services/notification/mattermost.py # Mattermost 알림 서비스 모듈 # =================================================================== # Mattermost로 알림 메시지를 발송하는 전용 서비스입니다. # 웹훅 및 API 방식을 모두 지원합니다. # =================================================================== """ Mattermost 알림 서비스 모듈 Mattermost 채널로 알림 메시지를 발송합니다. 웹훅과 Bot API 두 가지 방식을 지원합니다. 사용 예시: from services.notification.mattermost import MattermostNotifier notifier = MattermostNotifier.from_config() notifier.send_message("서버 점검 알림입니다.") """ from typing import Dict, List, Optional, Any import requests from core.logging_utils import get_logger from core.config import get_config logger = get_logger(__name__) class MattermostNotifier: """ Mattermost 알림 발송 클래스 Mattermost 채널로 메시지를 발송합니다. 웹훅 방식과 Bot API 방식을 모두 지원합니다. Attributes: server_url: Mattermost 서버 URL bot_token: Bot 인증 토큰 channel_id: 기본 채널 ID webhook_url: 웹훅 URL """ def __init__( self, server_url: str = "", bot_token: str = "", channel_id: str = "", webhook_url: str = "" ): """ Args: server_url: Mattermost 서버 URL (예: https://mattermost.example.com) bot_token: Bot 인증 토큰 channel_id: 기본 채널 ID webhook_url: Incoming 웹훅 URL """ self.server_url = server_url.rstrip('/') if server_url else "" self.bot_token = bot_token self.channel_id = channel_id self.webhook_url = webhook_url @classmethod def from_config(cls) -> 'MattermostNotifier': """ 설정에서 인스턴스 생성 Returns: 설정이 적용된 MattermostNotifier 인스턴스 """ config = get_config() mm_config = config.mattermost return cls( server_url=mm_config.get('url', ''), bot_token=mm_config.get('token', ''), channel_id=mm_config.get('channel_id', ''), webhook_url=mm_config.get('webhook_url', ''), ) def _validate_api_config(self) -> bool: """API 방식 설정 검증""" if not self.server_url: logger.error("Mattermost 서버 URL이 설정되지 않았습니다.") return False if not self.server_url.startswith(('http://', 'https://')): logger.error(f"유효하지 않은 서버 URL: {self.server_url}") return False if not self.bot_token: logger.error("Bot 토큰이 설정되지 않았습니다.") return False if not self.channel_id: logger.error("채널 ID가 설정되지 않았습니다.") return False return True def send_message( self, message: str, channel_id: Optional[str] = None, use_webhook: bool = False, attachments: Optional[List[Dict]] = None, props: Optional[Dict] = None ) -> bool: """ 메시지 발송 Args: message: 발송할 메시지 channel_id: 채널 ID (None이면 기본 채널) use_webhook: 웹훅 사용 여부 attachments: 첨부 데이터 (Mattermost 형식) props: 추가 속성 Returns: 발송 성공 여부 """ if use_webhook: return self._send_via_webhook(message, attachments, props) else: return self._send_via_api(message, channel_id, attachments, props) def _send_via_webhook( self, message: str, attachments: Optional[List[Dict]] = None, props: Optional[Dict] = None ) -> bool: """ 웹훅으로 메시지 발송 Args: message: 발송할 메시지 attachments: 첨부 데이터 props: 추가 속성 Returns: 발송 성공 여부 """ if not self.webhook_url: logger.error("웹훅 URL이 설정되지 않았습니다.") return False payload = {"text": message} if attachments: payload["attachments"] = attachments if props: payload["props"] = props try: response = requests.post( self.webhook_url, json=payload, headers={"Content-Type": "application/json"}, timeout=10 ) if response.status_code == 200: logger.info("Mattermost 웹훅 전송 성공") return True else: logger.error(f"웹훅 전송 실패: {response.status_code} - {response.text}") return False except requests.exceptions.Timeout: logger.error("웹훅 전송 타임아웃") return False except requests.exceptions.RequestException as e: logger.error(f"웹훅 전송 예외: {e}") return False def _send_via_api( self, message: str, channel_id: Optional[str] = None, attachments: Optional[List[Dict]] = None, props: Optional[Dict] = None ) -> bool: """ Bot API로 메시지 발송 Args: message: 발송할 메시지 channel_id: 채널 ID attachments: 첨부 데이터 props: 추가 속성 Returns: 발송 성공 여부 """ if not self._validate_api_config(): return False target_channel = channel_id or self.channel_id url = f"{self.server_url}/api/v4/posts" headers = { "Authorization": f"Bearer {self.bot_token}", "Content-Type": "application/json" } payload = { "channel_id": target_channel, "message": message } if attachments: payload["props"] = payload.get("props", {}) payload["props"]["attachments"] = attachments if props: payload["props"] = {**payload.get("props", {}), **props} try: response = requests.post(url, json=payload, headers=headers, timeout=10) if response.status_code in [200, 201]: logger.info("Mattermost API 전송 성공") return True else: logger.error(f"API 전송 실패: {response.status_code} - {response.text}") return False except requests.exceptions.Timeout: logger.error("API 전송 타임아웃") return False except requests.exceptions.RequestException as e: logger.error(f"API 전송 예외: {e}") return False def send_formatted_message( self, title: str, text: str, color: str = "#3498db", fields: Optional[List[Dict]] = None, channel_id: Optional[str] = None, use_webhook: bool = False ) -> bool: """ 서식화된 메시지 발송 (Attachment 사용) Args: title: 메시지 제목 text: 메시지 본문 color: 테두리 색상 (hex) fields: 필드 목록 [{"title": "", "value": "", "short": bool}] channel_id: 채널 ID use_webhook: 웹훅 사용 여부 Returns: 발송 성공 여부 """ attachment = { "fallback": f"{title}: {text}", "color": color, "title": title, "text": text, } if fields: attachment["fields"] = fields return self.send_message( message="", channel_id=channel_id, use_webhook=use_webhook, attachments=[attachment] ) def send_alert( self, title: str, message: str, level: str = "info", channel_id: Optional[str] = None ) -> bool: """ 알림 메시지 발송 (레벨에 따른 색상) Args: title: 알림 제목 message: 알림 내용 level: 알림 레벨 (info, warning, error, success) channel_id: 채널 ID Returns: 발송 성공 여부 """ colors = { "info": "#3498db", "warning": "#f39c12", "error": "#e74c3c", "success": "#27ae60" } icons = { "info": "ℹ️", "warning": "⚠️", "error": "🚨", "success": "✅" } color = colors.get(level, colors["info"]) icon = icons.get(level, icons["info"]) return self.send_formatted_message( title=f"{icon} {title}", text=message, color=color, channel_id=channel_id ) def send_mattermost_notification( message: str, channel_id: Optional[str] = None, use_webhook: bool = False ) -> bool: """ Mattermost 알림 간편 함수 설정에서 자동으로 설정을 로드하여 메시지를 발송합니다. Args: message: 발송할 메시지 channel_id: 채널 ID (None이면 기본 채널) use_webhook: 웹훅 사용 여부 Returns: 발송 성공 여부 """ notifier = MattermostNotifier.from_config() return notifier.send_message(message, channel_id, use_webhook)