Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions app/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from app.grad_2025.router import private_router as grad_2025_private_router
from app.grad_2025.router import public_router as grad_2025_public_router
from app.posts.router import router as posts_router
from app.slack.router import router as slack_router
from app.tags.router import router as tags_router
from app.users.router import router as users_router

Expand All @@ -34,3 +35,4 @@

router.include_router(public_router)
router.include_router(private_router)
router.include_router(slack_router, prefix="/slack", tags=["Slack"])
39 changes: 39 additions & 0 deletions app/grad_2025/repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,3 +16,42 @@ async def list_univ_majors(self) -> list[Grad2025]:
select(Grad2025).where(Grad2025.review_status == ReviewStatus.APPROVED)
)
return list(result.all())

async def list_all_univ_majors(self) -> list[Grad2025]:
"""전체 학교/전공 목록을 조회합니다 (승인 여부 무관)."""
result = await self.session.scalars(
select(Grad2025).order_by(Grad2025.review_status, Grad2025.name)
)
return list(result.all())

async def find_by_name(self, name: str) -> Grad2025 | None:
"""이름으로 학교/전공을 찾습니다."""
result = await self.session.scalars(
select(Grad2025).where(Grad2025.name == name)
)
return result.first()

async def find_by_id(self, grad_id: int) -> Grad2025 | None:
"""ID로 학교/전공을 찾습니다."""
result = await self.session.scalars(
select(Grad2025).where(Grad2025.id == grad_id)
)
return result.first()

async def approve(self, grad_id: int) -> Grad2025 | None:
"""학교/전공을 승인 처리합니다."""
grad = await self.find_by_id(grad_id)
if grad is None:
return None
grad.review_status = ReviewStatus.APPROVED
await self.session.flush()
return grad

async def update_name(self, grad_id: int, new_name: str) -> Grad2025 | None:
"""학교/전공 이름을 변경합니다."""
grad = await self.find_by_id(grad_id)
if grad is None:
return None
grad.name = new_name
await self.session.flush()
return grad
4 changes: 4 additions & 0 deletions app/slack/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,10 @@

class SlackConfig(BaseConfig):
SLACK_WEBHOOK_URL: str | None = None
SLACK_SIGNING_SECRET: str | None = None # 슬랙 요청 검증용
SLACK_ALLOWED_USER_IDS: list[
str
] = [] # 허용된 사용자 ID 목록 (빈 리스트면 모두 허용)


slack_settings = SlackConfig.create()
132 changes: 132 additions & 0 deletions app/slack/handlers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
from app.grad_2025.enums import ReviewStatus
from app.grad_2025.repository import Grad2025Repository
from app.utils.dependency import dependency


@dependency
class GradCommandHandler:
"""슬랙 /grad 슬래시 커맨드 처리 핸들러"""

grad_repository: Grad2025Repository

async def handle_command(self, text: str) -> str:
"""커맨드 텍스트를 파싱하고 적절한 핸들러를 호출합니다."""
parts = text.strip().split(maxsplit=1)
if not parts:
return self._help_message()

action = parts[0].lower()
args = parts[1] if len(parts) > 1 else ""

match action:
case "list":
return await self.handle_list()
case "status":
return await self.handle_status(args)
case "approve":
return await self.handle_approve(args)
case "rename":
return await self.handle_rename(args)
case "help":
return self._help_message()
case _:
return f"❌ 알 수 없는 커맨드: `{action}`\n\n{self._help_message()}"

async def handle_list(self) -> str:
"""전체 학과 목록과 승인 여부를 반환합니다."""
grads = await self.grad_repository.list_all_univ_majors()

if not grads:
return "📚 등록된 학과가 없습니다."

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"]

for grad in grads:
if grad.review_status == ReviewStatus.APPROVED:
lines.append(f"✅ {grad.name}")
else:
lines.append(f"⏳ {grad.name} _(대기중)_")

return "\n".join(lines)

async def handle_status(self, name: str) -> str:
"""특정 학과의 승인 여부를 반환합니다."""
if not name:
return "❌ 학과명을 입력해주세요.\n사용법: `/grad status 홍익대학교 시각디자인`"

grad = await self.grad_repository.find_by_name(name)

if not grad:
return f"❌ `{name}` 학과를 찾을 수 없습니다."

if grad.review_status == ReviewStatus.APPROVED:
return 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)

if not grad:
return f"❌ `{name}` 학과를 찾을 수 없습니다."

if grad.review_status == ReviewStatus.APPROVED:
return 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` - 이 도움말 보기"
)
48 changes: 48 additions & 0 deletions app/slack/router.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
from typing import Annotated

from fastapi import Form, Header, HTTPException, Request
from fastapi.responses import PlainTextResponse

from app.core.router import create_router

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(
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:
"""
/grad 슬래시 커맨드를 처리합니다.

사용 가능한 커맨드:
- /grad list - 전체 학과 목록과 승인 여부
- /grad status 학과명 - 특정 학과의 승인 여부
- /grad approve 학과명 - 학과 승인 처리
- /grad rename 기존이름 → 새이름 - 학과 이름 변경
- /grad help - 도움말
"""
# 슬랙 서명 검증
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,
):
raise HTTPException(status_code=401, detail="Invalid Slack signature")

# 사용자 권한 확인
if not is_user_allowed(user_id):
return "❌ 이 커맨드를 실행할 권한이 없습니다."

# 커맨드 처리
return await handler.handle_command(text)
42 changes: 42 additions & 0 deletions app/slack/service.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
import hashlib
import hmac
import time

import httpx

from app.core.config import Environment, core_settings
Expand Down Expand Up @@ -35,3 +39,41 @@ async def notify_new_univ_major(self, univ_major: str) -> None:
phase = core_settings.ENVIRONMENT.value.upper()
message = f"[{phase}] 🎓 새로운 학교/전공 검수 요청: *{univ_major}*"
await self._send_message(message)


def verify_slack_signature(
timestamp: str,
body: bytes,
signature: str,
) -> bool:
"""슬랙 요청의 서명을 검증합니다."""
signing_secret = slack_settings.SLACK_SIGNING_SECRET
if not signing_secret:
# 로컬 개발 환경에서는 서명 검증 스킵
return True

# 요청이 5분 이상 된 경우 거부 (replay attack 방지)
if abs(time.time() - int(timestamp)) > 60 * 5:
return False

# 서명 생성
sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}"
my_signature = (
"v0="
+ hmac.new(
signing_secret.encode(),
sig_basestring.encode(),
hashlib.sha256,
).hexdigest()
)

return hmac.compare_digest(my_signature, signature)


def is_user_allowed(user_id: str) -> bool:
"""사용자가 커맨드 실행 권한이 있는지 확인합니다."""
allowed_users = slack_settings.SLACK_ALLOWED_USER_IDS
# 허용 목록이 비어있으면 모두 허용
if not allowed_users:
return True
return user_id in allowed_users
Loading