1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
from __future__ import annotations
import time
from contextlib import contextmanager
from typing import TYPE_CHECKING, List, 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_api.slack_conversations_info import SlackConversationsInfo
from typing_extensions import Literal
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.open_conversations.values():
if conversation.buffer_pointer == buffer_pointer:
return conversation
return None
class SlackConversation:
def __init__(
self,
workspace: SlackWorkspace,
info: SlackConversationsInfo,
):
self.workspace = workspace
self._info = info
self._members: Optional[List[str]] = None
# TODO: buffer_pointer may be accessed by buffer_switch before it's initialized
self.buffer_pointer: 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
@classmethod
async def create(cls, workspace: SlackWorkspace, conversation_id: str):
info_response = await workspace.api.fetch_conversations_info(conversation_id)
return cls(workspace, info_response["channel"])
@property
def _api(self) -> SlackApi:
return self.workspace.api
@property
def id(self) -> str:
return self._info["id"]
async def name(self) -> str:
if "is_im" in self._info and self._info["is_im"] is True:
im_user = await self.workspace.users[self._info["user"]]
return im_user.nick()
elif self._info["is_mpim"] is True:
if self._members is None:
members_response = await self._api.fetch_conversations_members(self)
self._members = members_response["members"]
await self.workspace.users.initialize_items(self._members)
member_users = await gather(
*(self.workspace.users[user_id] for user_id in self._members)
)
return ",".join([user.nick() for user in member_users])
else:
return self._info["name"]
def name_prefix(
self, name_type: Union[Literal["full_name"], Literal["short_name"]]
) -> str:
if "is_im" in self._info and self._info["is_im"] is True:
if name_type == "short_name":
return " "
else:
return ""
elif self._info["is_mpim"]:
if name_type == "short_name":
return "@"
else:
return ""
else:
return "#"
@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 open_if_open(self):
if "is_open" in self._info:
if self._info["is_open"]:
await self.open_buffer()
elif self._info.get("is_member"):
await self.open_buffer()
async def open_buffer(self):
name = await self.name()
full_name = f"{shared.SCRIPT_NAME}.{self.workspace.name}.{self.name_prefix('full_name')}{name}"
short_name = self.name_prefix("short_name") + name
self.buffer_pointer = weechat.buffer_new(
full_name, get_callback_name(self.buffer_input_cb), "", "", ""
)
weechat.buffer_set(self.buffer_pointer, "short_name", short_name)
weechat.buffer_set(
self.buffer_pointer, "localvar_set_nick", self.workspace.my_user.nick()
)
self.workspace.open_conversations[self.id] = self
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
|
