Final Project Agile 1차가 끝이나고 관리자 입장에서 오류, 비용, 관리를 위한 Slack Chatbot 기능을 배웠습니다.
우선 Slack 설치와 가입 및 워크스페이스 생성을 완료했다는 가정하에 다음 단계부터 진행하겠습니다.
○ Slack API 링크로 접속해서 Create New App을 눌러서 새로운 App 생성
○ From scratch 클릭
○ App Name 입력과 & Slack workspace 선택 후 Create App 클릭
○ ChatBot 생성을 위해서는 3개의 Value가 필요
● Signing Secret
● Slack Bot Token
● Slack App Token
○ 아래 Signing Secret 키를 show 하고 복사하여 저장해두기
○ OAuth & Permissions 클릭
○ Scopes에 Bot Token Scopes에서 4가지의 OAuth Scope를 추가
● chat:write
● channels:history
● im:history
● groups:history
○ OAuth Tokens에 Install to "워크스페이스 이름" 클릭 후 권한 허용
○ 해당 Slack workspace에 챗봇이 추가되었음을 확인
○ Key > SLACK_BOT_TOKEN(Python 코드에서 사용 예정)을 복사하여 저장하기
○ Redirect URLs에 아래의 코드를 입력하여 넣기(꼭 Save URLs 누르기)
https://oauth.pstmn.io/v1/browser-callback
○ Socket Mode에서 Enable Socket Mode 활성화 → Token 이름 입력 → 아래 두 개의 token 추가
○ Token 복사하기(Python 코드에서 사용 예정)
○ (선택) Slack Bot 꾸미기
ㅇ원하는 아이콘 다운로드
○ Basic Information으로 이동 후 아래로 스크롤 → Display Information에 App icon 선택 및 배경 색 선택
○ Slack 확인
○ Event Subscriptions에서 Enable Events 활성화
● 생성한 Slack Bot과 Python Chatbot을 활용해 여러 이벤트 처리 가능
○ 아래의 4가지 기능을 Subscribe to bot events에 아래의 4가지 기능을 추가 후 Save Changes
● message.channels: 해당 App/Bot이 추가된 public channel의 메시지를 Listen
● message.groups: 해당 App/Bot이 추가된 private channel의 메시지를 Listen
● message.im: 해당 App/Bot이 추가된 DM의 메시지를 Listen
● message.mpim: 해당 App/Bot이 추가된 Multi-person DM의 메시지를 Listen
○ reinstall slack bot 클릭 후 허용 클릭
○ Python 프로젝트를 새로 생성 후 가상환경 설치 및 아래 코드를 터미널에서 입력
# 가상환경 설치
py -3.12 -m venv .venv
# 가상환경 접속
.\.venv\Scripts\activate
# install
python -m pip install --upgrade pip
# boto3는 AWS에 올릴 경우 사용(여기서는 로컬에서 env파일로 키-값 관리)
pip install slack_sdk boto3 python-dotenv slack_bolt
○ Chatbot을 적용할 Slack채널의 아이디 확인 및 복사
● 저의 경우 slack-bot에 알람, slack-bot-error에 에러를 넣기 위해 두 개의 링크를 복사
e.g. https://......../C084RFDH59P 의 / 마지막 것만 사용
○ constant.py 생성 후 복사한 채널 링크 넣기(+ 같은 경로에 sns_slack.py, utils.py 추가)
※ 파일 구조 sns/slack/layer/common/constant.py
# constant.py
import enum
class SLACK_TOKENS(enum.Enum):
SLACK_BOT_TOKEN = (enum.auto(), "/sns/slack/.env/SLACK_BOT_TOKEN")
SLACK_APP_TOKEN = (enum.auto(), "/sns/slack/.evn/SLACK_APP_TOKEN")
SLACK_SIGNING_SECRET = (enum.auto(), "/sns/slack/.env/SLACK_SIGNING_SECRET")
class SLACK_CHANNELS(enum.Enum):
ALARM = (enum.auto(), "slack-bot 채널 링크", "알람") # 알람 채널의 아이디가 들어가야함
ERROR = (enum.auto(), "slack-bot-error 채널 링크", "에러") # error 채널의 아이디가 들어가야함
class SERVICE_TYPE(enum.Enum):
TEST = (enum.auto(), "테스트용입니다.")
DEV = (enum.auto(), "개발용입니다.")
# https://api.slack.com/reference/block-kit/blocks#actions_examples
class MESSAGE_BLOCKS(enum.Enum):
SERVICE = (enum.auto(), [
{
"type": "header",
"text": {
"type": "plain_text",
"text": "{service_nm}"
}
},
{
"type": "divider"
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "{service_msg}"
}
}
], "서비스 메세지")
SUB_MSG = (enum.auto(), [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "{service_nm}의 쓰레드 메세지입니다."
},
"accessory": {
"type": "button",
"text": {
"type": "plain_text",
"text": "google 접속"
},
"value": "click_me",
"url": "https://google.com",
"action_id": "button-action"
}
}
], "서비스 메세지의 쓰레드 메세지")
ERROR = (enum.auto(), [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*Error Message*\n{error_msg}"
},
"accessory": {
"type": "button",
"text": {
"type": "plain_text",
"text": "AWS Log"
},
"value": "aws_log_link",
"url": "{aws_log_link_url}",
"action_id": "button-action"
}
}
], "오류 메세지")○ sns_slack.py
import os
import datetime, logging, time, copy
from slack_sdk import WebClient
from slack_sdk.errors import SlackClientError
from .constant import SLACK_CHANNELS, MESSAGE_BLOCKS, SERVICE_TYPE
from .utils import init_alarm
import warnings
warnings.filterwarnings(action='ignore')
class slack_alarm:
def __init__(self, p_slack_channel:SLACK_CHANNELS):
init_alarm()
self.slack_channel = p_slack_channel
self.client = WebClient(token=os.environ.get('SLACK_BOT_TOKEN', None))
self.thread_ts = None
def __send_message(self, p_message_blocks:list[dict], p_thread_ts:str=None) -> dict:
try:
logging.debug(f"[slack_alarm][__send_message] START")
# https://api.slack.com/methods/chat.postMessage
# 해당 채널에 메세지 전달
result = self.client.chat_postMessage(
channel=self.slack_channel.value[1],
blocks=p_message_blocks,
thread_ts=p_thread_ts
)
return result
except SlackClientError as e:
logging.error(f"[slack_alarm][__send_message] Error posting message: {e}")
def get_ts_of_service_message(self, p_service_nm:str) -> str:
logging.debug(f"[slack_alarm][get_ts_of_service_message] START")
if self.thread_ts:
return self.thread_ts
today = time.mktime(datetime.date.today().timetuple())
# 오늘 작성한 message 조회
history = self.client.conversations_history(channel=self.slack_channel.value[1], oldest=today)["messages"]
for msg in history:
try:
if p_service_nm in msg['text']:
self.thread_ts = msg['ts']
break
except KeyError as e:
logging.error(f"[slack_alarm][get_ts_of_service_message] {str(e)}")
continue
return self.thread_ts
def send_service_message(self, p_service_type:SERVICE_TYPE) -> str:
logging.debug(f"[slack_alarm][send_service_message] START")
if not isinstance(p_service_type, SERVICE_TYPE):
logging.error("[slack_alarm][send_service_message] error of p_service_type")
return
elif self.get_ts_of_service_message(p_service_type.name):
return self.thread_ts
message = copy.deepcopy(MESSAGE_BLOCKS.SERVICE.value[1])
message[0]['text']['text'] = message[0]['text']['text'].format(service_nm=p_service_type.name)
message[2]['text']['text'] = message[2]['text']['text'].format(service_msg=p_service_type.value[1])
self.thread_ts = self.__send_message(p_message_blocks=message)['ts']
return self.thread_ts
def send_sub_message(self, p_service_type:SERVICE_TYPE):
if not isinstance(p_service_type, SERVICE_TYPE):
logging.error("[slack_alarm][send_sub_message] error of p_service_type")
return
elif not self.thread_ts:
logging.error("[slack_alarm][send_sub_message] no thread_ts")
return
message = copy.deepcopy(MESSAGE_BLOCKS.SUB_MSG.value[1])
message[0]['text']['text'] = message[0]['text']['text'].format(service_nm=p_service_type.name)
self.thread_ts = self.__send_message(p_message_blocks=message, p_thread_ts=self.thread_ts)['ts']
return self.thread_ts
def send_error_message(self, p_lambda_nm:str, p_error_msg:str):
if not self.thread_ts:
logging.error("[slack_alarm][send_sub_message] no thread_ts")
return
message = copy.deepcopy(MESSAGE_BLOCKS.ERROR.value[1])
message[0]['text']['text'] = message[0]['text']['text'].format(error_msg=p_error_msg)
aws_log_link_url = f"https://ap-northeast-2.console.aws.amazon.com/cloudwatch/home?region=ap-northeast-2#logsV2:log-groups/log-group/$252Faws$252Flambda$252F{p_lambda_nm}"
message[0]['accessory']['url'] = message[0]['accessory']['url'].format(aws_log_link_url=aws_log_link_url)
self.thread_ts = self.__send_message(p_message_blocks=message, p_thread_ts=self.thread_ts)['ts']
return self.thread_ts
○ utils.py
import os
import boto3
from dotenv import load_dotenv
# .env 파일 로드
load_dotenv()
from .constant import SLACK_TOKENS
def __set_environ(p_slack_token:SLACK_TOKENS):
ssm = boto3.client('ssm')
parameter = ssm.get_parameter(Name=p_slack_token.value[1], WithDecryption=True)
# os.environ['key값(환경변수 키)'] = '환경변수 value값'
os.environ[p_slack_token.name] = parameter['Parameter']['Value']
def init_alarm():
SLACK_BOT_TOKEN = os.environ.get('SLACK_BOT_TOKEN', None)
if not SLACK_BOT_TOKEN:
raise EnvironmentError("SLACK_BOT_TOKEN is not set in the .env file or environment variables.")
def init_event():
SLACK_BOT_TOKEN = os.environ.get('SLACK_BOT_TOKEN', None)
SLACK_APP_TOKEN = os.environ.get('SLACK_APP_TOKEN', None)
SLACK_SIGNING_SECRET = os.environ.get('SLACK_SIGNING_SECRET', None)
if not SLACK_BOT_TOKEN or not SLACK_APP_TOKEN or not SLACK_SIGNING_SECRET:
raise EnvironmentError("One or more Slack tokens are not set in the .env file or environment variables.")○ 만든 Chat-Bot에 오른쪽 키 → 앱 세부정보 보기 → 이 앱을 채널에 추가 클릭(원하는 채널 클릭)
○ .env에(sns/.env) 저장한 3가지의 키를 넣기
SLACK_SIGNING_SECRET = Signing_secret 키
SLACK_BOT_TOKEN = slack_bot_token 키
SLACK_APP_TOKEN = 마지막에 복사한 키
○ alarm.py 생성 후 터미널에서 실행 (sns/slack/alarm.py)
py sns/slack/alarm.pyimport os, logging
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from layer.common.utils import init_alarm
from layer.common.constant import SLACK_CHANNELS
# SLACK_BOT_TOKEN을 환경변수에 추가하는 함수
init_alarm()
def main(p_channel_id:str, p_message:str):
# 슬렉서버로 메세지를 전달할 객체(client 변수에 슬랙에 있는 알람 챗봇을 넣기)
client = WebClient(token=os.environ.get('SLACK_BOT_TOKEN', None))
try:
# https://api.slack.com/methods/chat.postMessage
# 해당 채널에 메세지 전달
result = client.chat_postMessage(
channel=p_channel_id,
text=p_message
)
logging.info(result)
except SlackApiError as e:
logging.error(f"Error posting message: {e}")
if __name__ == "__main__":
# 슬렉 채널 아이디
channel_ALARM = SLACK_CHANNELS.ALARM.value[1] # Alarm 채널
channel_ERROR = SLACK_CHANNELS.ERROR.value[1] # Error 채널
# 전달할 메세지
message_ALARM = "Hello Slack-Bot" # Alarm 채널에 보내는 메세지
message_ERROR = "Hello Error" # Error 채널에 보내는 메세지
main(p_channel_id=channel_ALARM, p_message=message_ALARM)
main(p_channel_id=channel_ERROR, p_message=message_ERROR)※ 두 개의 채널에 추가하여 두 개의 메세지가 나옴을 확인
○ event.py 생성 및 실행
※ 정확히 똑같은 것만 입력해야 값이 나오게 됨
● 챗봇이 있는 채널에서 '안녕' 입력 → 안녕 "닉네임"
● 챗봇이 있는 채널에서 '블로그' 입력 → 블로그 주소 출력
● 챗봇이 있는 채널에서 '깃헙' 입력 → 깃허브 주소 출력
● 챗봇이 있는 채널에서 '버튼' 입력 → 안녕~ "닉네임" + 버튼 등장
● 챗봇이 있는 채널에서 등장한 '버튼' 클릭 → '닉네임' Clicked 출력
[결과 화면]
import os
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from layer.common.utils import init_event
init_event()
# Install the Slack app and get xoxb- token in advance
app = App(
token=os.environ.get('SLACK_BOT_TOKEN', None),
signing_secret=os.environ.get('SLACK_SIGNING_SECRET', None)
)
@app.message("안녕")
def message_hello(message, say):
say(f"안녕 <@{message['user']}>!")
@app.message("블로그")
def message_hello(message, say):
say(f"https://joowon582.tistory.com")
@app.message("깃헙")
def message_hello(message, say):
say(f"https://github.com/Leejoowon123")
# Listens to incoming messages that contain "hello"
@app.message("버튼")
def message_hello(message, say):
# say() sends a message to the channel where the event was triggered
say(
# blocks 사용법
# https://api.slack.com/reference/block-kit/blocks
blocks=[
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"안녕~ <@{message['user']}>!"
},
"accessory": {
"type": "button",
"text": {
"type": "plain_text",
"text": "Click me..."
},
"action_id": "button_click" #해당 아이디(button_click)를 누르면 app.action이 실행이 됨
}
}
],
text=f"Hey there <@{message['user']}>!"
)
@app.action("button_click")
def action_button_click(body, ack, say):
# Acknowledge the action
ack()
say(f"<@{body['user']['id']}> Clicked")
if __name__ == "__main__":
SocketModeHandler(app, os.environ.get('SLACK_APP_TOKEN', None)).start()
○ events.py에 있는 블록 기본 틀 사이트
※ Examples에 View this block in Kit Builder 클릭(알아서 조합해서 코드에 복붙하면 끝)
'Networks > Project' 카테고리의 다른 글
| SK networks AI Camp - Final Project(Museify) (1) | 2024.12.04 |
|---|---|
| SK networks AI Camp - mini Project4(Chatbot) (6) | 2024.11.05 |
| SK networks AI Camp - ToyProject(AWS에 이미지 업로드 및 받아오기) (1) | 2024.10.17 |
| SK networks AI Camp - toyproject AWS 및 Github 트러블 이슈 (5) | 2024.10.17 |
| SK networks AI Camp - mini Project2(Active Senior) (0) | 2024.08.19 |