슬랙 봇으로 커뮤너티 활성화하기(with CRM 메시지, 마인크래프트)

이번 글은 슬랙 기반 커뮤너티에서 참여자들에게 독려의 CRM 메세지를 보낸 경험과 커뮤너티 안의 게임채널에서 방문알람 봇을 만든 사례를 작성합니다. CS에 대한 개념이 살짝 필요하긴 하지만 기본적인 API 개념과 파이썬을 활용하여 쉽게 만들어볼 수 있는 사례라서 API를 제공하는 커뮤너티 어플리케이션에서 유용하게 쓰는데 도움이 되길 바랍니다!

---

1. 개요

필자는 개발자의 글쓰기 모임인 글또10기에 참여하고 있으며 또한 운영진으로 활동하고 있다. 아무래도 개발자 기반의 커뮤니티다 보니까 글쓰는 활동이나 내부 커뮤너티 활동들을 봇을 이용해서 활용할 기회가 많다. 예를 들면, 글 제출을 한다던지 혹은 제출된 글에 대해서 LLM으로 피드백을 해준다던지 자동화가 되어 있는 기능들이 많다. 이번 글에서는 Customer Relational Managment(a.k.a. CRM) 운영진의 일환으로 설날 맞이 보낸 CRM 메세지와 게임채널에서 마인크래프트를 이용한 서버를 운영하고 출석체크 봇을 만든 경험을 남기고자 한다.

 

2. 글또 CRM 메세지 보내기

2.1. CRM 메시지 기획하기

우리의 커뮤너티는 글쓰기를 주제로 약 4-6개월간 지속되는 활동으로 그 기간동안 활동한 내역을 집계하면 재밌는 얘기가 많이 나온다. 글또에서는 CRM 봇인 지니 봇이 비정기적으로 지금까지의 활동내역을 보여주고 Activation을 넛지하는 전통(?)이 있다. 지난 기수도 역시 소소하게 이런 메세지를 보냈었고 반응이 되게 좋았었다. 아래는 이번기수에 내가 받은(사실 내가 보낸) 메세지이다. 

 

10기 실제로 보낸(그리고 내가 받은) 메시지

 

역시 10기에도 보내자고 얘기가 계속 나왔고 결국 설날에 맞춰 보내기로 하였다. 내가 CRM 팀으로 이전한 건 이번이 처음이었기 때문에 추상적으로 어떻게 하겠다는 감은 있어도 실제로 해본적은 없기 때문에 좀 더 명확하게 이 프로세스를 전수 받고자 자처하여 메세지 봇을 만드는데 지원하였다. 사실 메세지를 보내는 시스템은 구축이 되어있었다. CRM 팀이 처음인 나는 지니봇이 작동하는 프로세스를 명확하게 하고 팀원들에게 공유하면 좋겠다고 생각했고 figjam을 이용해서 다음과 같이 정리했다. 참고로 우리의 지니봇은 주로 커뮤너티에서 전달하는 쿠폰이나 개인화 메세지에 초점을 맞춰 전달하고 있다.

 이 과정에 크게는 3가지의 단계가 있는데 

1. (데이터 마트 구축): 기존 또봇이 수집한 데이터 Log를 기반으로 참여자들의 글 제출, 메시지, 스레드 댓글, 이모지의 데이터를 집계하여 Data Mart를 만드는 과정
2. (지표 생성) 정보를 가지고 활동, 글, 커피챗 레벨을 지표로 만들고 친한 사람(팔로우)을 정의하는 과정
3. (메시지 전송) 참여자의 정보를 토대로 총합된 1,2의 정보를 불러와 각자의 지표에 맞춰 슬랙 봇으로 전송하는 과정

각자의 R&R이 나뉘고 나의 역할인 메시지 전송에서 작업을 시작했다. 이는 동민님의 Github에 잘 정리되어 있어서 답습하기 편했다. 하지만, 지니 봇은 특정 이벤트에만 작동시키는 경우가 많기 때문에 CRM 팀에서도 자주 사용하는 스크립트는 아니다. 따라서, 먼저 이를 파악하기 위해 동민님의 Github 레포를 뜯기 시작했다.

https://github.com/ddongmiin/geultto_genie_bot

GitHub - ddongmiin/geultto_genie_bot: 개발자글쓰기 커뮤니티 글또에서 "참여자 활동 목표 개선"을 위해

 

여기서 자주 사용하는 script는 /core/bigquery.py로 글또의 데이터는 빅쿼리에 적재되기 때문에 데이터를 저장하고 불러오는 등에 대한 메소드가 구현되어 있다. 빅쿼리와 슬랙을 주고 받는 기능에 대하여 개념적으로는 알지만 희미해서 새롭게 스크립트(message_basic.py)를 만들었다.

'''
genie봇이 메세지를 보내는 기본기능을 정의합니다. token은 관리자 문의
해당 스크립트를 실행하면 번호로 선택하여 실행하며, 기능을 수행할 대상에 대한 정보(user_id 혹은 스레드정보)를 전달해야합니다.
'''
import sys
import os
print(sys.version)
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from core.bigquery import BigqueryProcessor
bigquery_client = BigqueryProcessor(
    env_name="GOOGLE_APPLICATION_CREDENTIALS", database_id="geultto_10th"
)
SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN") # 윈도우 로컬 - 시스템 - 환경변수에 저장
USER_ID = ""  # 메시지를 보낼 대상의 Slack User ID


#하기 두개의 인자는 스레드의 링크복사 URL에 정보가 저장되어있음
URL = ''# 개인정보 이슈 삭제
URL_LIST = URL.split('/')
CHANNEL_ID = URL_LIST[4] #  채널 ID
THREAD_TS = URL_LIST[5][1:-7] + '.' + URL_LIST[5][-7:] #   스레드 메시지의 timestamp

# WebClient 초기화
client = WebClient(token=SLACK_BOT_TOKEN)

def send_message_to_user(user_id, message):
    '''
    Slack 메세지 전송 함수
    '''
    try:
        response = client.chat_postMessage(
            channel=user_id,
            text=message
        )
        # print(f"메시지가 성공적으로 전송되었습니다: {response['ts']}")
        return response['ts']
    except SlackApiError as e:
        print(f"오류 발생: {e.response['error']}")

def send_reply_to_thread(channel_id, thread_ts, message):
    '''
    스레드에 댓글 남기는 함수
    '''
    try:
        response = client.chat_postMessage(
            channel=channel_id,
            text=message,
            thread_ts=thread_ts
        )
        print(f"스레드에 댓글이 성공적으로 전송되었습니다: {response['ts']}")
    except SlackApiError as e:
        print(f"오류 발생: {e.response['error']}")

def get_bigquery(query):
    '''
    단순 빅쿼리 조회용
    '''
    try:
        result = bigquery_client.run_query_to_dataframe(query)
        print('Big query 조회 성공')
        # print(result)
        return result
    except Exception as e:
        print(f'Bigquery 오류 발생 {e}')

def send_bigquery_to_slack(user_id, query):
    data = get_bigquery(query)
    if data:
        for row in data:
            message = f'Bigquery 데이터: {row}'
            send_message_to_user(user_id,message)
    else:
        print('전송할 데이터가 없습니다.')


# 메뉴 기반 실행
if __name__ == "__main__":
    print("실행할 기능을 선택하세요:")
    print("1. 사용자에게 메시지 보내기")
    print("2. 스레드에 댓글 달기")
    print("3. Bigquery 데이터 조회")
    print("4. Bigquery 데이터 사용하여 사용자에게 전송")

    choice = input("번호를 입력하세요: ")

    if choice == "1":
        message = "안녕하세요! Slack Bot에서 보낸 메시지입니다."
        send_message_to_user(USER_ID, message)
    elif choice == "2":
        message = "스레드에 댓글 달기 테스트중입니다."
        send_reply_to_thread(CHANNEL_ID, THREAD_TS, message)
    elif choice == "3":
        query = 'SELECT * FROM geultto.geultto_10th.user_super_mart LIMIT 1;'
        get_bigquery(query)
    elif choice == "4":
        query = 'SELECT * FROM geultto.geultto_10th.user_super_mart LIMIT 1;'
        send_bigquery_to_slack(USER_ID, query)
    else:
        print("잘못된 입력입니다. 다시 실행해주세요.")

 

2.2.  사전 테스트 진행

최종적으로 우리가 보낼 메세지는 총 639명을 대상으로한 Direct Message이다. 사실 이정도의 많은 사람들에게 메세지를 보내는게 부담이여서 먼저 운영진 34명을 대상으로 설 일주일 전 테스트를 진행했다. 초안의 메세지는 다음과 같고 몇 가지 문제를 발견할 수 있었다.

모바일 기준 메시지, 확실히 개행이 빡빡하다.

 

• 컨텐츠 문제

1. 들여쓰기 내여쓰기가 들쭉 날쭉함
2. 요약하여 쓸 수 있는 내용이 서술형으로 작성 됨

1번을 고민해보자면 슬랙은 멀티플랫폼이기 때문에 데스크톱, 모바일에서의 개행이 다르다는 점이다. 특히, 사람들에게 보내는 메세지가 처음 설계에서는 짧지 않았기 때문에 불릿포인트는 들여쓰기가, 줄글은 내어쓰기가 되어 가독성이 좋지 않았다.

2번은 불릿 포인트는 개조식, 그 밖은 서술형인 형식이 무너진 탓이였다. 되도록 불릿 포인트는 요약하여 작성하고 줄글은 문단마다 맨 뒤로 빼는 한편 "인용" Syntax를 이용해서 대화형임을 강조했다. CRM 메시지이기 때문에 사람마다 줄글이 미묘하게 다르기 때문에 이를 맞추려고 말 그대로 메세지를 2시간정도 "깍았다". 밑에 메세지가 최종본인데 나름 깔끔해졌다.(활동요약 개행이 좀 거슬린다.)

 

정리된 메시지 최종본

두 번째로는 테크적인 관점인데, 운영진 30명 정도에 보내는 정도야 별 문제 없었지만, 639명에게 보내게 되면 꽤나 골치아픈 경우의 수가 떠올랐다.

1. slack api 의 초당 메세지 limit으로 인한 끊김
2. 누구에게 어떤 메세지를 보냈다는 로그
3. 너무 느린 속도..

1번은 API TOKEN 발급받은 동민님께 확인 받았다. 제일 좋은 tier라고 하셔서 문제는 없을 거라고 하셨고 나중에 확인해본결과 3 Tier 였다. 실제로 보내는 것도 문제가 없었긴 했는데 살짝 아슬한 정도가 아닐까 싶다. 다음에 체크해볼 것! 

https://api.slack.com/apis/rate-limits

Rate Limits

 

2번은 혹여나 전송 시 에러가 날 수 있는 상황을 가정한다면, 누구까지 보냈는가가 중요했다. 그래야 받지 못한 사람부터 다시 시작할 수 있기 때문이다. 게다가 CRM 메세지 답게 모든 사람들이 다양한 경우의 수의 메시지를 받을 수 있기 때문에 확인 수단이 필요했다. 그래서 loop안에 받은사람, 메세지 내용, 해당 스레드의 ts식별자등을 저장한 csv파일을 로컬에 저장하는 로직을 추가했다. 

#코드 중 일부 
    #for문
	cnt += 1
    df_temp = pd.DataFrame({'name':name_list,'text':text_list,'ts':ts_list})    
    if cnt % 100 == 0:
        time.sleep(1)
        print(f'{cnt}명, {row[1]}까지 전달완료')
time_record = time.strftime("%Y%m%d_%H%M%S", time.localtime(time.time()))
pd.concat([df_sended,df_temp]).to_csv(f'./history/message_sended_{time_record}.csv', encoding = 'utf-8-sig')
print(f'최종 {cnt}명 전달완료')
end = time.time()
print(round(end - start), '초 소요됨')

3번은 사실 해결못한 부분이기도 했는데, 속도가 너-무 느렸다. 빅쿼리에서 데이터를 불러오는건 그렇다고 쳐도 가져온 데이터를 Pandas DataFrame을 .iterrow()로 접근하여 루프를 돌리는데  운영진 34명에게는 23초였으니 639명은 5분이 넘어갈께 뻔했다. 병렬식으로 보낼 수도 있을 것 같은데 굳이 지금 프로세스에서 바꾸고 싶지 않았다. 잘 작동하니까.. 🤣 그리고 이걸 깨달을 때 쯤이면 이미 보내기로한 due date 직전이였다. 

그정도까진 아니긴해 ~

 

결국 6분에 걸쳐 모든 사람들에게 메시지는 정상적으로 전송되었지만 후문으로는 왜 비동기식으로 하지 않았는가 에 대한 물음이 있었다고 한다 ㅋ_ㅎ (역시 개발자분들 ! 다음엔 그렇게 해볼게요) 사실 비동기식 처리 방법에 대해서 얼마안된 수민님의 글을 읽은 적이 있는데 사람은 망각의 동물이 맞나보다. 

어쨋든 돌아가잖아 한잔해

"테블리" 서비스의 핵심 기능 "개발"이라쓰고 "노가다"라고 읽는다 (1) - 데이터 수집(티스토리편)

"테블리" 서비스의 핵심 기능 "개발"이라쓰고 "노가다"라고 읽는다 (1) - 데이터 수집(티스토리편)

 

또버지이신 은찬님도 슬랙봇 비동기 요청에 글을 쓰셨다! 마지막 CRM 메세지에 적용해보려고 한다.

https://daco2020.tistory.com/854

Python aiocache 로 비동기 Slack API 요청 캐싱하기

 

2.3. CRM 메시지 마무리

설 전날 결국 메시지를 보내 마무리 했다. 배포하는 날 기도하는 심정이 이런걸까? 금요일날 보내기 위해서 점심부터 계속 에러는 없을까 그 짤막한 코드를 보면서 계속 체크해봤다. 점점 쌓이는 지니봇의 DM.. 

6분은 좀 너무하긴 하네요

정확히 저녁 8시에 보내기 시작한 봇은 무리없이 6분간 순차 전송을 마무리하였고 사람들의 반응도 좋았어서, CRM팀으로서 뿌듯함을 느꼈다.

 

3. 갑자기 마인크래프트크요?

3.1. 마크 활동 시작

슬랙 봇에 이어지는 이야기인데, 서론은 이렇다. 갑자기 어떤분의 권유로 마인크래프트가 흥하기 시작했다. 그것도 설연휴에. 그렇게 정수님은 GCP을 가입하신 무료 크레딧 $400달러로 서버를 팠고 약 5 일차 정도 재밌게 진행했다. 근데 마인크래프트에서는 채팅기능에 더하여 사람이 접속하는 정보나 업적달성, 죽음 표시 등이 표기된다. 어 이거 데이터잖아? 하는 생각이 들었고 GCP 내부에 접근하면 LLM을 이용해서 인게임에서 마인크래프트 공략 등을 물어보는 세팅도 가능할 것 같아서 정수님께 문의 했다.  

그렇다 나는 말벌 아니 데이터 아저씨다..호다닥 🏃‍♂️

/Logs 파일을 확인해보니 날짜별로 분할되어있는 파일과 최신 log 2가지가 존재했다.  현재 기록들은 latest.log에 실시간으로 적히고 일정 용량이  차면 날짜별로 압축파일 .gz 파일로 저장되는 모양이였다. 

Logs 디렉토리의 파일들

여기서 재밌는 생각이 난게 채팅기록을 가지고 있으니 이걸 가지고 3-4일동안 진행된 정보를 요약하면 재밌겠다는 생각이 들었다. 게임채널이지만 시간이 나지 못해 참여 못한 사람들에게 안의 상황을 공유하면서 공감대를 키우고 유입 유도를 하는 한편, 기존 참여자들에겐 정보 요약을 통해 추억을 회상하는 넛지를 줄 수 있을 것 같아서 날짜별로 로그를 이용해서 LLM 에 태웠다.

초기 요약은 깔끔하다

여기서 재밌는 이슈가 있는데, 채팅파일이 워낙 길다보니까 무료버전의 ChatGPT, Claude, 뤼튼(OPEN AI 기반이긴 하지만)에서 모두 텍스트 Length 초과로 거절당했다... 반면 Perplexity 와 Gemini는 토큰이 넉넉한지 요약은 해주더라. 결국 Gemini에게 요약하는 형식을 가지고 3일차 정도 밀어 넣어봤는데 Context Length 이슈로 3일차 정도 뒤에는 자기 멋대로 요약을 하기 시작했다... 아놔. 결국 마음편하게 ChatGPT를 다시 구독했다.

Context가 길어진 후, 응답 구조가 변경되었다.

반응 감사해요 ~

좀 더 프롬프트를 깍고 일정하게 형식을 few shot해놓으면 퀄리티가 좋아지겠지만 그렇게 까지 시간을 쓰고 싶지 않았고 날짜도 5일차 정도 밖에 되지 않았기 때문에 간단하게 정리만 했다.

3.2 접속 알림 봇을 만들까?

어쩌다보니 마인크래프트 GCP 서버 관리까지 참여하고 있어서, 이 때를 틈타 CRM에서 했던 연장으로 간단하게 출석 체크 봇을 만들고 싶어졌다. 사실 생존게임이라는게 혼자하기보다는 누구랑 같이 할때 제일 재밌는거라 함께 하는게 젤 중요하다 생각했다. (그리고 2주정도 하다 말꺼니까)   그러던 중 문득 생각난 것이, 마인크래프트가 꽤나 API에 적극적이라, 실제로 인터넷 방송에서는 도네이션을 연결해서 스트리머에게 필요한 물품을 준다던가 갑자기 점프를 시킨다더가 즉사를 시킨다던가 하는 식으로 재미를 더하고 있었던 장면이 생각났다.

https://www.youtube.com/watch?v=bgA6BCSSQaY

 

그래서 간단하게 먼저 출입 봇을 만들어서 누가 활동하고 있는지 등에 대한 정보를 전달해주면 그걸 보고 다른사람이 함께 한다던지에 대한 넛지를 줄 수 있을 거란 생각이 들었다. GPT 한테 물어봤다.

역시 GOD - PT

이를 바탕으로 요약한 절차는 다음과 같다.

1. 먼저 서버의 설정에서 서버의 상태(접속인원 등)을 확인할 수 있는 설정과 포트를 열고
2. Python을 이용해서 해당 정보를 가져오고
3. 슬랙 봇을 이용해서 스레드에 쏴주면 된다.

3번은 CRM에서 진행했던 거라 아주 기억이 생생할... 줄알았지만 Github에 머지해놓은것 있으니까 이걸 참고하고 1번 마크 서버의 정보를 가져올 수 있는 mcstatus 모듈이 당연히 있더라 

https://pypi.org/project/mcstatus/

mcstatus

위 내용을 조합해서 2번을 만들면 아주 쉽게 다음과 같이 만들 수 있다. 마인크래프트와 파이썬 스크립트의 위치는 동일한 GCP 이기 때문에 Localhost를 이용해서 어렵지 않게 서버정보를 불러올 수 있다. 이렇게 정보를 가져올 수 있다는걸 확인했으니, 이제 슬랙 봇 api를 이용해서 보내기만 하면된다.

import os
import time
import requests
import json
from mcstatus import JavaServer

# 🛠 환경변수에서 Slack Bot Token 불러오기
SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
if not SLACK_BOT_TOKEN:
    raise ValueError("❌ 환경변수 SLACK_BOT_TOKEN이 설정되지 않았습니다.")

# 🛠 Slack 설정
CHANNEL_ID = "C01XXXXXX"
THREAD_TS = "000000.00000"  # 기존 메시지 thread_ts 값

# 🛠 마인크래프트 서버 설정
SERVER_IP = "127.0.0.1"
QUERY_PORT = 25565
server = JavaServer.lookup(f"{SERVER_IP}:{QUERY_PORT}")

# 🛠 ID → 이름 매핑 (등록되지 않은 경우 ID 그대로 출력)
player_name_map = {
 "ID" : "이름"
}

# 🔄 이전 접속자 목록 저장
previous_players = set()

# 📌 Slack 스레드에 댓글 남기기
def send_slack_reply(message):
    url = "https://slack.com/api/chat.postMessage"
    headers = {"Authorization": f"Bearer {SLACK_BOT_TOKEN}", "Content-Type": "application/json"}

    data = {
        "channel": CHANNEL_ID,
        "text": message,
        "thread_ts": THREAD_TS  # 기존 메시지의 스레드에 댓글 작성
    }

    response = requests.post(url, headers=headers, data=json.dumps(data))
    response_data = response.json()

    if response_data["ok"]:
        print(f"✅ Slack 스레드에 댓글 작성 완료: {message}")
    else:
        print("❌ Slack 댓글 작성 실패:", response_data)

# 📌 서버 인원 체크 루프 실행
def monitor_minecraft_server():
    global previous_players

    print("🔄 마인크래프트 서버 접속 감지 시작...")

    while True:
        try:
            # 서버 상태 가져오기
            status = server.status()
            current_players = set(player.name for player in status.players.sample) if status.players.sample else set()

            # 🔹 새로운 플레이어 입장 확인
            new_players = current_players - previous_players
            left_players = previous_players - current_players

            message_list = []

            # ✅ 여러 명이 동시에 입장하면 한 번에 메시지 출력
            if new_players:
                joined_names = [player_name_map.get(player, player) for player in new_players]
                joined_message = f"🎉 {', '.join(joined_names)} 님이 입장했습니다!"
                message_list.append(joined_message)

            # ✅ 여러 명이 동시에 퇴장하면 한 번에 메시지 출력
            if left_players:
                left_names = [player_name_map.get(player, player) for player in left_players]
                left_message = f"👋 {', '.join(left_names)} 님이 퇴장했습니다!"
                message_list.append(left_message)

            # ✅ 현재 인원도 한 번만 출력
            if message_list:
                total_players = len(current_players)
                message_list.append(f"🔹 현재 접속 인원: {total_players}명")
                send_slack_reply("\n".join(message_list))

            # 🔹 접속자 목록 업데이트
            previous_players = current_players

        except Exception as e:
            print(f"❌ 서버 정보를 가져오는 중 오류 발생: {e}")

        time.sleep(5)

# 🔥 실행
if __name__ == "__main__":
    monitor_minecraft_server()

 

위 코드를 실행하여도 GCP SSH가 닫기면 떨어지기 때문에 screen을 이용해서 백그라운드에서도 돌게 만들었다.

# screen 생성
screen -S mc_slack_bot
# 파이썬 스크립트 실행
source /가상환경명/bin/activate
python monitor_sendmsg.py
# CTL + A , D를 통해 screen 빠져나오기

#스크린 확인(사용자 디렉토리 귀속)
screen -ls

완성!

4. 마무리

Slack API가 잘 되어있고 GCP같은 클라우드 환경도 언제든지 쉽게 접속할 수 있는 환경이라 쉽게 배웠다. 더불어 옆에 있는 개발자분들에게 물어가면서 배워가는 점도 한 몫하고, 아이디어만 있다면 ChatGPT 를 이용해서 코드 구현부분을 외주 줄 수 있다는 점도 좋은 것 같다. 최적화나 설계 부분에 대해서는 여전히 CS적인 지식이 필요하지만 짧게 짧게 Prototype을 만들고 구현하는 경험이 재미있었다. 예전에는 프로그램을 만들기 위해서 가능한 기능성이 높은 구조를 높게 잡고 하다보니 지치거나 포기하는 경우가 많았는데 이제 경험이 쌓이다보니 작은 단위로 Divide(&Validation) and Conquer 를 하는 개발도 익숙해진 것 같다. 남은 글또 기간이나 내 일에서도 SaaS 활용과 엔지니어링을 지속적으로 키워나가면 확실한 무기가 될 수 있을 것 같다.

그리고 재밌고 유용하잖아 한잔해 🍻