from __future__ import annotations
import json
import socket
import ssl
import time
from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional
import weechat
from websocket import ABNF, WebSocketConnectionClosedException, create_connection
from slack.proxy import Proxy
from slack.shared import shared
from slack.slack_api import SlackApi
from slack.slack_conversation import SlackConversation
from slack.slack_user import SlackBot, SlackUser
from slack.task import Future, create_task
from slack.util import get_callback_name
if TYPE_CHECKING:
from slack_api.slack_bots_info import SlackBotInfo
from slack_api.slack_users_info import SlackUserInfo
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_item(key))
return self[key]
async def initialize_items(self, item_ids: Iterable[str]):
item_ids_to_init = set(item_id for item_id in item_ids if item_id not in self)
if item_ids_to_init:
items_info_task = create_task(self._fetch_items_info(item_ids_to_init))
for item_id in item_ids_to_init:
self[item_id] = create_task(self._create_item(item_id, items_info_task))
async def _create_item(
self,
item_id: str,
items_info_task: Optional[Future[Dict[str, SlackUserInfo]]] = None,
) -> SlackUser:
if items_info_task:
items_info = await items_info_task
return SlackUser(self.workspace, item_id, items_info[item_id])
else:
return await SlackUser.create(self.workspace, item_id)
async def _fetch_items_info(self, item_ids: Iterable[str]):
response = await self.workspace.api.fetch_users_info(item_ids)
return {info["id"]: info for info in response["users"]}
class SlackBots(Dict[str, Future[SlackBot]]):
def __init__(self, workspace: SlackWorkspace):
super().__init__()
self.workspace = workspace
def __missing__(self, key: str):
self[key] = create_task(self._create_item(key))
return self[key]
async def initialize_items(self, item_ids: Iterable[str]):
item_ids_to_init = set(item_id for item_id in item_ids if item_id not in self)
if item_ids_to_init:
items_info_task = create_task(self._fetch_items_info(item_ids_to_init))
for item_id in item_ids_to_init:
self[item_id] = create_task(self._create_item(item_id, items_info_task))
async def _create_item(
self,
item_id: str,
items_info_task: Optional[Future[Dict[str, SlackBotInfo]]] = None,
) -> SlackBot:
if items_info_task:
items_info = await items_info_task
return SlackBot(self.workspace, item_id, items_info[item_id])
else:
return await SlackBot.create(self.workspace, item_id)
async def _fetch_items_info(self, item_ids: Iterable[str]):
response = await self.workspace.api.fetch_bots_info(item_ids)
return {info["id"]: info for info in response["bots"]}
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.users = SlackUsers(self)
self.bots = SlackBots(self)
self.conversations: Dict[str, SlackConversation] = {}
@property
def is_connected(self):
return self._is_connected
@is_connected.setter
def is_connected(self, value: bool):
self._is_connected = value
weechat.bar_item_update("input_text")
async def connect(self):
rtm_connect = await self.api.fetch_rtm_connect()
self.id = rtm_connect["team"]["id"]
self.enterprise_id = rtm_connect["team"].get("enterprise_id")
self.my_user = await self.users[rtm_connect["self"]["id"]]
await self.connect_ws(rtm_connect["url"])
# "types": "public_channel,private_channel,im",
user_channels_response = await self.api.fetch_users_conversations(
"public_channel"
)
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())
self.is_connected = True
async def connect_ws(self, url: str):
sslopt_ca_certs = {}
if hasattr(ssl, "get_default_verify_paths") and callable(
ssl.get_default_verify_paths
):
ssl_defaults = ssl.get_default_verify_paths()
if ssl_defaults.cafile is not None:
sslopt_ca_certs = {"ca_certs": ssl_defaults.cafile}
proxy = Proxy()
proxy_options = {
"proxy_type": proxy.type,
"http_proxy_host": proxy.address,
"http_proxy_port": proxy.port,
"http_proxy_auth": (proxy.username, proxy.password),
"http_proxy_timeout": self.config.network_timeout.value,
}
# TODO: Handle errors
self.ws = create_connection(
url,
self.config.network_timeout.value,
sslopt=sslopt_ca_certs,
**proxy_options,
)
self.hook = weechat.hook_fd(
self.ws.sock.fileno(),
1,
0,
0,
get_callback_name(self.ws_read_cb),
"",
)
self.ws.sock.setblocking(False)
def ws_read_cb(self, data: str, fd: int) -> int:
while True:
try:
opcode, recv_data = self.ws.recv_data(control_frame=True)
except ssl.SSLWantReadError:
# No more data to read at this time.
return weechat.WEECHAT_RC_OK
except (WebSocketConnectionClosedException, socket.error) as e:
# TODO: Handle error
# handle_socket_error(e, team, "receive")
print(e)
return weechat.WEECHAT_RC_OK
if opcode == ABNF.OPCODE_PONG:
# TODO: Maybe record last time anything was received instead
self.last_pong_time = time.time()
return weechat.WEECHAT_RC_OK
elif opcode != ABNF.OPCODE_TEXT:
return weechat.WEECHAT_RC_OK
self.ws_recv(json.loads(recv_data.decode()))
def ws_recv(self, data: Any):
print(f"received: {data}")
