166 lines
5.8 KiB
Python
166 lines
5.8 KiB
Python
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))
|