diff --git a/app/slack/config.py b/app/slack/config.py index 17dcda3..028168d 100644 --- a/app/slack/config.py +++ b/app/slack/config.py @@ -4,6 +4,7 @@ class SlackConfig(BaseConfig): SLACK_WEBHOOK_URL: str | None = None SLACK_SIGNING_SECRET: str | None = None # 슬랙 요청 검증용 + SLACK_BOT_TOKEN: str | None = None # Bot User OAuth Token (xoxb-...) SLACK_ALLOWED_USER_IDS: list[ str ] = [] # 허용된 사용자 ID 목록 (빈 리스트면 모두 허용) diff --git a/app/slack/handlers.py b/app/slack/handlers.py index f31c95d..3f6cf83 100644 --- a/app/slack/handlers.py +++ b/app/slack/handlers.py @@ -1,3 +1,5 @@ +from typing import Any + from app.grad_2025.enums import ReviewStatus from app.grad_2025.repository import Grad2025Repository from app.utils.dependency import dependency @@ -5,128 +7,203 @@ @dependency class GradCommandHandler: - """슬랙 /grad 슬래시 커맨드 처리 핸들러""" + """슬랙 봇 멘션 및 버튼 처리 핸들러""" grad_repository: Grad2025Repository - async def handle_command(self, text: str) -> str: - """커맨드 텍스트를 파싱하고 적절한 핸들러를 호출합니다.""" - parts = text.strip().split(maxsplit=1) + async def handle_mention(self, text: str) -> dict[str, Any]: + """ + 봇 멘션 텍스트를 파싱하고 적절한 핸들러를 호출합니다. + Block Kit 형식의 메시지를 반환합니다. + """ + # 봇 멘션 제거 후 텍스트 파싱 + # 예: "<@U123ABC> list" -> "list" + parts = text.strip().split() + # 첫 번째가 멘션이면 제거 + if parts and parts[0].startswith("<@"): + parts = parts[1:] + if not parts: - return self._help_message() + return self._help_blocks() action = parts[0].lower() - args = parts[1] if len(parts) > 1 else "" + args = " ".join(parts[1:]) if len(parts) > 1 else "" match action: case "list": - return await self.handle_list() + return await self.handle_list_blocks() + case "pending": + return await self.handle_pending_blocks() case "status": - return await self.handle_status(args) - case "approve": - return await self.handle_approve(args) - case "rename": - return await self.handle_rename(args) + return await self.handle_status_blocks(args) case "help": - return self._help_message() + return self._help_blocks() case _: - return f"❌ 알 수 없는 커맨드: `{action}`\n\n{self._help_message()}" + return { + "text": f"❌ 알 수 없는 커맨드: `{action}`", + "blocks": self._help_blocks()["blocks"], + } - async def handle_list(self) -> str: - """전체 학과 목록과 승인 여부를 반환합니다.""" + async def handle_list_blocks(self) -> dict[str, Any]: + """전체 학과 목록과 승인 여부를 Block Kit 형식으로 반환합니다.""" grads = await self.grad_repository.list_all_univ_majors() if not grads: - return "📚 등록된 학과가 없습니다." + return {"text": "📚 등록된 학과가 없습니다."} approved_count = sum( 1 for g in grads if g.review_status == ReviewStatus.APPROVED ) total_count = len(grads) - lines = [f"📚 *학과 목록* ({approved_count}/{total_count} 승인됨)\n"] + blocks: list[dict[str, Any]] = [ + { + "type": "header", + "text": { + "type": "plain_text", + "text": f"📚 학과 목록 ({approved_count}/{total_count} 승인됨)", + }, + }, + {"type": "divider"}, + ] for grad in grads: if grad.review_status == ReviewStatus.APPROVED: - lines.append(f"✅ {grad.name}") + blocks.append( + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": f"✅ *{grad.name}*", + }, + } + ) else: - lines.append(f"⏳ {grad.name} _(대기중)_") + blocks.append( + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": f"⏳ *{grad.name}* _(대기중)_", + }, + "accessory": { + "type": "button", + "text": {"type": "plain_text", "text": "승인"}, + "style": "primary", + "action_id": f"approve_grad_{grad.id}", + "value": str(grad.id), + }, + } + ) + + return {"text": "학과 목록", "blocks": blocks} + + async def handle_pending_blocks(self) -> dict[str, Any]: + """대기중인 학과만 버튼과 함께 반환합니다.""" + grads = await self.grad_repository.list_all_univ_majors() + pending = [g for g in grads if g.review_status == ReviewStatus.PENDING] + + if not pending: + return {"text": "✅ 대기중인 학과가 없습니다. 모두 승인 완료!"} + + blocks: list[dict[str, Any]] = [ + { + "type": "header", + "text": { + "type": "plain_text", + "text": f"⏳ 승인 대기중 ({len(pending)}개)", + }, + }, + {"type": "divider"}, + ] + + for grad in pending: + blocks.append( + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": f"*{grad.name}*", + }, + "accessory": { + "type": "button", + "text": {"type": "plain_text", "text": "승인"}, + "style": "primary", + "action_id": f"approve_grad_{grad.id}", + "value": str(grad.id), + }, + } + ) - return "\n".join(lines) + return {"text": "승인 대기중 학과", "blocks": blocks} - async def handle_status(self, name: str) -> str: + async def handle_status_blocks(self, name: str) -> dict[str, Any]: """특정 학과의 승인 여부를 반환합니다.""" if not name: - return "❌ 학과명을 입력해주세요.\n사용법: `/grad status 홍익대학교 시각디자인`" + return { + "text": "❌ 학과명을 입력해주세요.\n사용법: `@bot status 홍익대학교 시각디자인`" + } grad = await self.grad_repository.find_by_name(name) if not grad: - return f"❌ `{name}` 학과를 찾을 수 없습니다." + return {"text": f"❌ `{name}` 학과를 찾을 수 없습니다."} if grad.review_status == ReviewStatus.APPROVED: - return f"✅ *{grad.name}*\n상태: 승인됨" + return {"text": f"✅ *{grad.name}*\n상태: 승인됨"} else: - return f"⏳ *{grad.name}*\n상태: 대기중 (ID: {grad.id})" - - async def handle_approve(self, name: str) -> str: - """특정 학과를 승인 처리합니다.""" - if not name: - return "❌ 학과명을 입력해주세요.\n사용법: `/grad approve 홍익대학교 시각디자인`" - - grad = await self.grad_repository.find_by_name(name) + blocks: list[dict[str, Any]] = [ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": f"⏳ *{grad.name}*\n상태: 대기중", + }, + "accessory": { + "type": "button", + "text": {"type": "plain_text", "text": "승인"}, + "style": "primary", + "action_id": f"approve_grad_{grad.id}", + "value": str(grad.id), + }, + } + ] + return {"text": f"{grad.name} 상태", "blocks": blocks} + + async def handle_approve_button(self, grad_id: int, user_id: str) -> dict[str, Any]: + """버튼 클릭으로 학과를 승인 처리합니다.""" + grad = await self.grad_repository.find_by_id(grad_id) if not grad: - return f"❌ `{name}` 학과를 찾을 수 없습니다." + return {"text": "❌ 학과를 찾을 수 없습니다."} if grad.review_status == ReviewStatus.APPROVED: - return f"ℹ️ *{grad.name}*은(는) 이미 승인된 상태입니다." + return {"text": f"ℹ️ *{grad.name}*은(는) 이미 승인된 상태입니다."} await self.grad_repository.approve(grad.id) - return f"✅ *{grad.name}*을(를) 승인했습니다!" - - async def handle_rename(self, args: str) -> str: - """학과 이름을 변경합니다.""" - # "기존이름 → 새이름" 또는 "기존이름 -> 새이름" 형식 파싱 - separators = ["→", "->", "=>"] - old_name = None - new_name = None - - for sep in separators: - if sep in args: - parts = args.split(sep, 1) - if len(parts) == 2: - old_name = parts[0].strip() - new_name = parts[1].strip() - break - - if not old_name or not new_name: - return ( - "❌ 올바른 형식으로 입력해주세요.\n" - "사용법: `/grad rename 기존이름 → 새이름`\n" - "예시: `/grad rename 홍익대학교 시각디자인 → 홍익대 시디`" - ) - - grad = await self.grad_repository.find_by_name(old_name) - - if not grad: - return f"❌ `{old_name}` 학과를 찾을 수 없습니다." - - # 새 이름이 이미 존재하는지 확인 - existing = await self.grad_repository.find_by_name(new_name) - if existing: - return f"❌ `{new_name}` 이름은 이미 사용 중입니다." - - await self.grad_repository.update_name(grad.id, new_name) - return f"✏️ 학과 이름을 변경했습니다.\n`{old_name}` → `{new_name}`" - - def _help_message(self) -> str: - """도움말 메시지를 반환합니다.""" - return ( - "📖 *사용 가능한 커맨드*\n\n" - "• `/grad list` - 전체 학과 목록과 승인 여부\n" - "• `/grad status 학과명` - 특정 학과의 승인 여부\n" - "• `/grad approve 학과명` - 학과 승인 처리\n" - "• `/grad rename 기존이름 → 새이름` - 학과 이름 변경\n" - "• `/grad help` - 이 도움말 보기" - ) + return {"text": f"✅ *{grad.name}*을(를) <@{user_id}>님이 승인했습니다!"} + + def _help_blocks(self) -> dict[str, Any]: + """도움말 메시지를 Block Kit 형식으로 반환합니다.""" + return { + "text": "사용 가능한 커맨드", + "blocks": [ + { + "type": "header", + "text": {"type": "plain_text", "text": "📖 사용 가능한 커맨드"}, + }, + {"type": "divider"}, + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": ( + "• `@bot list` - 전체 학과 목록과 승인 여부\n" + "• `@bot pending` - 대기중 학과 (승인 버튼 포함)\n" + "• `@bot status 학과명` - 특정 학과의 승인 여부\n" + "• `@bot help` - 이 도움말 보기" + ), + }, + }, + ], + } diff --git a/app/slack/router.py b/app/slack/router.py index ba0b483..987daa2 100644 --- a/app/slack/router.py +++ b/app/slack/router.py @@ -1,48 +1,181 @@ -from typing import Annotated +import json +from typing import Any +from urllib.parse import parse_qs -from fastapi import Form, Header, HTTPException, Request -from fastapi.responses import PlainTextResponse +import httpx +from fastapi import BackgroundTasks, HTTPException, Request +from fastapi.responses import JSONResponse, Response from app.core.router import create_router +from .config import slack_settings from .handlers import GradCommandHandler from .service import is_user_allowed, verify_slack_signature router = create_router() -@router.post("/commands/grad", response_class=PlainTextResponse) -async def handle_grad_command( +async def send_slack_message( + channel: str, + message: dict[str, Any], + thread_ts: str | None = None, +) -> None: + """Slack API를 통해 메시지를 전송합니다.""" + bot_token = slack_settings.SLACK_BOT_TOKEN + if not bot_token: + return + + payload: dict[str, Any] = { + "channel": channel, + **message, + } + if thread_ts: + payload["thread_ts"] = thread_ts + + try: + async with httpx.AsyncClient() as client: + await client.post( + "https://slack.com/api/chat.postMessage", + headers={"Authorization": f"Bearer {bot_token}"}, + json=payload, + timeout=5.0, + ) + except Exception: + pass + + +async def update_slack_message(response_url: str, message: dict[str, Any]) -> None: + """response_url로 슬랙 메시지를 업데이트합니다.""" + try: + async with httpx.AsyncClient() as client: + await client.post( + response_url, + json={ + **message, + "replace_original": True, + }, + timeout=5.0, + ) + except Exception: + pass + + +# ============================================================ +# Event API - 봇 멘션 처리 +# ============================================================ + + +@router.post("/events") +async def handle_slack_events( + request: Request, + handler: GradCommandHandler, + background_tasks: BackgroundTasks, +) -> Response | JSONResponse: + """ + Slack Event API 엔드포인트. + 봇 멘션(@bot) 이벤트를 처리합니다. + """ + body = await request.body() + + # 서명 검증 + timestamp = request.headers.get("x-slack-request-timestamp") + signature = request.headers.get("x-slack-signature") + if timestamp and signature: + if not verify_slack_signature(timestamp, body, signature): + raise HTTPException(status_code=401, detail="Invalid Slack signature") + + data = json.loads(body) + + # URL Verification (Slack 앱 설정 시 필요) + if data.get("type") == "url_verification": + return JSONResponse({"challenge": data.get("challenge")}) + + # Event 처리 + event = data.get("event", {}) + event_type = event.get("type") + + if event_type == "app_mention": + text = event.get("text", "") + channel = event.get("channel") + thread_ts = event.get("thread_ts") or event.get("ts") + user_id = event.get("user") + + # 권한 확인 + if not is_user_allowed(user_id): + background_tasks.add_task( + send_slack_message, + channel, + {"text": "❌ 이 커맨드를 실행할 권한이 없습니다."}, + thread_ts, + ) + return Response(status_code=200) + + # 핸들러 호출 + message = await handler.handle_mention(text) + + # 메시지 전송 (쓰레드에 응답) + background_tasks.add_task(send_slack_message, channel, message, thread_ts) + + return Response(status_code=200) + + +# ============================================================ +# Interactions - 버튼 클릭 처리 +# ============================================================ + + +@router.post("/interactions") +async def handle_slack_interactions( request: Request, handler: GradCommandHandler, - text: Annotated[str, Form()] = "", - user_id: Annotated[str, Form()] = "", - x_slack_request_timestamp: Annotated[str | None, Header()] = None, - x_slack_signature: Annotated[str | None, Header()] = None, -) -> str: + background_tasks: BackgroundTasks, +) -> Response: """ - /grad 슬래시 커맨드를 처리합니다. - - 사용 가능한 커맨드: - - /grad list - 전체 학과 목록과 승인 여부 - - /grad status 학과명 - 특정 학과의 승인 여부 - - /grad approve 학과명 - 학과 승인 처리 - - /grad rename 기존이름 → 새이름 - 학과 이름 변경 - - /grad help - 도움말 + Slack Interactivity 엔드포인트. + 버튼 클릭 등 인터랙션을 처리합니다. """ - # 슬랙 서명 검증 - if x_slack_request_timestamp and x_slack_signature: - body = await request.body() - if not verify_slack_signature( - x_slack_request_timestamp, - body, - x_slack_signature, - ): + body = await request.body() + + # 서명 검증 + timestamp = request.headers.get("x-slack-request-timestamp") + signature = request.headers.get("x-slack-signature") + if timestamp and signature: + if not verify_slack_signature(timestamp, body, signature): raise HTTPException(status_code=401, detail="Invalid Slack signature") - # 사용자 권한 확인 - if not is_user_allowed(user_id): - return "❌ 이 커맨드를 실행할 권한이 없습니다." + # payload는 URL-encoded form data로 옴 + form_data = parse_qs(body.decode()) + payload_str = form_data.get("payload", ["{}"])[0] + payload = json.loads(payload_str) + + # 버튼 클릭 처리 + if payload.get("type") == "block_actions": + actions = payload.get("actions", []) + user_id = payload.get("user", {}).get("id", "") + response_url = payload.get("response_url") + + # 권한 확인 + if not is_user_allowed(user_id): + if response_url: + background_tasks.add_task( + update_slack_message, + response_url, + {"text": "❌ 이 작업을 실행할 권한이 없습니다."}, + ) + return Response(status_code=200) + + for action in actions: + action_id = action.get("action_id", "") + + # 학과 승인 버튼 + if action_id.startswith("approve_grad_"): + grad_id = int(action.get("value", 0)) + message = await handler.handle_approve_button(grad_id, user_id) + + # 응답 전송 + if response_url: + background_tasks.add_task( + update_slack_message, response_url, message + ) - # 커맨드 처리 - return await handler.handle_command(text) + return Response(status_code=200)