from __future__ import annotations
import hashlib
from collections import OrderedDict
from typing import TYPE_CHECKING, Dict, List, Mapping, NoReturn, Optional, Tuple, Union
import weechat
from slack.error import SlackError
from slack.python_compatibility import removeprefix
from slack.shared import shared
from slack.slack_buffer import SlackBuffer
from slack.slack_message import SlackMessage, SlackTs
from slack.slack_thread import SlackThread
from slack.slack_user import SlackBot, SlackUser, nick_color
from slack.task import gather, run_async
if TYPE_CHECKING:
from slack_api.slack_conversations_info import SlackConversationsInfo
from slack_rtm.slack_rtm_message import (
SlackMessageChanged,
SlackMessageDeleted,
SlackMessageReplied,
SlackShRoomJoin,
SlackShRoomUpdate,
SlackUserTyping,
)
from typing_extensions import Literal
from slack.slack_workspace import SlackWorkspace
def invalidate_nicklists():
for workspace in shared.workspaces.values():
for conversation in workspace.open_conversations.values():
conversation.nicklist_needs_refresh = True
def sha1_hex(string: str) -> str:
return str(hashlib.sha1(string.encode()).hexdigest())
def hash_from_ts(ts: SlackTs) -> str:
return sha1_hex(str(ts))
class SlackConversationMessageHashes(Dict[SlackTs, str]):
def __init__(self, conversation: SlackConversation):
self._conversation = conversation
self._inverse_map: Dict[str, SlackTs] = {}
def __setitem__(self, key: SlackTs, value: str) -> NoReturn:
raise RuntimeError("Set from outside isn't allowed")
def __delitem__(self, key: SlackTs) -> None:
if key in self:
hash_key = self[key]
del self._inverse_map[hash_key]
super().__delitem__(key)
def _setitem(self, key: SlackTs, value: str) -> None:
super().__setitem__(key, value)
def __missing__(self, key: SlackTs) -> str:
hash_len = 3
full_hash = hash_from_ts(key)
short_hash = full_hash[:hash_len]
while any(
existing_hash.startswith(short_hash) for existing_hash in self._inverse_map
):
hash_len += 1
short_hash = full_hash[:hash_len]
if short_hash[:-1] in self._inverse_map:
ts_with_same_hash = self._inverse_map.pop(short_hash[:-1])
other_full_hash = hash_from_ts(ts_with_same_hash)
other_short_hash = other_full_hash[:hash_len]
while short_hash == other_short_hash:
hash_len += 1
short_hash = full_hash[:hash_len]
other_short_hash = other_full_hash[:hash_len]
self._setitem(ts_with_same_hash, other_short_hash)
self._inverse_map[other_short_hash] = ts_with_same_hash
other_message = self._conversation.messages[ts_with_same_hash]
run_async(self._conversation.rerender_message(other_message))
if other_message.thread_buffer is not None:
run_async(other_message.thread_buffer.update_buffer_props())
for reply in other_message.replies.values():
run_async(self._conversation.rerender_message(reply))
self._setitem(key, short_hash)
self._inverse_map[short_hash] = key
return self[key]
def get_ts(self, ts_hash: str) -> Optional[SlackTs]:
hash_without_prefix = removeprefix(ts_hash, "$")
return self._inverse_map.get(hash_without_prefix)
class SlackConversation(SlackBuffer):
def __init__(
self,
workspace: SlackWorkspace,
info: SlackConversationsInfo,
):
super().__init__()
self._workspace = workspace
self._info = info
self._members: Optional[List[str]] = None
self._messages: OrderedDict[SlackTs, SlackMessage] = OrderedDict()
self._nicklist: Dict[Union[SlackUser, SlackBot], str] = {}
self.nicklist_needs_refresh = True
self.message_hashes = SlackConversationMessageHashes(self)
@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 id(self) -> str:
return self._info["id"]
@property
def workspace(self) -> SlackWorkspace:
return self._workspace
@property
def context(self) -> Literal["conversation", "thread"]:
return "conversation"
@property
def messages(self) -> Mapping[SlackTs, SlackMessage]:
return self._messages
@property
def type(self) -> Literal["channel", "private", "mpim", "im"]:
if self._info["is_im"] is True:
return "im"
elif self._info["is_mpim"] is True:
return "mpim"
elif self._info["is_private"] is True:
return "private"
else:
return "channel"
@property
def buffer_type(self) -> Literal["private", "channel"]:
return "private" if self.type in ("im", "mpim") else "channel"
@property
def last_read(self) -> SlackTs:
return SlackTs(self._info["last_read"])
@last_read.setter
def last_read(self, value: SlackTs):
self._info["last_read"] = value
self.set_unread_and_hotlist()
async def sort_key(self) -> str:
type_sort_key = {
"channel": 0,
"private": 1,
"mpim": 2,
"im": 3,
}[self.type]
name = await self.name()
return f"{type_sort_key}{name}".lower()
async def name(self) -> str:
if 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:
members = await self.load_members(load_all=True)
member_users = await gather(
*(
self.workspace.users[user_id]
for user_id in members
if user_id != self.workspace.my_user.id
)
)
return ",".join(sorted(user.nick() for user in member_users))
else:
return self._info["name"]
def name_prefix(
self,
name_type: Literal["full_name", "short_name", "short_name_without_padding"],
) -> str:
if self.type == "im":
if name_type == "short_name":
return " "
else:
return ""
elif self.type == "mpim":
if name_type == "short_name" or name_type == "short_name_without_padding":
return "@"
else:
return ""
elif self.type == "private":
return "&"
else:
return "#"
async def name_with_prefix(
self,
name_type: Literal["full_name", "short_name", "short_name_without_padding"],
) -> str:
return f"{self.name_prefix(name_type)}{await self.name()}"
def should_open(self):
if "is_open" in self._info:
if self._info["is_open"]:
return True
elif self._info.get("is_member"):
return True
return False
async def get_name_and_buffer_props(self) -> Tuple[str, Dict[str, str]]:
name_without_prefix = await self.name()
name = f"{self.name_prefix('full_name')}{name_without_prefix}"
short_name = self.name_prefix("short_name") + name_without_prefix
return name, {
"short_name": short_name,
"title": "topic",
"input_multiline": "1",
"nicklist": "0" if self.type == "im" else "1",
"nicklist_display_groups": "0",
"localvar_set_type": self.buffer_type,
"localvar_set_slack_type": self.type,
"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 gather(self.nicklist_update(), self.fill_history())
async def open_buffer(self, switch: bool = False):
await super().open_buffer(switch)
self.workspace.open_conversations[self.id] = self
async def rerender_message(self, message: SlackMessage):
await super().rerender_message(message)
parent_message = message.parent_message
if parent_message and parent_message.thread_buffer:
await parent_message.thread_buffer.rerender_message(message)
async def load_members(self, load_all: bool = False):
if self._members is None:
members_response = await self._api.fetch_conversations_members(
self, limit=None if load_all else 1000
)
self._members = members_response["members"]
self.workspace.users.initialize_items(self._members)
return self._members
async def fetch_replies(self, thread_ts: SlackTs) -> List[SlackMessage]:
replies_response = await self._api.fetch_conversations_replies(self, thread_ts)
messages = [
SlackMessage(self, message) for message in replies_response["messages"]
]
if thread_ts != messages[0].ts:
raise SlackError(
self.workspace,
f"First message in conversations.replies response did not match thread_ts {thread_ts}",
replies_response,
)
if thread_ts not in self._messages:
self._messages[thread_ts] = messages[0]
parent_message = self._messages[thread_ts]
replies = messages[1:]
for reply in replies:
parent_message.replies[reply.ts] = reply
self._messages[reply.ts] = reply
parent_message.replies = OrderedDict(sorted(parent_message.replies.items()))
self._messages = OrderedDict(sorted(self._messages.items()))
parent_message.reply_history_filled = True
return replies
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)
conversation_messages = [
SlackMessage(self, message) for message in history["messages"]
]
for message in reversed(conversation_messages):
self._messages[message.ts] = message
if self.display_thread_replies():
await gather(
*(
self.fetch_replies(message.ts)
for message in conversation_messages
if message.is_thread_parent
)
)
self._messages = OrderedDict(sorted(self._messages.items()))
messages = [
message
for message in self._messages.values()
if self.should_display_message(message)
]
sender_user_ids = [m.sender_user_id for m in messages if m.sender_user_id]
if self.display_reaction_nicks():
reaction_user_ids = [
user_id
for m in messages
for reaction in m.reactions
for user_id in reaction["users"]
]
user_ids = sender_user_ids + reaction_user_ids
else:
user_ids = sender_user_ids
self.workspace.users.initialize_items(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))
for message in messages:
await self.print_message(message)
self.history_filled = True
self.history_pending = False
async def nicklist_update(self):
if self.nicklist_needs_refresh and self.type != "im":
self.nicklist_needs_refresh = False
members = await self.load_members()
users = await gather(
*(self.workspace.users[user_id] for user_id in members)
)
for user in users:
self.nicklist_add_user(user)
def nicklist_add_user(
self, user: Union[SlackUser, SlackBot], nick: Optional[str] = None
):
if user in self._nicklist or self.type == "im":
return
nicklist_nick = nick if nick else user.nick(only_nick=True)
# TODO: weechat.color.nicklist_away
color = (
""
if not shared.config.look.color_nicks_in_nicklist
else nick_color(nick)
if nick and (not isinstance(user, SlackUser) or not user.is_self)
else user.nick_color()
)
prefix = (
shared.config.look.external_user_suffix.value
if isinstance(user, SlackUser) and user.is_external
else ""
)
visible = 1 if isinstance(user, SlackUser) else 0
nick_pointer = weechat.nicklist_add_nick(
self.buffer_pointer, "", nicklist_nick, color, prefix, "", visible
)
self._nicklist[user] = nick_pointer
def nicklist_remove_user(self, user: Union[SlackUser, SlackBot]):
if self.type == "im":
return
nick_pointer = self._nicklist.pop(user)
weechat.nicklist_remove_nick(self.buffer_pointer, nick_pointer)
def display_thread_replies(self) -> bool:
buffer_value = weechat.buffer_get_string(
self.buffer_pointer, "localvar_display_thread_replies_in_channel"
)
if buffer_value:
return bool(weechat.config_string_to_boolean(buffer_value))
return shared.config.look.display_thread_replies_in_channel.value
def display_reaction_nicks(self) -> bool:
buffer_value = weechat.buffer_get_string(
self.buffer_pointer, "localvar_display_reaction_nicks"
)
if buffer_value:
return bool(weechat.config_string_to_boolean(buffer_value))
return shared.config.look.display_reaction_nicks.value
def should_display_message(self, message: SlackMessage) -> bool:
return (
not message.is_reply
or message.is_thread_broadcast
or self.display_thread_replies()
)
async def add_new_message(self, message: SlackMessage):
# TODO: Remove old messages
self._messages[message.ts] = message
if self.should_display_message(message):
if self.history_filled:
await self.print_message(message)
else:
weechat.buffer_set(
self.buffer_pointer, "hotlist", message.priority.value
)
parent_message = message.parent_message
if parent_message:
parent_message.replies[message.ts] = message
if parent_message.thread_buffer:
await parent_message.thread_buffer.print_message(message)
elif message.thread_ts is not None:
await self.fetch_replies(message.thread_ts)
if message.sender_user_id:
user = await self.workspace.users[message.sender_user_id]
if message.is_reply:
if parent_message and parent_message.thread_buffer:
weechat.hook_signal_send(
"typing_set_nick",
weechat.WEECHAT_HOOK_SIGNAL_STRING,
f"{parent_message.thread_buffer.buffer_pointer};off;{user.nick()}",
)
else:
weechat.hook_signal_send(
"typing_set_nick",
weechat.WEECHAT_HOOK_SIGNAL_STRING,
f"{self.buffer_pointer};off;{user.nick()}",
)
async def change_message(
self, data: Union[SlackMessageChanged, SlackMessageReplied]
):
ts = SlackTs(data["ts"])
message = self._messages.get(ts)
if message:
message.update_message_json(data["message"])
await self.rerender_message(message)
async def delete_message(self, data: SlackMessageDeleted):
ts = SlackTs(data["deleted_ts"])
if ts in self.message_hashes:
del self.message_hashes[ts]
message = self._messages.get(ts)
if message:
message.deleted = True
await self.rerender_message(message)
async def update_message_room(
self, data: Union[SlackShRoomJoin, SlackShRoomUpdate]
):
ts = SlackTs(data["room"]["thread_root_ts"])
message = self._messages.get(ts)
if message:
message.update_message_json_room(data["room"])
await self.rerender_message(message)
async def reaction_add(self, message_ts: SlackTs, reaction: str, user_id: str):
message = self._messages.get(message_ts)
if message:
message.reaction_add(reaction, user_id)
await self.rerender_message(message)
async def reaction_remove(self, message_ts: SlackTs, reaction: str, user_id: str):
message = self._messages.get(message_ts)
if message:
message.reaction_remove(reaction, user_id)
await self.rerender_message(message)
async def typing_add_user(self, data: SlackUserTyping):
if not shared.config.look.typing_status_nicks:
return
user = await self.workspace.users[data["user"]]
if "thread_ts" not in data:
weechat.hook_signal_send(
"typing_set_nick",
weechat.WEECHAT_HOOK_SIGNAL_STRING,
f"{self.buffer_pointer};typing;{user.nick()}",
)
else:
thread_ts = SlackTs(data["thread_ts"])
parent_message = self._messages.get(thread_ts)
if parent_message and parent_message.thread_buffer:
weechat.hook_signal_send(
"typing_set_nick",
weechat.WEECHAT_HOOK_SIGNAL_STRING,
f"{parent_message.thread_buffer.buffer_pointer};typing;{user.nick()}",
)
async def open_thread(self, thread_hash: str, switch: bool = False):
thread_ts = self.message_hashes.get_ts(thread_hash)
if thread_ts:
thread_message = self.messages.get(thread_ts)
if thread_message is None:
# TODO: Fetch message
return
if thread_message.thread_buffer is None:
thread_message.thread_buffer = SlackThread(thread_message)
await thread_message.thread_buffer.open_buffer(switch)
async def print_message(self, message: SlackMessage):
await super().print_message(message)
sender = await message.sender
if message.subtype in ["channel_leave", "group_leave"]:
self.nicklist_remove_user(sender)
else:
nick = await message.nick(colorize=False, only_nick=True)
self.nicklist_add_user(sender, nick)
async def mark_read(self):
last_read_line_ts = self.last_read_line_ts()
if last_read_line_ts and last_read_line_ts != self.last_read:
await self._api.conversations_mark(self, last_read_line_ts)
