firstgarden/fgtools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initial commit - unified FGTools from static, weather, mattermost-noti

Browse Source
This commit is contained in:
KWON
2025-12-31 09:56:37 +09:00
commit 4ff5dba4b1
29 changed files with 5786 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

344
services/notification/notion.py Normal file
View File

@ -0,0 +1,344 @@
# ===================================================================
# services/notification/notion.py
# Notion 웹훅 처리 서비스 모듈
# ===================================================================
# Notion 웹훅 이벤트를 처리하고 알림 메시지를 생성합니다.
# 페이지 생성, 수정, 삭제 등의 이벤트를 지원합니다.
# ===================================================================
"""
Notion 웹훅 처리 서비스 모듈
Notion 웹훅 이벤트를 수신하고 처리하여 알림 메시지를 생성합니다.
사용 예시:
from services.notification.notion import NotionWebhookHandler
handler = NotionWebhookHandler(api_secret)
message = handler.handle_event(event_data)
"""
import json
import logging
from datetime import datetime
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__)
# Notion API 설정
NOTION_API_BASE = "https://api.notion.com/v1"
NOTION_VERSION = "2022-06-28"
# 이벤트 타입별 라벨
EVENT_TYPE_LABELS = {
"page.created": "📝 새 페이지 생성",
"page.content_updated": "✏️ 페이지 내용 수정",
"page.properties_updated": "🔄 페이지 속성 변경",
"page.deleted": "🗑️ 페이지 삭제",
"page.restored": "♻️ 페이지 복구",
"page.moved": "📁 페이지 이동",
"page.locked": "🔒 페이지 잠금",
"page.unlocked": "🔓 페이지 잠금 해제",
"database.created": "📊 새 데이터베이스 생성",
"database.content_updated": "📊 데이터베이스 업데이트",
"block.created": "➕ 블록 생성",
"block.changed": "📝 블록 변경",
"block.deleted": "➖ 블록 삭제",
"comment.created": "💬 새 댓글",
"comment.updated": "💬 댓글 수정",
"comment.deleted": "💬 댓글 삭제",
}
def get_page_details(page_id: str, api_secret: str) -> Optional[Dict]:
"""
Notion 페이지 상세 정보 조회
Args:
page_id: 페이지 ID
api_secret: Notion API 시크릿
Returns:
페이지 상세 정보 또는 None
"""
url = f"{NOTION_API_BASE}/pages/{page_id}"
headers = {
"Authorization": f"Bearer {api_secret}",
"Notion-Version": NOTION_VERSION,
"Content-Type": "application/json"
}
try:
response = requests.get(url, headers=headers, timeout=10)
if response.status_code == 200:
return response.json()
else:
logger.error(f"페이지 조회 실패: {response.status_code} - {response.text}")
return None
except Exception as e:
logger.error(f"페이지 조회 예외: {e}")
return None
def get_database_details(database_id: str, api_secret: str) -> Optional[Dict]:
"""
Notion 데이터베이스 상세 정보 조회
Args:
database_id: 데이터베이스 ID
api_secret: Notion API 시크릿
Returns:
데이터베이스 상세 정보 또는 None
"""
url = f"{NOTION_API_BASE}/databases/{database_id}"
headers = {
"Authorization": f"Bearer {api_secret}",
"Notion-Version": NOTION_VERSION,
"Content-Type": "application/json"
}
try:
response = requests.get(url, headers=headers, timeout=10)
if response.status_code == 200:
return response.json()
else:
logger.error(f"데이터베이스 조회 실패: {response.status_code}")
return None
except Exception as e:
logger.error(f"데이터베이스 조회 예외: {e}")
return None
class NotionWebhookHandler:
"""
Notion 웹훅 이벤트 핸들러
Notion에서 전송하는 웹훅 이벤트를 처리하고
알림 메시지를 생성합니다.
Attributes:
api_secret: Notion API 시크릿
allowed_workspaces: 허용된 워크스페이스 이름 목록
"""
def __init__(
self,
api_secret: Optional[str] = None,
allowed_workspaces: Optional[List[str]] = None
):
"""
Args:
api_secret: Notion API 시크릿 (None이면 설정에서 로드)
allowed_workspaces: 허용된 워크스페이스 목록 (None이면 모두 허용)
"""
if api_secret is None:
config = get_config()
api_secret = config.notion.get('api_secret', '')
self.api_secret = api_secret
self.allowed_workspaces = allowed_workspaces or []
self.headers = {
"Authorization": f"Bearer {self.api_secret}",
"Notion-Version": NOTION_VERSION,
"Content-Type": "application/json"
}
def fetch_entity_detail(self, entity_type: str, entity_id: str) -> Optional[Dict]:
"""
엔티티 상세 정보 조회
Args:
entity_type: 엔티티 타입 ('page', 'database', 'block' 등)
entity_id: 엔티티 ID
Returns:
엔티티 상세 정보 또는 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}"
else:
logger.warning(f"지원하지 않는 엔티티 타입: {entity_type}")
return None
try:
response = requests.get(url, headers=self.headers, timeout=10)
if response.status_code == 200:
return response.json()
else:
logger.error(f"엔티티 조회 실패: {response.status_code}")
return None
except Exception as e:
logger.error(f"엔티티 조회 예외: {e}")
return None
def extract_title_from_properties(self, properties: Dict) -> str:
"""
속성에서 제목 추출
Args:
properties: Notion 속성 딕셔너리
Returns:
제목 문자열 (없으면 기본값)
"""
# 일반적인 제목 속성 이름들
title_keys = ['Name', 'name', '이름', '제목', 'Title', '작업 이름']
for key in title_keys:
if key in properties:
prop = properties[key]
if prop.get('type') == 'title':
title_arr = prop.get('title', [])
if title_arr:
return ''.join(t.get('plain_text', '') for t in title_arr)
# title 타입 속성 찾기
for prop in properties.values():
if prop.get('type') == 'title':
title_arr = prop.get('title', [])
if title_arr:
return ''.join(t.get('plain_text', '') for t in title_arr)
return "(제목 없음)"
def extract_editor_from_properties(self, properties: Dict) -> str:
"""
속성에서 편집자 정보 추출
Args:
properties: Notion 속성 딕셔너리
Returns:
편집자 이름 (없으면 기본값)
"""
editor_keys = ['최종 편집자', 'Last edited by', 'Editor']
for key in editor_keys:
if key in properties:
prop = properties[key]
if prop.get('type') == 'last_edited_by':
editor = prop.get('last_edited_by', {})
return editor.get('name', '(알 수 없음)')
return "(알 수 없음)"
def handle_event(self, event: Dict) -> Optional[str]:
"""
Notion 웹훅 이벤트 처리
Args:
event: 웹훅 이벤트 데이터
Returns:
알림 메시지 또는 None (무시할 이벤트)
"""
# 이벤트 로깅
logger.info(f"수신된 Notion 이벤트: {json.dumps(event, ensure_ascii=False)[:500]}")
# 워크스페이스 확인
workspace_id = event.get('workspace_id')
if not workspace_id:
logger.warning("workspace_id 없음 - 이벤트 무시")
return None
# 허용된 워크스페이스 확인 (설정된 경우)
if self.allowed_workspaces:
workspace_name = self.get_workspace_name(workspace_id)
if workspace_name not in self.allowed_workspaces:
logger.info(f"허용되지 않은 워크스페이스: {workspace_name}")
return None
# 이벤트 정보 추출
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')
# 상세 정보 조회
detail = self.fetch_entity_detail(entity_type, entity_id)
if not detail:
logger.warning("상세 정보 조회 실패")
return None
# 정보 추출
properties = detail.get('properties', {})
page_title = self.extract_title_from_properties(properties)
editor_name = self.extract_editor_from_properties(properties)
page_url = detail.get('url', 'URL 없음')
# 이벤트 라벨
event_label = EVENT_TYPE_LABELS.get(event_type, f"📢 Notion 이벤트: `{event_type}`")
# 메시지 구성
message = (
f"{event_label}\n"
f"- 🕒 시간: {readable_time or '알 수 없음'}\n"
f"- 📄 페이지: {page_title}\n"
f"- 👤 작업자: {editor_name}\n"
f"- 🔗 [바로가기]({page_url})"
)
logger.info(f"생성된 메시지: {message[:200]}...")
return message
def get_workspace_name(self, workspace_id: str) -> str:
"""
워크스페이스 이름 조회
Note: Notion API에서 별도의 워크스페이스 조회 엔드포인트가 없어
현재는 하드코딩되어 있습니다.
Args:
workspace_id: 워크스페이스 ID
Returns:
워크스페이스 이름
"""
# TODO: 실제 API가 지원되면 구현
return "퍼스트가든"
def create_summary_from_page(self, page_data: Dict) -> str:
"""
페이지 데이터에서 요약 메시지 생성
Args:
page_data: 페이지 상세 데이터
Returns:
요약 메시지
"""
properties = page_data.get('properties', {})
title = self.extract_title_from_properties(properties)
url = page_data.get('url', 'URL 없음')
return f"📌 노션 페이지 업데이트됨\n**제목**: {title}\n🔗 [바로가기]({url})"