372 lines
11 KiB
Python
372 lines
11 KiB
Python
# ===================================================================
|
||
# 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)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
"""
|
||
Mattermost 알림 서비스 모듈 테스트
|
||
|
||
사용법:
|
||
python services/notification/mattermost.py
|
||
"""
|
||
logger.info("=== Mattermost 알림 서비스 모듈 테스트 ===")
|
||
|
||
try:
|
||
config = get_config()
|
||
|
||
logger.info(f"설정 로드 완료")
|
||
|
||
# 알림기 초기화
|
||
notifier = MattermostNotifier.from_config()
|
||
logger.info("\nMattermostNotifier 초기화 완료")
|
||
|
||
logger.info("\n제공 기능:")
|
||
logger.info("- send_message: 기본 메시지 발송")
|
||
logger.info("- send_alert: 알림 메시지 발송")
|
||
logger.info("- send_weather: 날씨 정보 발송")
|
||
logger.info("- send_daily_report: 일일 보고서 발송")
|
||
|
||
logger.info("\n지원 방식:")
|
||
logger.info("- 웹훅 방식 (Incoming Webhook)")
|
||
logger.info("- Bot API 방식")
|
||
|
||
logger.info("\n✓ Mattermost 알림 서비스 모듈 테스트 완료")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Mattermost 모듈 테스트 실패: {e}")
|
||
import traceback
|
||
logger.error(traceback.format_exc())
|