노션 DB 추가/수정 시 슬랙 알림 자동화 (1) - 구현

사내에서 문서 정리, 의사소통을 위해 노션과 슬랙을 주로 사용하고 있습니다.

그런데 여기서 생긴 귀찮음 중 하나가 회의실 예약 과정이었는데요.

Info

기존 회의실 예약 과정
1. 노션 회의실 캘린더에서 회의실 등록하기
2. 슬랙 회의실 채널에 1번 내용 공유하기

 

1번 과정까지는 회의실 관리를 위해 필요하다고 생각이 드는데요.

문제는 2번이었습니다. "회의실 예약했으면 됐지 메시지까지 수동으로 공유해야 돼? 너무 귀찮은데??"

그렇게 해서 시작된 자동화 퀘스트....

회의봇 미리보기

이 글에서는 파이썬으로 노션 API, 슬랙 API, 도커 배포까지 실제 실무에 적용하며 구현한 내용을 공유하고자 합니다.

1. 노션 DB 읽어오기

노션 API를 활용하면 노션에 작성된 DB의 데이터를 읽어 올 수 있습니다.

이를 위해 필요한 것이 2가지 있는데, Notion Token과 Databas ID입니다.

Notion Token

노션 개발자 포털에 들어가면 토큰을 발급받을 수 있어요

노션 개발자 포털

토큰을 발급받았다면 잘 메모해 두세요~

Database ID

데이터베이스 ID는 노션에서 확인 가능한데요. 확인할 데이터베이스 페이지에서 우상단에 있는 공유를 눌러보면

공유 - 링크 복사

링크 복사 버튼을 누른 뒤 붙여 넣기 하면 URL이 나오는데 https://www.notion.so/{user_id}/{database_id}/~

{database_id}가 여기에 해당됩니다.

구현하기

여기까지 왔으면 끝났습니다(?)ㅋㅋ

이제 조회만 하면 되거든요.

from datetime import datetime, timezone
from notion_client import Client

notion = Client(auth="Notion Token")
DATABASE_ID = "Database ID"

def fetch_pages():
    today = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0).isoformat()

    response = notion.databases.query(
        database_id=DATABASE_ID,
        filter={"property": "날짜", "date": {"on_or_after": today}},
    )
    return response.get("results", [])

코드는 매우 간단한데요. notion_client를 활용하면 손쉽게 Notion DB를 연동할 수 있습니다.

눈여겨볼 점은 filter에 on_or_after 값을 통해 오늘 날짜 이후의 데이터를 가져온다는 점입니다.

이 부분은 구현하는 방향에 따라 달라질 수 있으니 노션 API 레퍼런스를 확인하면 커스텀 가능합니다.

def extract_page_info(page):
    # print(page)
    props = page.get("properties", {})
    get_title = lambda p: (p or {}).get("title", [{}])[0].get("text", {}).get("content", "제목 없음") if (p or {}).get("title") else "제목 없음"
    get_location = lambda p: "지정 안됨" if not p or p.get("select") is None else p.get("select", {}).get("name", "지정 안됨")
    get_people = lambda p: [person.get("name", "이름 없음") for person in (p or {}).get("people", [])]
    get_message = lambda p: (p.get("rich_text", [{}])[0].get("text", {}).get("content", "") if p and p.get("rich_text") else "")

    return {
        "name": get_title(props.get("회의명")),
        "start": props.get("날짜", {}).get("date", {}).get("start"),
        "end": props.get("날짜", {}).get("date", {}).get("end"),
        "location": get_location(props.get("위치")) or "지정 안됨",
        "attendees": get_people(props.get("참석자")),
        "edited_time": page.get("last_edited_time"),
        "message": get_message(props.get("메시지")),
        "url": page.get("url"),
    }

이런 포맷팅 함수를 만들어서, DB 조회 후 결과로 받은 결과를 필요한 정보를 발라내어 가공하여 줍니다.

여기서 삽질 포인트는 DB의 프로퍼티는 필수 입력 값이 아니기 때문에 빈 값이 올 수 있다는 점, 그래서 예외 처리를 꼼꼼하게 해주어야 합니다!

---

이제 노션에서 데이터를 읽어왔고 가공까지 마쳤으니 세부적인 로직을 짜주었습니다.

1. 3분마다 노션 데이터(회의실 DB)를 읽어온다.
2. 데이터를 읽어올 때마다 json 파일로 저장해 둔다.
3. 다시 데이터를 읽어올 때, 이전에 저장한 내용과 비교한다. (3분 전 데이터와 현재 시간 데이터)
4. 추가된 회의가 있거나 수정된 회의가 있는 경우로 분리한다.
5. 슬랙에 보낼 메시지를 포맷팅 하여 발송한다.

위와 같은 내용을 정리하여 반영하였습니다.

아래에는 슬랙에 대한 내용으로 넘어가 보겠습니다.

2. 슬랙 API

슬랙으로 메시지를 보내기 위해선 WEBHOOK이 필요합니다.

웹훅은 생성은 됐다고 가정하고 과정은 생략하겠습니다.

import os
import requests
import json

WEBHOOK_URL = "WEBHOOK_URL"


def send_slack_message(message: str, status: str):
    if status not in ["add", "update"]:
        raise ValueError("status는 'add' 또는 'update'만 가능합니다.")

    status_text = "*🔔 새로운 회의가 추가되었습니다!*\n" if status == "add" else "*✏️ 회의 정보가 수정되었습니다!*\n"

    print(f'{status_text}{message}')

    payload = {
        "attachments": [
            {
                "pretext": f'{status_text}',
                "color": "#2ECC71" if status == "add" else "#F1C40F",
                "text": message
            }
        ]
    }

    headers = {"Content-Type": "application/json"}
    response = requests.post(WEBHOOK_URL, headers=headers, data=json.dumps(payload))

    if response.status_code == 200:
        print(f"✅ ({status})Slack 메시지 전송 성공!")
    else:
        print(f"❌ 오류 발생: {response.status_code}, {response.text}")

이렇게 하면 원하는 채널에 커스텀한 메시지를 발송할 수 있습니다. 참 쉽죠?

그런데 노션 필드 중 "참석자"라는 필드가 있는데 여기 태그 된 인원은 슬랙에서도 태그를 해주어야 했는데요.

실제로 노션에서 태그 하듯이 "@송유찬" 이렇게 보낸다고 태그가 되진 않더라고요.

슬랙 API로 메시지를 보낼 때 태그할 때는 <@{slack_id}> 이렇게 담아야 하는 걸 알았습니다.

슬랙 프로필에서 멤버 ID 복사

여기서 {slack_id}는 프로필에서 확인 가능합니다.

그럼 모든 slack_id를 저장해 두면 태그 할 수 있겠다고 생각했고 모든 ID를 취합, 정리하여 저장해두어야 하나? 생각이 들었는데요.

그건 너무 비효율적인 것 같아(신규 입사자가 생기기도 하고) 방법을 더 찾아보던 중 슬랙 API를 통해 모든 멤버의 슬랙 ID를 조회할 수 있다는 사실을 알았습니다 ㅋㅋ

Slack ID 조회

OAuth Tokens

위에서 만들었던 웹훅에서 OAuth Token을 얻을 수 있습니다.

import os
import time

from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

SLACK_TOKEN = "SLACK_TOKEN"
slack = WebClient(token=SLACK_TOKEN)


def get_slack_user_ids():
    try:
        response = slack.users_list()
        users = response['members']

        user_dict = {}
        for user in users:
            real_name = user['profile'].get('real_name', '')
            user_id = user['id']
        
        return user_dict

    except SlackApiError as e:
        print(f"오류 발생: {e.response['error']}")
        return None

이렇게 해서 "송유찬": "{slack_id}" 형태로 전사원의 slack id를 가져올 수 있었고 태그 기능까지 구현을 마쳤습니다.
다음 글에서는 도커를 이용한 배포 과정을 공유해 보겠습니다.