SK networks AI Camp - Slack ChatBot

Networks/Project

SK networks AI Camp - Slack ChatBot

코딩하는 Español되기 2024. 12. 10. 16:00

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.py

import 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

현재글SK networks AI Camp - Slack ChatBot

코딩에서 살아남기

스페인어, 일상, 코딩, SK networks AI Camp 관련 글 올립니다! 많이 봐주세요.

딥러닝, 회고록, AI, MySQL, docker, 데이터, encore, 스페인어, playdata, Crawling, 데이터 전처리, 명령어, db, sk networks, Python, 프로젝트, django, 코딩, 머신러닝, 크롤링,

Today :
Yesterday :

코딩에서 살아남기