slack_api.py•12.7 kB
"""
Slack API 클라이언트 구현
"""
import os
from typing import Any, Dict, List, Optional
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from dotenv import load_dotenv
# 환경 변수 로드
load_dotenv()
class SlackClient:
"""Slack API 클라이언트 클래스"""
def __init__(self, token: Optional[str] = None):
"""
Slack 클라이언트 초기화
Args:
token: Slack Bot Token. None일 경우 환경변수에서 가져옴
"""
self.token = token or os.getenv("SLACK_BOT_TOKEN")
if not self.token:
raise ValueError("SLACK_BOT_TOKEN이 필요합니다. 환경변수나 매개변수로 제공해주세요.")
self.client = WebClient(token=self.token)
def send_message(self, channel: str, text: str) -> Dict[str, Any]:
"""
채널에 메시지 전송
Args:
channel: 채널 ID
text: 테스트 test
Returns:
Dict containing success status and message details
"""
try:
# 채널명 정규화
if not channel.startswith('#') and not channel.startswith('C'):
channel = f"#{channel}"
response = self.client.chat_postMessage(
channel=channel,
text=text,
as_user=True
)
return {
"success": True,
"message": "메시지가 성공적으로 전송되었습니다.",
"channel": response["channel"],
"timestamp": response["ts"],
"text": text
}
except SlackApiError as e:
return {
"success": False,
"error": f"Slack API 오류: {e.response['error']}",
"message": "메시지 전송에 실패했습니다.",
"details": str(e)
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "예상치 못한 오류가 발생했습니다."
}
def get_channels(self) -> Dict[str, Any]:
"""
접근 가능한 모든 채널 목록 조회
Returns:
Dict containing list of channels with their details
"""
try:
channels = []
# 공개 채널 조회
public_response = self.client.conversations_list(
types="public_channel",
exclude_archived=True
)
# 비공개 채널 조회
private_response = self.client.conversations_list(
types="private_channel",
exclude_archived=True
)
# 공개 채널 처리
for channel in public_response["channels"]:
channels.append({
"id": channel["id"],
"name": channel["name"],
"is_private": False,
"is_member": channel.get("is_member", False),
"member_count": channel.get("num_members", 0),
"purpose": channel.get("purpose", {}).get("value", ""),
"topic": channel.get("topic", {}).get("value", "")
})
# 비공개 채널 처리
for channel in private_response["channels"]:
channels.append({
"id": channel["id"],
"name": channel["name"],
"is_private": True,
"is_member": channel.get("is_member", False),
"member_count": channel.get("num_members", 0),
"purpose": channel.get("purpose", {}).get("value", ""),
"topic": channel.get("topic", {}).get("value", "")
})
return {
"success": True,
"channels": channels,
"total_count": len(channels),
"message": f"총 {len(channels)}개의 채널을 조회했습니다."
}
except SlackApiError as e:
return {
"success": False,
"error": f"Slack API 오류: {e.response['error']}",
"message": "채널 목록 조회에 실패했습니다."
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "예상치 못한 오류가 발생했습니다."
}
def get_channel_history(self, channel_id: str, limit: int = 10) -> Dict[str, Any]:
"""
채널의 메시지 히스토리 조회
Args:
channel_id: 채널 ID
limit: 10
Returns:
Dict containing message history
"""
try:
# limit 값 검증
if limit <= 0:
limit = 10
elif limit > 100:
limit = 100
response = self.client.conversations_history(
channel=channel_id,
limit=limit
)
messages = []
for message in response["messages"]:
# 사용자 정보 조회
user_info = {"name": "Unknown", "real_name": "Unknown"}
if "user" in message:
try:
user_response = self.client.users_info(user=message["user"])
user_info = {
"name": user_response["user"].get("name", "Unknown"),
"real_name": user_response["user"].get("real_name", "Unknown")
}
except:
pass
messages.append({
"text": message.get("text", ""),
"user": user_info,
"timestamp": message.get("ts", ""),
"type": message.get("type", "message"),
"subtype": message.get("subtype", "")
})
return {
"success": True,
"messages": messages,
"channel_id": channel_id,
"message_count": len(messages),
"message": f"총 {len(messages)}개의 메시지를 조회했습니다."
}
except SlackApiError as e:
return {
"success": False,
"error": f"Slack API 오류: {e.response['error']}",
"message": "메시지 히스토리 조회에 실패했습니다."
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "예상치 못한 오류가 발생했습니다."
}
def send_direct_message(self, user_id: str, text: str) -> Dict[str, Any]:
"""
특정 사용자에게 다이렉트 메시지 전송
Args:
user_id: U08R03K4C5U
text: 안녕
Returns:
Dict containing success status and message details
"""
try:
# DM 채널 열기
dm_response = self.client.conversations_open(users=[user_id])
dm_channel = dm_response["channel"]["id"]
# 다이렉트 메시지 전송
response = self.client.chat_postMessage(
channel=dm_channel,
text=text,
as_user=True
)
return {
"success": True,
"message": "메시지가 성공적으로 전송되었습니다.",
"user_id": user_id,
"channel": dm_channel,
"timestamp": response["ts"],
"text": text
}
except SlackApiError as e:
return {
"success": False,
"error": f"Slack API 오류: {e.response['error']}",
"message": "메시지 전송에 실패했습니다."
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "오류가 발생했습니다."
}
def get_users(self) -> Dict[str, Any]:
"""
워크스페이스의 사용자 목록 조회
Returns:
Dict containing list of users
"""
try:
response = self.client.users_list()
users = []
for user in response["members"]:
if not user.get("deleted", False) and not user.get("is_bot", False):
users.append({
"id": user["id"],
"name": user.get("name", ""),
"real_name": user.get("real_name", ""),
"display_name": user.get("profile", {}).get("display_name", ""),
"email": user.get("profile", {}).get("email", ""),
"is_admin": user.get("is_admin", False),
"is_owner": user.get("is_owner", False)
})
return {
"success": True,
"users": users,
"total_count": len(users),
"message": f"총 {len(users)}명의 사용자를 조회했습니다."
}
except SlackApiError as e:
return {
"success": False,
"error": f"Slack API 오류: {e.response['error']}",
"message": "사용자 목록 조회에 실패했습니다."
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "오류가 발생했습니다."
}
def delete_message(self, channel: str, timestamp: str) -> Dict[str, Any]:
"""
메시지 삭제
Args:
channel: 5_강의장-python
timestamp: 삭제할 메시지의 타임스탬프
Returns:
Dict containing success status
"""
try:
response = self.client.chat_delete(
channel=channel,
ts=timestamp
)
return {
"success": True,
"message": "메시지가 성공적으로 삭제되었습니다.",
"channel": channel,
"timestamp": timestamp
}
except SlackApiError as e:
return {
"success": False,
"error": f"Slack API 오류: {e.response['error']}",
"message": "메시지 삭제에 실패했습니다.",
"details": str(e)
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "오류가 발생했습니다."
}
def update_message(self, channel: str, timestamp: str, text: str) -> Dict[str, Any]:
"""
메시지 업데이트
Args:
channel: 채널 ID 또는 이름
timestamp: 업데이트할 메시지의 타임스탬프
text: 새로운 메시지 내용
Returns:
Dict containing success status
"""
try:
response = self.client.chat_update(
channel=channel,
ts=timestamp,
text=text
)
return {
"success": True,
"message": "메시지가 성공적으로 업데이트되었습니다.",
"channel": channel,
"timestamp": timestamp,
"new_text": text
}
except SlackApiError as e:
return {
"success": False,
"error": f"Slack API 오류: {e.response['error']}",
"message": "메시지 업데이트에 실패했습니다.",
"details": str(e)
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "예상치 못한 오류가 발생했습니다."
}