Add slack example
This commit is contained in:
110
examples/slack/app/slack_ops.py
Normal file
110
examples/slack/app/slack_ops.py
Normal file
@@ -0,0 +1,110 @@
|
||||
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",
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
Reference in New Issue
Block a user