[Slack] 연말 결산 이모티콘 순위

연말을 맞이하여, 슬랙에서 수많은 이모티콘 중 어떤 이모티콘이 인기가 있는지 뽑아보려고 한다.

우선 SLACK API 에서 별도 APP 을 생성해줘야한다.

https://api.slack.com/apps

Slack API: Applications | Slack

 

오른쪽 위에 있는 "Create New App"을 클릭한다. 

 

그리고 Scratch 를 선택한다.

 

앱 이름과 워크스페이스를 설정하자.

 

그러면 Basic Information 이 나온다.

 

여기에서 API호출을 위해 필요한 권한을 설정하려면 "OAuth & Permissions"에 들어가서 UserToken Scopes 을 설정해준다.

channels 권한은 공개채널을 위한 것이고,
이모지 순위를 위한 emoji 권한.
비공개 채널을 위한 goups 권한,
그리고 userid로 출력되는걸 사용자 확인을 위한 users:read 를 넣었다.
(여기에서는 쓰이지 않지만, 권한을 추가하면 reinstall을 해야되서 한번에 설정했다.)

그리고 앱을 설정하면 "OAuth Tokens"을 알 수 있다.

 

편의를 위해 Python을 사용했다.

.env에 SLACK_TOKEN을 설정

#SlackToken
SLACK_TOKEN=<YOUR SLACK TOKEN>

Python 코드

import requests
import json
import time
import os
from dotenv import load_dotenv
from concurrent.futures import ThreadPoolExecutor, as_completed

load_dotenv()

# 모든 채널 가져오기
def get_all_channels(headers):
    channels = []
    next_cursor = None

    while True:
        params = {
            "limit": 100,
        }
        if next_cursor:
            params["cursor"] = next_cursor

        r = requests.get("https://slack.com/api/conversations.list", headers=headers, params=params)
        json_response = r.json()

        if not json_response.get("ok"):
            print("API 호출 실패: ", json_response.get("error", "Unknown Error"))
            break

        channels.extend(json_response.get("channels", []))

        next_cursor = json_response.get("response_metadata", {}).get("next_cursor")
        if not next_cursor:
            break

        time.sleep(0.2)  # 딜레이를 줄여도 병렬 처리 시 문제가 없음

    print(f"총 채널 수: {len(channels)}")
    return [channel["id"] for channel in channels]


# 특정 채널의 메시지에서 이모지 추출
def fetch_channel_emoji(channel, headers, start_of_2024, end_of_2024):
    emojis = {}
    next_cursor = None

    while True:
        params = {
            "channel": channel,
            "inclusive": 1,
            "oldest": start_of_2024,
            "latest": end_of_2024,
            "limit": 1000,
        }
        if next_cursor:
            params["cursor"] = next_cursor

        r = requests.get("https://slack.com/api/conversations.history", headers=headers, params=params)
        json_response = r.json()

        if "messages" not in json_response:
            break

        for message in json_response["messages"]:
            if "reactions" in message:
                for reaction in message["reactions"]:
                    if reaction["name"] in emojis:
                        emojis[reaction["name"]] += reaction["count"]
                    else:
                        emojis[reaction["name"]] = reaction["count"]

        next_cursor = json_response.get("response_metadata", {}).get("next_cursor")
        if not next_cursor:
            break

    return emojis


# 병렬 처리로 모든 채널의 이모지 데이터를 가져옴
def count_emoji_parallel(channels, headers):
    # 2024년 시작과 끝에 해당하는 Unix timestamp
    start_of_2024 = int(time.mktime(time.strptime("2024-01-01", "%Y-%m-%d")))
    end_of_2024 = int(time.mktime(time.strptime("2024-12-31 23:59:59", "%Y-%m-%d %H:%M:%S")))

    all_emojis = {}

    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = {executor.submit(fetch_channel_emoji, channel, headers, start_of_2024, end_of_2024): channel for channel in channels}

        for future in as_completed(futures):
            channel_emojis = future.result()
            for emoji, count in channel_emojis.items():
                if emoji in all_emojis:
                    all_emojis[emoji] += count
                else:
                    all_emojis[emoji] = count

    return all_emojis


# 이모지 상위 10개 추출
def sort_top_10(emojis):
    emojis_sorted = sorted(emojis.items(), key=lambda x: x[1], reverse=True)[:10]
    return emojis_sorted


if __name__ == '__main__':
    token = os.getenv("SLACK_TOKEN")

    if token is None:
        print("환경변수에 토큰이 설정되어 있지 않습니다.")
    else:
        print("불러온 토큰 값:", token)
        headers = {
            "Authorization": f"Bearer {token}"
        }

        # 모든 채널 가져오기
        channels = get_all_channels(headers)

        # 병렬 처리로 2024년 이모지 데이터 추출
        emojis = count_emoji_parallel(channels, headers)

        # 이모지 TOP 10 정렬
        emojis_sorted = sort_top_10(emojis)

        # 결과 파일로 출력
        print('결과 파일로 출력 중')
        with open('result_2024_top10.txt', 'w') as f:
            for emoji, count in emojis_sorted:
                f.write(f"{emoji}: {count}\n")
        print('결과 파일로 출력 완료!')

 

처음 개발했을때, 채널 리스트가 전부 나오지 않아서 당황했는데, next_cursor 로 해결.

100개씩 나눠서 돌렸다.

참고 https://api.slack.com/methods/conversations.list

conversations.list API method

 

처음엔 너무 느려서,

concurrent.futures 를 통해 병렬처리 했더니 그나마 쓸만했다.

참고 https://docs.python.org/ko/3/library/concurrent.futures.html

concurrent.futures — Launching parallel tasks

 

text로 출력된 값을 슬랙 에 붙여넣어보면 이모지가 나온다.

:+1: 795
:eyes: 554
:pray: 407
:넵: 366
:white_check_mark: 200
:thumbsup_all: 194
:blob-clap: 149
:yes: 130
:짱: 130
:pepe_baksu: 92