코드로 우주평화

슬랙 봇으로 '채널 초대' 기능 구현하기 본문

나는 이렇게 논다/글또 슬랙 봇 개발기

슬랙 봇으로 '채널 초대' 기능 구현하기

daco2020 2024. 8. 31. 00:31
반응형

앞서 우리는 슬랙 봇에서 사용자가 특정 멤버를 여러 채널에 초대할 수 있도록 모달을 띄우는 기능을 구현했습니다.

 

 

슬랙 봇으로 '채널 초대' 모달 띄우기

이전 글에서 /관리자 명령어를 통해 관리자만 접근할 수 있는 메뉴를 만드는 방법에 대해 설명했었습니다. 2024.08.29 - [나는 이렇게 논다/글또 슬랙 봇 개발기] - 슬랙 봇으로 나만 보이는 '관리자

daco2020.tistory.com

 

 

이번에는 '채널 초대'를 마무리 짓기 위해, 사용자가 모달에서 제출한 정보를 바탕으로 실제로 채널 초대를 수행하는 방법을 설명하겠습니다.

 

 

채널 초대 뷰 핸들링

채널 초대를 처리하는 뷰 이벤트 함수는 handle_invite_channel_view로 지었습니다. 뷰 모달에서 제출된 데이터를 받아서 처리하죠.

 

이 함수의 주요 역할은
1. 제출된 데이터를 확인하고, 필요한 경우 모든 공개 채널의 ID를 가져옵니다.
2. 선택된 채널에 사용자를 초대하고, 각 채널에 대한 결과를 관리 채널에 전송합니다.

 

그럼 핸들러 함수부터 먼저 정의하겠습니다.

app.view("invite_channel_view")
async def handle_invite_channel_view(
    ack: AsyncAck,
    body: ViewBodyType,
    client: AsyncWebClient,
    view: ViewType,
    say: AsyncSay,
    user: User,
    service: SlackService,
) -> None:
    """채널 초대를 수행합니다."""
    await ack()

 

핸들러 함수는 먼저 ack()를 호출해서 슬랙에게 요청을 잘 받았다는 것을 알리는 작업을 수행합니다.

 

 

모달에서 제출된 데이터 처리

모달에서 사용자가 선택한 멤버와 채널 데이터를 body 객체로부터 추출합니다.

    values = body["view"]["state"]["values"]
    user_id = values["user"]["select_user"]["selected_user"]
    channel_ids = values["channel"]["select_channels"]["selected_channels"]

    if not channel_ids:
        channel_ids = await _fetch_public_channel_ids(client)

 

여기서 user_id는 초대할 멤버의 ID를 나타내고, channel_ids는 초대할 채널들의 ID 리스트를 나타내는데요. 만약 사용자가 채널을 하나도 선택하지 않았다면, 이후 _fetch_public_channel_ids 함수를 통해 모든 공개 채널의 ID를 가져와서 사용하게 됩니다.

 

 

공개 채널 ID 조회 함수

`_fetch_public_channel_ids` 함수는 모든 공개 채널의 ID를 조회합니다. 이 과정에서 API 요청에 실패할 가능성을 고려해 tenacity를 사용해 재시도를 설정했습니다. (tenacity는 에러가 발생할 경우 다시 재시도를 해주는 라이브러리입니다. tenacity에 대해서 더 자세히 알고 싶다면 예외가 발생한 함수를 다시 실행하려면? 글을 참고해주세요)

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_fixed(1),
    reraise=True,
)
async def _fetch_public_channel_ids(client: AsyncWebClient) -> list[str]:
    """모든 공개 채널의 아이디를 조회합니다."""
    res = await client.conversations_list(limit=500, types="public_channel")
    return [channel["id"] for channel in res["channels"]]

 

이 함수는 최대 500개의 채널 ID를 가져올 수 있고, 결과는 채널 ID의 리스트로 반환됩니다. 참고로 한 번에 최대 조회 가능한 채널 수는 1000개 입니다.

 

 

채널 초대 작업 수행

다시 `handle_invite_channel_view` 함수로 돌아와 이제 실제로 각 채널에 사용자를 초대하는 작업을 수행해보겠습니다. 먼저 관리 채널에 초대 작업이 시작됨을 알리고, 각 채널에 대해 _invite_channel 함수를 호출해 초대를 진행합니다.

    await client.chat_postMessage(
        channel=settings.ADMIN_CHANNEL,
        text=f"<@{user_id}> 님의 채널 초대를 시작합니다.\n\n채널 수 : {len(channel_ids)} 개\n",
    )

    for channel_id in channel_ids:
        await _invite_channel(client, user_id, channel_id)

    await client.chat_postMessage(
        channel=settings.ADMIN_CHANNEL,
        text="채널 초대가 완료되었습니다.",
    )

 

이제 `_invite_channel` 함수가실제 초대를 실행하고 그 결과 메시지를 관리 채널에 전송하도록 구현해보겠습니다.

 

 

채널 초대 함수

_invite_channel 함수는 실제로 사용자를 지정된 채널에 초대하는 작업을 수행합니다. 이 과정에서 Slack API를 사용하며, 몇 가지 예외 상황도 처리해주겠습니다.

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_fixed(1),
    reraise=True,
)
async def _invite_channel(
    client: AsyncWebClient,
    user_id: str,
    channel_id: str,
) -> None:
    """채널에 멤버를 초대합니다."""
    try:
        await client.conversations_invite(channel=channel_id, users=user_id)
        result = " -> ✅ (채널 초대)"
    except SlackApiError as e:
        # 봇이 채널에 없는 경우, 채널에 참여하고 초대합니다.
        if e.response["error"] == "not_in_channel":
            await client.conversations_join(channel=channel_id)
            await client.conversations_invite(channel=channel_id, users=user_id)
            result = " -> ✅ (봇도 함께 채널 초대)"
        elif e.response["error"] == "already_in_channel":
            result = " -> ✅ (이미 채널에 참여 중)"
        elif e.response["error"] == "cant_invite_self":
            result = " -> ✅ (봇이 자기 자신을 초대)"
        else:
            link = "<https://api.slack.com/methods/conversations.invite#errors|문서 확인하기>"
            result = f" -> 😵 ({e.response['error']}) 👉 {link}"

    await client.chat_postMessage(
        channel=settings.ADMIN_CHANNEL,
        text=f"\n<#{channel_id}>" + result,
    )

 

이 함수는 기본적으로 사용자를 채널에 초대하지만, 만약 봇이 해당 채널에 참여하지 않은 경우 conversations_join 메서드를 사용해 봇을 채널에 참여시킨 후 초대를 진행합니다. 이렇게 하면 봇이 채널에 존재하지 않아 초대를 실패하는 경우가 없어집니다. 또한 이미 채널에 참여하는 경우, 봇이 자기 자신을 초대하는 경우도 성공 메시지를 주도록 처리합니다.

하지만, 만약 개발자가 인지하지 못하는 에러가 발생하는 경우에는 실패 메시지를 반환하겠습니다. 이때 에러에 대한 공식문서를 쉽게 참조할 수 있도록 메시지에 문서 링크를 함께 전송합니다. 이렇게 하면 에러를 발견한 관리자가 문제를 쉽게 파악하고 해결할 수 있을 겁니다.

 

 

결과물

이렇게 해서 관리자가 멤버를 여러 채널에 초대할 수 있는 기능을 완성했습니다. 아래 이미지는 테스트 워크스페이스에서 실행해본 결과입니다. 저는 이미 모든 채널에 들어가 있기 때문에 (이미 채널에 참여 중) 이라고 표시되네요.

 

 

 

채널 초대는 슬랙 봇을 활용한 작업 자동화의 중요한 부분으로, 관리자에게 매우 유용한 기능이 될 수 있습니다. 특히, 봇을 전체 채널에 추가하고 사용하려면 이런 채널 초대 기능이 필수적이죠. 그렇지 않으면 모든 채널에 봇을 일일히 추가해줘야할테니까요.

 

이상으로 '채널 초대' 기능 구현을 마치겠습니다. 다음에도 슬랙 봇으로 재미난 기능을 구현하는 방법에 대해서 다뤄보겠습니다. 😊

반응형