from __future__ import annotations
import time
from contextlib import contextmanager
from typing import TYPE_CHECKING, List, Literal, Optional, Union
import weechat
from slack.shared import shared
from slack.slack_message import SlackMessage
from slack.task import gather
from slack.util import get_callback_name
if TYPE_CHECKING:
from slack.slack_api import SlackApi
from slack.slack_workspace import SlackWorkspace
def get_conversation_from_buffer_pointer(
buffer_pointer: str,
) -> Optional[SlackConversation]:
for workspace in shared.workspaces.values():
for conversation in workspace.conversations.values():
if conversation.buffer_pointer == buffer_pointer:
return conversation
return None
class SlackConversation:
def __init__(self, workspace: SlackWorkspace, id: str):
self.workspace = workspace
self.id = id
# TODO: buffer_pointer may be accessed by buffer_switch before it's initialized
self.buffer_pointer: str = ""
self.name: str
self.is_loading = False
self.history_filled = False
self.history_pending = False
self.completion_context: Union[
Literal["NO_COMPLETION"],
Literal["PENDING_COMPLETION"],
Literal["ACTIVE_COMPLETION"],
Literal["IN_PROGRESS_COMPLETION"],
] = "NO_COMPLETION"
self.completion_values: List[str] = []
self.completion_index = 0
@property
def _api(self) -> SlackApi:
return self.workspace.api
@contextmanager
def loading(self):
self.is_loading = True
weechat.bar_item_update("input_text")
try:
yield
finally:
self.is_loading = False
weechat.bar_item_update("input_text")
@contextmanager
def completing(self):
self.completion_context = "IN_PROGRESS_COMPLETION"
try:
yield
finally:
self.completion_context = "ACTIVE_COMPLETION"
async def init(self):
with self.loading():
info = await self._api.fetch_conversations_info(self)
info_channel = info["channel"]
if info_channel["is_im"] is True:
self.name = "IM" # TODO
elif info_channel["is_mpim"] is True:
self.name = "MPIM" # TODO
else:
self.name = info_channel["name"]
full_name = f"{shared.SCRIPT_NAME}.{self.workspace.name}.{self.name}"
self.buffer_pointer = weechat.buffer_new(
full_name, get_callback_name(self.buffer_input_cb), "", "", ""
)
weechat.buffer_set(self.buffer_pointer, "short_name", self.name)
weechat.buffer_set(
self.buffer_pointer, "localvar_set_nick", self.workspace.my_user.nick()
)
async def fill_history(self):
if self.history_filled or self.history_pending:
return
with self.loading():
self.history_pending = True
history = await self._api.fetch_conversations_history(self)
start = time.time()
messages = [SlackMessage(self, message) for message in history["messages"]]
sender_user_ids = [m.sender_user_id for m in messages if m.sender_user_id]
await self.workspace.users.initialize_items(sender_user_ids)
messages_rendered = await gather(
*(message.render_message() for message in messages)
)
for rendered in reversed(messages_rendered):
weechat.prnt(self.buffer_pointer, rendered)
print(f"history w/o fetch took: {time.time() - start}")
self.history_filled = True
self.history_pending = False
def buffer_input_cb(self, data: str, buffer: str, input_data: str) -> int:
weechat.prnt(buffer, "Text: %s" % input_data)
return weechat.WEECHAT_RC_OK
