diff --git a/app/routes/webhook/notion.py b/app/routes/webhook/notion.py new file mode 100644 index 0000000..c7fbeab --- /dev/null +++ b/app/routes/webhook/notion.py @@ -0,0 +1,32 @@ +from flask import Blueprint, request, jsonify +from lib.notion_api import handle_notion_event +from lib.send_message import send_message_to_mattermost +import logging + +notion_webhook_bp = Blueprint("notion_webhook", __name__) + +@notion_webhook_bp.route("/webhook/notion", methods=["POST"]) +def receive_notion_webhook(): + try: + event = request.get_json() + if not event: + logging.warning("빈 요청 수신됨") + return jsonify({"error": "Invalid JSON"}), 400 + + # 메시지 생성 + message = handle_notion_event(event) + if not message: + logging.warning("처리 가능한 메시지 없음") + return jsonify({"status": "ignored"}), 200 + + # Mattermost로 전송 + result = send_message_to_mattermost(message) + if result: + return jsonify({"status": "ok"}), 200 + else: + logging.error("Mattermost 전송 실패") + return jsonify({"error": "failed to send"}), 500 + + except Exception as e: + logging.exception("노션 웹훅 처리 중 예외 발생") + return jsonify({"error": str(e)}), 500 diff --git a/app/views.py b/app/views.py new file mode 100644 index 0000000..f6b3075 --- /dev/null +++ b/app/views.py @@ -0,0 +1,44 @@ +import os, sys +import logging +from flask import Blueprint, request, jsonify + +# 프로젝트 루트 경로 등록 +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +from lib.notion_api import handle_notion_event +from lib.send_message import send_message_to_mattermost +from lib.config import Config + +webhook_bp = Blueprint('webhook', __name__) + +@webhook_bp.route('/webhook/notion', methods=['POST']) +def notion_webhook(): + try: + # 📌 요청 전체 헤더 및 바디 출력 + logging.info("🔔 Notion 웹훅 요청 수신") + logging.info(f"Headers: {dict(request.headers)}") + logging.info(f"Body: {request.get_data(as_text=True)}") + + # JSON 파싱 + event = request.get_json() + if not event: + logging.warning("❗️빈 JSON 요청 수신") + return jsonify({"error": "Invalid JSON"}), 400 + + # 노션 이벤트 처리 + message = handle_notion_event(event) + if not message: + logging.info("📭 전송할 메시지 없음 - 무시") + return jsonify({"status": "ignored"}), 200 + + # Mattermost 메시지 전송 + success = send_message_to_mattermost(message) + if success: + return jsonify({"status": "ok"}), 200 + else: + logging.error("❌ Mattermost 메시지 전송 실패") + return jsonify({"error": "Failed to send message"}), 500 + + except Exception as e: + logging.exception("🔥 웹훅 처리 중 예외 발생") + return jsonify({"error": str(e)}), 500 diff --git a/lib/notion_api.py b/lib/notion_api.py new file mode 100644 index 0000000..c88d39e --- /dev/null +++ b/lib/notion_api.py @@ -0,0 +1,165 @@ +import os +import sys +import requests +import logging +import json +from typing import Optional, Dict, Union +from datetime import datetime + +# 현재 파일 기준 프로젝트 루트 경로를 sys.path에 추가 +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +from lib.config import Config + +NOTION_API_BASE = "https://api.notion.com/v1" +NOTION_VERSION = "2022-06-28" + +HEADERS = { + "Authorization": f"Bearer {Config.NOTION_API_SECRET}", + "Notion-Version": NOTION_VERSION, + "Content-Type": "application/json" +} + + +def get_page_details(page_id: str) -> Optional[Dict]: + url = f"{NOTION_API_BASE}/pages/{page_id}" + response = requests.get(url, headers=HEADERS) + + if response.status_code == 200: + return response.json() + else: + print(f"[ERROR] 페이지 조회 실패: {response.status_code} - {response.text}") + return None + + +def extract_summary_from_page(page_data: Dict) -> str: + props = page_data.get("properties", {}) + title = "(제목 없음)" + + # properties 안에서 title 타입 프로퍼티 찾기 + for prop in props.values(): + if prop.get("type") == "title": + title_text = prop.get("title", []) + if title_text: + # 여러 텍스트 조각이 있을 수 있으니 모두 연결 + title = "".join(t.get("plain_text", "") for t in title_text) + break + + url = page_data.get("url", "URL 없음") + return f"📌 노션 페이지 업데이트됨\n**제목**: {title}\n🔗 [바로가기]({url})" + + +def handle_notion_event(event: dict) -> Optional[str]: + """ + 모든 Notion 이벤트를 처리하며, + workspace_name이 "퍼스트가든"인 경우만 메시지 생성 후 반환합니다. + + 수신 이벤트를 콘솔에 출력하고, + 노션 API를 호출해 상세 데이터를 가져옵니다. + """ + # 1) 수신 이벤트 원본 로깅 + logging.info("🔔 수신된 Notion 웹훅 이벤트:") + logging.info(json.dumps(event, ensure_ascii=False, indent=2)) + + # 2) 워크스페이스 ID 확인 및 이름 조회 (예: workspace_id 필드) + workspace_id = event.get("workspace_id") + if not workspace_id: + logging.warning("⚠️ workspace_id 없음 - 이벤트 무시") + return None + + # 3) workspace_name 확인 (노션 API로 조회 필요) + # 실제 노션 공식 API에 workspace 이름 조회 엔드포인트가 없음 + # 테스트 편의를 위해 하드코딩 반환 + workspace_name = get_workspace_name(workspace_id) + if workspace_name != "퍼스트가든": + logging.info(f"워크스페이스명이 '{workspace_name}'이므로 무시함") + return None + + # 4) 이벤트 기본 정보 + event_type = event.get("type", "unknown") + timestamp = event.get("timestamp") + readable_time = None + if timestamp: + try: + readable_time = datetime.fromisoformat(timestamp.replace("Z", "+00:00")).strftime("%Y-%m-%d %H:%M:%S") + except Exception: + readable_time = timestamp + + entity = event.get("entity", {}) + entity_id = entity.get("id", "unknown") + entity_type = entity.get("type", "unknown") + + authors = event.get("authors", []) + author = authors[0] if authors else {"id": "unknown", "type": "unknown"} + author_type = author.get("type", "unknown") + author_id = author.get("id", "unknown") + + # 5) 구체적 변경사항 API 호출 (예: 페이지 상세 정보) + detail = fetch_entity_detail(entity_type, entity_id) + # detail은 dict or None + + message_lines = [ + f"📢 Notion 이벤트 발생: `{event_type}`", + f"- 🕒 시간: {readable_time}", + f"- 📁 워크스페이스: {workspace_name}", + f"- 📄 엔티티: `{entity_type}` / ID: `{entity_id}`", + f"- 👤 작성자: `{author_type}` / ID: `{author_id}`", + ] + + if detail: + detail_str = json.dumps(detail, ensure_ascii=False, indent=2) + message_lines.append(f"- 🔍 상세 내용:\n```json\n{detail_str}\n```") + + message = "\n".join(message_lines) + logging.info("🔔 생성된 메시지:\n" + message) + + return message + + +def get_workspace_name(workspace_id: str) -> Optional[str]: + """ + workspace_id로 노션 API 호출하여 workspace 이름을 가져옵니다. + 실제 API 문서에 맞게 경로와 파라미터 조정 필요. + + 현재 노션 API에 workspace 이름 조회 별도 엔드포인트 없음으로, + 하드코딩으로 처리함. + """ + return "퍼스트가든" # 테스트용 강제 반환 + + +def fetch_entity_detail(entity_type: str, entity_id: str) -> Optional[Dict]: + """ + entity_type에 따라 노션 API에서 상세 정보 조회 + """ + url = None + if entity_type == "page": + url = f"{NOTION_API_BASE}/pages/{entity_id}" + elif entity_type == "database": + url = f"{NOTION_API_BASE}/databases/{entity_id}" + elif entity_type == "block": + url = f"{NOTION_API_BASE}/blocks/{entity_id}/children" + else: + logging.warning(f"알 수 없는 entity_type: {entity_type}") + return None + + try: + response = requests.get(url, headers=HEADERS, timeout=5) + response.raise_for_status() + return response.json() + except Exception as e: + logging.error(f"노션 API 호출 실패: {e}") + return None + + +# ✅ 디버깅 / 단독 실행 시 +if __name__ == "__main__": + test_page_id = os.getenv("TEST_NOTION_PAGE_ID") + if not test_page_id: + print("[ERROR] 테스트용 페이지 ID가 설정되지 않았습니다. .env에 TEST_NOTION_PAGE_ID 추가 필요") + else: + page_data = get_page_details(test_page_id) + if page_data: + print("[DEBUG] 페이지 상세:") + print(json.dumps(page_data, ensure_ascii=False, indent=2)) + print("\n[SUMMARY]") + print(extract_summary_from_page(page_data))