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
|
from __future__ import annotations
from itertools import chain
from typing import TYPE_CHECKING, Dict, Mapping, Tuple
from slack.slack_buffer import SlackBuffer
from slack.slack_message import SlackMessage, SlackTs
from slack.task import gather
if TYPE_CHECKING:
from typing_extensions import Literal
from slack.slack_workspace import SlackWorkspace
class SlackThread(SlackBuffer):
def __init__(self, parent: SlackMessage) -> None:
super().__init__()
self.parent = parent
@property
def workspace(self) -> SlackWorkspace:
return self.parent.workspace
@property
def context(self) -> Literal["conversation", "thread"]:
return "thread"
@property
def messages(self) -> Mapping[SlackTs, SlackMessage]:
return self.parent.replies
async def get_name_and_buffer_props(self) -> Tuple[str, Dict[str, str]]:
conversation_name = await self.parent.conversation.name_with_prefix("full_name")
name = f"{conversation_name}.${self.parent.hash}"
short_name = f" ${self.parent.hash}"
return name, {
"short_name": short_name,
"title": "topic",
"input_multiline": "1",
"localvar_set_type": self.parent.conversation.buffer_type,
"localvar_set_slack_type": "thread",
"localvar_set_nick": self.workspace.my_user.nick(),
"localvar_set_channel": name,
"localvar_set_server": self.workspace.name,
}
async def buffer_switched_to(self):
await self.fill_history()
async def print_history(self):
if self.history_filled:
return
self.history_filled = True
with self.loading():
messages = chain([self.parent], self.parent.replies.values())
for message in messages:
await self.print_message(message, backlog=True)
async def fill_history(self):
if self.history_pending:
return
if self.parent.reply_history_filled:
await self.print_history()
return
with self.loading():
self.history_pending = True
messages = await self.parent.conversation.fetch_replies(self.parent.ts)
sender_user_ids = [m.sender_user_id for m in messages if m.sender_user_id]
self.workspace.users.initialize_items(sender_user_ids)
sender_bot_ids = [
m.sender_bot_id
for m in messages
if m.sender_bot_id and not m.sender_user_id
]
self.workspace.bots.initialize_items(sender_bot_ids)
await gather(*(message.render(self.context) for message in messages))
await self.print_history()
self.history_pending = False
|
