111 lines
2.8 KiB
Python
111 lines
2.8 KiB
Python
from typing import Optional
|
|
from typing import List, Dict
|
|
|
|
from slack_sdk.web import WebClient, SlackResponse
|
|
from slack_bolt import BoltContext
|
|
|
|
# ----------------------------
|
|
# General operations in a channel
|
|
# ----------------------------
|
|
|
|
|
|
def find_parent_message(
|
|
client: WebClient, channel_id: Optional[str], thread_ts: Optional[str]
|
|
) -> Optional[dict]:
|
|
if channel_id is None or thread_ts is None:
|
|
return None
|
|
|
|
messages = client.conversations_history(
|
|
channel=channel_id,
|
|
latest=thread_ts,
|
|
limit=1,
|
|
inclusive=1,
|
|
).get("messages", [])
|
|
|
|
return messages[0] if len(messages) > 0 else None
|
|
|
|
|
|
def is_no_mention_thread(context: BoltContext, parent_message: dict) -> bool:
|
|
parent_message_text = parent_message.get("text", "")
|
|
return f"<@{context.bot_user_id}>" in parent_message_text
|
|
|
|
|
|
# ----------------------------
|
|
# WIP reply message stuff
|
|
# ----------------------------
|
|
|
|
|
|
def post_wip_message(
|
|
*,
|
|
client: WebClient,
|
|
channel: str,
|
|
thread_ts: str,
|
|
loading_text: str,
|
|
messages: List[Dict[str, str]],
|
|
user: str,
|
|
) -> SlackResponse:
|
|
system_messages = [msg for msg in messages if msg["role"] == "system"]
|
|
return client.chat_postMessage(
|
|
channel=channel,
|
|
thread_ts=thread_ts,
|
|
text=loading_text,
|
|
metadata={
|
|
"event_type": "chat-gpt-convo",
|
|
"event_payload": {"messages": system_messages, "user": user},
|
|
},
|
|
)
|
|
|
|
|
|
def update_wip_message(
|
|
client: WebClient,
|
|
channel: str,
|
|
ts: str,
|
|
text: str,
|
|
messages: List[Dict[str, str]],
|
|
user: str,
|
|
) -> SlackResponse:
|
|
system_messages = [msg for msg in messages if msg["role"] == "system"]
|
|
return client.chat_update(
|
|
channel=channel,
|
|
ts=ts,
|
|
text=text,
|
|
metadata={
|
|
"event_type": "chat-gpt-convo",
|
|
"event_payload": {"messages": system_messages, "user": user},
|
|
},
|
|
)
|
|
|
|
|
|
# ----------------------------
|
|
# Home tab
|
|
# ----------------------------
|
|
|
|
DEFAULT_HOME_TAB_MESSAGE = (
|
|
"To enable this app in this Slack workspace, you need to save your OpenAI API key. "
|
|
"Visit <https://platform.openai.com/account/api-keys|your developer page> to grap your key!"
|
|
)
|
|
|
|
DEFAULT_HOME_TAB_CONFIGURE_LABEL = "Configure"
|
|
|
|
|
|
def build_home_tab(message: str, configure_label: str) -> dict:
|
|
return {
|
|
"type": "home",
|
|
"blocks": [
|
|
{
|
|
"type": "section",
|
|
"text": {
|
|
"type": "mrkdwn",
|
|
"text": message,
|
|
},
|
|
"accessory": {
|
|
"action_id": "configure",
|
|
"type": "button",
|
|
"text": {"type": "plain_text", "text": configure_label},
|
|
"style": "primary",
|
|
"value": "api_key",
|
|
},
|
|
}
|
|
],
|
|
}
|