mattermost로 메시지 보내기 세팅

lib/config.py

@@ -0,0 +1,63 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+class Config:
+    # 서버 설정
+    SERVER_DOMAIN = os.getenv('SERVER_DOMAIN', 'localhost')
+    SERVER_PROTOCOL = os.getenv('SERVER_PROTOCOL', 'http')
+    WEBHOOK_SERVER_HOST = os.getenv('WEBHOOK_SERVER_HOST', '0.0.0.0')
+    WEBHOOK_SERVER_PORT = int(os.getenv('WEBHOOK_SERVER_PORT', 5000))
+
+    # 노션 API
+    NOTION_API_SECRET = os.getenv('NOTION_API_SECRET')
+
+    # 웹훅 서명 검증용 비밀키 (옵션)
+    NOTION_WEBHOOK_SECRET = os.getenv('NOTION_WEBHOOK_SECRET', None)
+
+    # Mattermost 관련
+    MATTERMOST_SERVER_URL = os.getenv('MATTERMOST_SERVER_URL', '').rstrip('/')
+    MATTERMOST_WEBHOOK_URL = os.getenv('MATTERMOST_WEBHOOK_URL')
+    MATTERMOST_BOT_TOKEN = os.getenv('MATTERMOST_BOT_TOKEN')
+    MATTERMOST_CH_ID = os.getenv('MATTERMOST_CH_ID')
+
+    # 기타 설정
+    DEBUG = os.getenv('DEBUG', 'false').lower() in ['true', '1', 'yes']
+    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
+
+    @classmethod
+    def mattermost_post_api_url(cls):
+        if cls.MATTERMOST_SERVER_URL:
+            return f"{cls.MATTERMOST_SERVER_URL}/api/v4/posts"
+        return None
+
+    @classmethod
+    def validate(cls):
+        missing = []
+
+        # NOTION_API_SECRET 필수 확인
+        if not cls.NOTION_API_SECRET:
+            missing.append('NOTION_API_SECRET')
+
+        # Mattermost는 둘 중 하나 이상 설정 필수
+        if not cls.MATTERMOST_WEBHOOK_URL and not (cls.MATTERMOST_BOT_TOKEN and cls.MATTERMOST_CH_ID):
+            if not cls.MATTERMOST_WEBHOOK_URL:
+                missing.append('MATTERMOST_WEBHOOK_URL')
+            if not cls.MATTERMOST_BOT_TOKEN:
+                missing.append('MATTERMOST_BOT_TOKEN')
+            if not cls.MATTERMOST_CH_ID:
+                missing.append('MATTERMOST_CH_ID')
+
+        if missing:
+            err_msg = (
+                f"환경변수 설정 오류: 아래 필수 항목이 누락되었습니다:\n"
+                + "\n".join(f" - {var}" for var in missing)
+                + "\n설정 후 다시 실행해주세요."
+            )
+            raise EnvironmentError(err_msg)
+        else:
+            print("필수 항목 검증 완료")
+            
+# 초기 검증 실행
+Config.validate()

lib/send_message.py

@@ -0,0 +1,83 @@
+import os, sys
+
+# 현재 파일 기준으로 프로젝트 루트 경로를 sys.path 에 추가
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+import requests
+import logging
+from lib.config import Config
+
+def send_message_to_mattermost(message: str) -> bool:
+    """
+    설정된 방식에 따라 Mattermost로 메시지를 전송합니다.
+    - 웹훅 URL이 있으면 웹훅으로 전송
+    - 봇 토큰과 채널 ID가 있으면 API로 전송
+    - 둘 다 있으면 두 방식 모두 시도
+    """
+    success = True
+
+    if Config.MATTERMOST_WEBHOOK_URL:
+        result = _send_via_webhook(message)
+        if not result:
+            logging.error("웹훅 전송 실패")
+            success = False
+
+    if Config.MATTERMOST_BOT_TOKEN and Config.MATTERMOST_CH_ID:
+        result = _send_via_bot_api(message)
+        if not result:
+            logging.error("봇 API 전송 실패")
+            success = False
+
+    if not Config.MATTERMOST_WEBHOOK_URL and not (Config.MATTERMOST_BOT_TOKEN and Config.MATTERMOST_CH_ID):
+        logging.error("Mattermost 전송 실패: 환경변수 설정이 없음")
+        success = False
+
+    return success
+
+
+
+def _send_via_webhook(message: str) -> bool:
+    try:
+        resp = requests.post(
+            Config.MATTERMOST_WEBHOOK_URL,
+            json={"text": message},
+            timeout=5
+        )
+        if resp.status_code == 200:
+            return True
+        logging.error(f"[WEBHOOK ERROR] {resp.status_code}: {resp.text}")
+        return False
+    except Exception as e:
+        logging.exception("[WEBHOOK EXCEPTION]")
+        return False
+
+
+def _send_via_bot_api(message: str) -> bool:
+    url = Config.mattermost_post_api_url()
+    if not url:
+        logging.error("MATTERMOST_SERVER_URL 설정이 필요합니다.")
+        return False
+
+    headers = {
+        "Authorization": f"Bearer {Config.MATTERMOST_BOT_TOKEN}",
+        "Content-Type": "application/json",
+    }
+    payload = {
+        "channel_id": Config.MATTERMOST_CH_ID,
+        "message": message,
+    }
+    try:
+        resp = requests.post(url, headers=headers, json=payload, timeout=5)
+        if resp.status_code == 201:
+            return True
+        logging.error(f"[BOT API ERROR] {resp.status_code}: {resp.text}")
+        return False
+    except Exception as e:
+        logging.exception("[BOT API EXCEPTION]")
+        return False
+
+
+if __name__ == "__main__":
+    test_msg = "✅ 테스트 메시지입니다: 시스템 점검 완료"
+    success = send_message_to_mattermost(test_msg)
+    print("[RESULT]", "전송 성공" if success else "전송 실패")