mattermost-noti/lib/notion_api.py

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
from lib.common import EVENT_TYPE_LABELS

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를 호출해 상세 데이터를 가져옵니다.
    """
    import logging
    from datetime import datetime

    # 1) 수신 이벤트 원본 로깅
    logging.info("🔔 수신된 Notion 웹훅 이벤트:")
    logging.info(json.dumps(event, ensure_ascii=False, indent=2))

    # 2) workspace_id 및 workspace_name 확인
    workspace_id = event.get("workspace_id")
    if not workspace_id:
        logging.warning("⚠️ workspace_id 없음 - 이벤트 무시")
        return None

    workspace_name = get_workspace_name(workspace_id)
    if workspace_name != "퍼스트가든":
        logging.info(f"워크스페이스명이 '{workspace_name}'이므로 무시함")
        return None

    # 3) 이벤트 기본 정보
    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")

    # 4) 작성자 정보
    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)

    if not detail:
        logging.warning("⚠️ 상세 페이지 정보를 가져오지 못함")
        return None

    # 6) 페이지 제목 추출
    page_title = "(제목 없음)"
    try:
        props = detail.get("properties", {})
        title_prop = props.get("작업 이름", {})
        if title_prop.get("type") == "title":
            title_arr = title_prop.get("title", [])
            if title_arr:
                page_title = "".join(t.get("plain_text", "") for t in title_arr)
    except Exception as e:
        logging.error(f"페이지 제목 추출 중 오류: {e}")

    # 7) 작성자 이름 추출 (최종 편집자 이름)
    editor_name = "(알 수 없음)"
    try:
        last_edited_by = props.get("최종 편집자", {}).get("last_edited_by", {})
        editor_name = last_edited_by.get("name", editor_name)
    except Exception as e:
        logging.error(f"작성자 이름 추출 중 오류: {e}")

    # 8) 페이지 URL
    page_url = detail.get("url", "URL 없음")

    # 9) 메시지 구성
    event_label = EVENT_TYPE_LABELS.get(event_type, f"📢 Notion 이벤트 발생: `{event_type}`")
    message = (
        f"{event_label}\n"
        f"- 🕒 시간: {readable_time}\n"
        f"- 📄 페이지 제목: {page_title}\n"
        f"- 👤 작업자: {editor_name}\n"
        f"- 🔗 [페이지 바로가기]({page_url})"
    )

    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))