[Slack Bot 만들기] 2. python으로 Slack bot 만들기

Slack app 생성 및 연동을 마무리 했다면, python과 Slack api를 사용해서 우리가 원하는 기능을 구현할 수 있습니다.

python 소스를 구현하기에 앞서 app 생성할 때 같이 생성되었던 token을 미리 로컬에 저장합니다.

token은 Slack api 메소드를 호출할 때 파라미터로 사용해야 필요하기 때문에 json파일로 따로 저장해놓고, python 소스에서 읽어와서 사용하는 방식으로 사용하겠습니다.

먼저 conversations.list 메소드를 이용해서 채널 목록을 조회하겠습니다.

# slack bot token 불러오기
# json으로 저장한 token.json 파일을 읽어와서 token key 조회
parent_path = '/location/of/parent/directory'
slack_token_path = parent_path + '/resource/token.json'
with open(slack_token_path, 'r') as token_json:
    slack_dict = json.load(token_json)

# slack app token
slack_token = slack_dict['token']

# 슬랙 채널 조회하기
# Bot이 활동할 채널의 이름
# channel_name = "til"
channel_name = 'today-i-learned'

# 채널 조회 api method (conversations.list)
# https://api.slack.com/methods/conversations.list 확인
URL = 'https://slack.com/api/conversations.list'

# 파라미터
# required args - token, accepted content type
params = {
    'token': slack_token,
    'Content-Type': 'application/x-www-form-urlencoded'
    }

# API 호출
response = requests.get(URL, params=params)

# 채널 리스트
channel_list = json_normalize(response.json()['channels'])
channel_id = list(channel_list.loc[channel_list['name'] == channel_name, 'id'])[0]

conversations.list 메소드를 통해서 채널 리스트를 조회하고, 해당 리스트에서 우리가 사용하고자 하는 채널의 이름을 통해 해당 채널의 id를 조회합니다.

- channel_list: api를 통해서 조회한 채널의 리스트

- channel_id: 조회한 리스트 중에서 우리가 사용할 채널의 id

channel_id를 사용해 우리가 원하는 채널에 메세지를 전송할 수 있습니다.

# bot이 전송할 메세지 생성
# args의 값에 따라서 메세지 내용 변경
args = sys.argv[0:]
if len(args) != 1 and args[1] == '0':
    message = f'현재시간 {datetime.datetime.now().strftime("%H:%M")}입니다.\n아직 til을 작성하지 않으신 분은 빠르게 작성해주세요.'
else:
    # til 작성한 user들의 id 조회
    user_list = find_users(slack_token, channel_id)

    print(user_list)

    # user_list의 user id를 통해서 user 이름 조회
    user_names, user_cnt = find_names(slack_token, channel_id, user_list)

    message = f'금일 til을 작성한 인원은 {user_names}총 {user_cnt}명 입니다.\n오늘 하루도 수고하셨습니다.'

chat.postMessage 메소드를 이용해서 채널에 메세지를 전송할 건데,

이 메소드를 사용하기 위해서는 파라미터로 전송할 메세지를 넘겨주어야 합니다.

저는 해당 프로그램을 크론에 등록하여서 정해진 시간에 메세지를 보내려고 하는데,

실행되는 시간에 따라 다른 메세지를 전송하려고 합니다.

그래서 sys.argv를 통해 매개변수의 값을 판별하여서 서로 다른 내용의 메세지를 만들었습니다.

매개변수의 값이 0인 경우에는 현재 시간을 조회하여서 다음과 같은 메세지를 전송하도록 했습니다.

"현재시간 HH:MM 입니다.
아직 til을 작성하지 않으신 분은 빠르게 작성해주세요."

매개변수의 값이 없거나 1인 경우에는 오늘 til을 작성한 사람들의 이름을 조회해서 다음과 같이 전송하도록 하였습니다.

"금일 til을 작성한 인원은 aaa, bbb, ccc, 총 3명 입니다.
오늘 하루도 수고하셨습니다."

먼저 til을 작성한 user들의 id를 조회하는 find_users() 함수부터 확인하겠습니다.

이 함수에서는 conversations.history를 통해서 작성된 메세지들을 조회합니다.

def find_users(slack_token: str, channel_id: str) -> list[str]:
    user_list = set([])

    # 대화 히스토리 api
    # https://api.slack.com/methods/conversations.history
    URL = 'https://slack.com/api/conversations.history'

    # oldest 시간 구하기 (현재 시간에서 1day를 빼고 계산)
    oldest = time.mktime((datetime.datetime.now() - datetime.timedelta(days = 1)).timetuple())

    # 파라미터
    params = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'token': slack_token,
        'channel': channel_id,
        'oldest': oldest
    }

    response = requests.get(URL, params=params)
    
    # user list에서 슬랙봇을 제외한 user들의 id를 구한다.
    for message in response.json()['messages']:
        if message['user'] == '{Slack Bot ID}':
            continue
        user_list.add(message['user'])

    return list(user_list)

앞에서 조회한 channel_id를 통해서 해당 채널의 대화들을 조회합니다.

이때 oldest 값을 파라미터에 넘겨주는데, 이 값은 얼마나 오래된 메세지까지 조회할 것인가에 대한 조건입니다.

저는 최근 24시간 이내에 올라온 메세지들을 조회하기 위해서 파이썬의 time 모듈을 이용하여서 현재 시간으로 부터 24시간 이전의 값을 구해서 이를 oldest로 전달해주었습니다.

그렇게 조회된 메세지 리스트에서 slack bot id를 제외한 나머지 메세지의 작성자 id를 리스트에 담아서 반환해 줍니다.

그 다음은 find_names() 함수를 통해서 메세지를 만들때 사용할 사용자의 이름을 조회합니다.

앞에서 조회한 users_list를 파라미터로 넘겨주어서 해당 id에 맞는 사용자의 이름을 반환하도록 해주었습니다.

# til 작성한 사용자들의 이름을 str, 숫자를 int로 반환하는 함수
def find_names(slack_token: str, channel_id: str, user_list: list[str]) -> (str, int):
    user_names, user_cnt = '', 0
    for user_id in user_list:
        user_names += find_user(slack_token, channel_id, user_id) + ', '
        user_cnt += 1
    
    return user_names, user_cnt

user_list로 받은 사용자 id를 find_user() 함수에 넘겨주어서 해당 id에 맞는 사용자의 이름을 반환받습니다.

그리고 사용자의 이름을 담은 리스트와 사용자 카운트를 반환해줍니다.

find_user 함수는 users.info 메소드를 사용하여서 입력받은 user_id의 사용자 정보를 받아옵니다.

response로 받은 결과에서 사용자의 real_name을 반환해줍니다.

# user_id에 해당하는 사용자의 이름을 str로 반환하는 함수
def find_user(slack_token: str, channel_id: str, user_id: str) -> str:
    # 사용자 정보 조회 api
    # https://api.slack.com/methods/users.info
    URL = 'https://slack.com/api/users.info'

    # 파라미터
    params = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'token': slack_token,
        'user': user_id,
    }

    response = requests.get(URL, params=params)
    
    return response.json()['user']['real_name']

위의 과정을 통해서 message가 만들어지면, chat.postMessage 메소드를 사용하여 채널에 메세지를 전송한다.

# 메시지 전송 api method (chat.postMessage)
# https://api.slack.com/methods/chat.postMessage
URL = 'https://slack.com/api/chat.postMessage'

# 파라미터
data = {
    'Context_Type': 'application/x-www-form-urlencoded',
    'token': slack_token,
    'channel': channel_id,
    'text': message
}

# API 호출
response = requests.post(URL, data=data)

print(message)

메세지를 전달하고자 하는 channel_id와 메세지 내용 message를 파라미터로 넘겨주어 메세지를 전달하도록 한다.

이렇게 완성된 코드는 다음과 같다.

import sys, datetime, time
import json
import requests
from pandas import json_normalize

# til 작성한 사용자들을 list로 반환하는 함수
def find_users(slack_token: str, channel_id: str) -> list[str]:
    user_list = set([])

    # 대화 히스토리 api
    # https://api.slack.com/methods/conversations.history
    URL = 'https://slack.com/api/conversations.history'

    # oldest 시간 구하기 (현재 시간에서 1day를 빼고 계산)
    oldest = time.mktime((datetime.datetime.now() - datetime.timedelta(days = 1)).timetuple())

    # 파라미터
    params = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'token': slack_token,
        'channel': channel_id,
        'oldest': oldest
    }

    response = requests.get(URL, params=params)
    
    for message in response.json()['messages']:
        if message['user'] == 'U01NKV5PPCP':
            continue
        user_list.add(message['user'])

    return list(user_list)


# user_id에 해당하는 사용자의 이름을 str로 반환하는 함수
def find_user(slack_token: str, channel_id: str, user_id: str) -> str:
    # 사용자 정보 조회 api
    # https://api.slack.com/methods/users.info
    URL = 'https://slack.com/api/users.info'

    # 파라미터
    params = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'token': slack_token,
        'user': user_id,
    }

    response = requests.get(URL, params=params)

    return response.json()['user']['real_name']


# til 작성한 사용자들의 이름을 str, 숫자를 int로 반환하는 함수
def find_names(slack_token: str, channel_id: str, user_list: list[str]) -> (str, int):
    user_names, user_cnt = '', 0
    for user_id in user_list:
        user_names += find_user(slack_token, channel_id, user_id) + ', '
        user_cnt += 1
    
    return user_names, user_cnt


# slack bot token 불러오기
# json으로 저장한 token.json 파일을 읽어와서 token key 조회
parent_path = '/Users/jeonjaemin/Desktop/git_repos/TIL_ALERT_SLACK_BOT'
slack_token_path = parent_path + '/resource/token.json'
with open(slack_token_path, 'r') as token_json:
    slack_dict = json.load(token_json)

slack_token = slack_dict['token']

# 슬랙 채널 조회하기
# Bot이 활동할 채널의 이름
# channel_name = "til"
channel_name = 'today-i-learned'

# 채널 조회 api method (conversations.list)
# https://api.slack.com/methods/conversations.list 확인
URL = 'https://slack.com/api/conversations.list'

# 파라미터
# required args - token, accepted content type
params = {
    'token': slack_token,
    'Content-Type': 'application/x-www-form-urlencoded'
    }

# API 호출
response = requests.get(URL, params=params)

# 채널 리스트
channel_list = json_normalize(response.json()['channels'])
channel_id = list(channel_list.loc[channel_list['name'] == channel_name, 'id'])[0]

print(channel_name, channel_id)


# bot이 전송할 메세지 생성
# args의 값에 따라서 메세지 내용 변경
args = sys.argv[0:]
if len(args) != 1 and args[1] == '0':
    message = f'현재시간 {datetime.datetime.now().strftime("%H:%M")}입니다.\n아직 til을 작성하지 않으신 분은 빠르게 작성해주세요.'
else:
    # til 작성한 user들의 id 조회
    user_list = find_users(slack_token, channel_id)

    print(user_list)

    # user_list의 user id를 통해서 user 이름 조회
    user_names, user_cnt = find_names(slack_token, channel_id, user_list)

    message = f'금일 til을 작성한 인원은 {user_names}총 {user_cnt}명 입니다.\n오늘 하루도 수고하셨습니다.'

# 메시지 전송 api method (chat.postMessage)
# https://api.slack.com/methods/chat.postMessage
URL = 'https://slack.com/api/chat.postMessage'

# 파라미터
data = {
    'Context_Type': 'application/x-www-form-urlencoded',
    'token': slack_token,
    'channel': channel_id,
    'text': message
}

# API 호출
response = requests.post(URL, data=data)

print(message)

완성된 프로그램을 실제로 실행시켜시키면 다음과 같은 메시지를 슬랙 채널에서 받을 수 있다.

파라미터를 0으로 주어서 프로그램을 실행시켰을때의 결과이다.

파라미터를 1로 주었을 때 결과이다.

설계한대로 파라미터에 따라서 서로 다른 메세지를 출력하는 것을 볼 수 있다.

'기타' 카테고리의 다른 글

[NATS] NATS 개념 정리 (0)	2021.04.18
[MQ] Message Queue란? (0)	2021.04.07
테스트 주도 개발 TDD와 BDD, DDD (0)	2021.03.28
[Slack Bot 만들기] 3. crontab 등록하여 자동실행하기 (0)	2021.02.26
[Slack Bot 만들기] 1. Slack app 생성 및 workspace 연동 (0)	2021.02.24

모르는 게 너무 많다.

[Slack Bot 만들기] 2. python으로 Slack bot 만들기

'기타' 카테고리의 다른 글

티스토리툴바

[Slack Bot 만들기] 2. python으로 Slack bot 만들기

'기타' 카테고리의 다른 글

'기타' Related Articles

티스토리툴바