Add contributor guide and project usage docs
This commit is contained in:
108
src/lazy_enter/slack_handler.py
Normal file
108
src/lazy_enter/slack_handler.py
Normal file
@@ -0,0 +1,108 @@
|
||||
"""Slack Socket Mode 이벤트 핸들링."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from slack_bolt import App
|
||||
from slack_bolt.adapter.socket_mode import SocketModeHandler
|
||||
from slack_sdk import WebClient
|
||||
|
||||
from lazy_enter.config import Config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SlackHandler:
|
||||
"""Slack 앱과의 통신을 담당한다."""
|
||||
|
||||
def __init__(self, config: Config) -> None:
|
||||
self.config = config
|
||||
self.app = App(token=config.slack_bot_token)
|
||||
self._handler = SocketModeHandler(self.app, config.slack_app_token)
|
||||
self._on_message_callback: callable | None = None
|
||||
self._on_command_callback: callable | None = None
|
||||
|
||||
self._register_listeners()
|
||||
|
||||
@property
|
||||
def client(self) -> WebClient:
|
||||
return self.app.client
|
||||
|
||||
def _is_authorized(self, user_id: str, channel_id: str) -> bool:
|
||||
"""허가된 사용자·채널인지 확인한다."""
|
||||
if self.config.allowed_user_id and user_id != self.config.allowed_user_id:
|
||||
return False
|
||||
if (
|
||||
self.config.allowed_channel_id
|
||||
and channel_id != self.config.allowed_channel_id
|
||||
):
|
||||
return False
|
||||
return True
|
||||
|
||||
def _register_listeners(self) -> None:
|
||||
"""Slack 이벤트 리스너를 등록한다."""
|
||||
|
||||
@self.app.event("message")
|
||||
def handle_message(event: dict, say: callable) -> None:
|
||||
user_id = event.get("user", "")
|
||||
channel_id = event.get("channel", "")
|
||||
text = event.get("text", "")
|
||||
|
||||
if not self._is_authorized(user_id, channel_id):
|
||||
return
|
||||
|
||||
if self._on_message_callback:
|
||||
self._on_message_callback(text, channel_id)
|
||||
|
||||
@self.app.command("/start-claude")
|
||||
def handle_start(ack: callable, body: dict) -> None:
|
||||
ack()
|
||||
user_id = body.get("user_id", "")
|
||||
channel_id = body.get("channel_id", "")
|
||||
|
||||
if not self._is_authorized(user_id, channel_id):
|
||||
return
|
||||
|
||||
if self._on_command_callback:
|
||||
self._on_command_callback("start", channel_id)
|
||||
|
||||
@self.app.command("/stop-claude")
|
||||
def handle_stop(ack: callable, body: dict) -> None:
|
||||
ack()
|
||||
user_id = body.get("user_id", "")
|
||||
channel_id = body.get("channel_id", "")
|
||||
|
||||
if not self._is_authorized(user_id, channel_id):
|
||||
return
|
||||
|
||||
if self._on_command_callback:
|
||||
self._on_command_callback("stop", channel_id)
|
||||
|
||||
def on_message(self, callback: callable) -> None:
|
||||
"""메시지 수신 콜백을 등록한다."""
|
||||
self._on_message_callback = callback
|
||||
|
||||
def on_command(self, callback: callable) -> None:
|
||||
"""슬래시 커맨드 콜백을 등록한다."""
|
||||
self._on_command_callback = callback
|
||||
|
||||
def send_message(
|
||||
self, channel: str, text: str, thread_ts: str | None = None
|
||||
) -> None:
|
||||
"""슬랙 채널에 메시지를 전송한다."""
|
||||
self.client.chat_postMessage(
|
||||
channel=channel,
|
||||
text=text,
|
||||
thread_ts=thread_ts,
|
||||
)
|
||||
|
||||
def start(self) -> None:
|
||||
"""Socket Mode 핸들러를 시작한다."""
|
||||
logger.info("Slack Socket Mode 시작")
|
||||
self._handler.start()
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Socket Mode 핸들러를 종료한다."""
|
||||
logger.info("Slack Socket Mode 종료")
|
||||
self._handler.close()
|
||||
Reference in New Issue
Block a user