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
|
from __future__ import annotations
from typing import Dict
import weechat
from slack.shared import shared
from slack.slack_api import SlackApi
from slack.slack_conversation import SlackConversation
from slack.slack_user import SlackUser
from slack.task import Future, create_task
class SlackUsers(Dict[str, Future[SlackUser]]):
def __init__(self, workspace: SlackWorkspace):
super().__init__()
self.workspace = workspace
def __missing__(self, key: str):
self[key] = create_task(self._create_user(key))
return self[key]
async def _create_user(self, user_id: str) -> SlackUser:
user = SlackUser(self.workspace, user_id)
await user.init()
return user
class SlackWorkspace:
def __init__(self, name: str):
self.name = name
self.config = shared.config.create_workspace_config(self.name)
self.api = SlackApi(self)
self.is_connected = False
self.nick = "TODO"
self.users = SlackUsers(self)
self.conversations: Dict[str, SlackConversation] = {}
async def connect(self):
# rtm_connect = await self.api.fetch("rtm.connect")
user_channels_response = await self.api.fetch_list(
"users.conversations",
"channels",
{
"exclude_archived": True,
# "types": "public_channel,private_channel,im",
"types": "public_channel",
"limit": 1000,
},
-1,
)
user_channels = user_channels_response["channels"]
for channel in user_channels:
conversation = SlackConversation(self, channel["id"])
self.conversations[channel["id"]] = conversation
create_task(conversation.init())
# print(rtm_connect)
# print([c["name"] for c in user_channels])
self.is_connected = True
weechat.bar_item_update("input_text")
|
