from __future__ import annotations
import re
from itertools import chain
import weechat
from slack.commands import find_command
from slack.python_compatibility import removeprefix, removesuffix
from slack.shared import MESSAGE_ID_REGEX_STRING, REACTION_CHANGE_REGEX_STRING, shared
from slack.slack_buffer import SlackBuffer
from slack.slack_conversation import SlackConversation
from slack.slack_user import get_user_nick, name_from_user_info
from slack.task import run_async
from slack.util import get_callback_name, get_resolved_futures
REACTION_PREFIX_REGEX_STRING = (
rf"{MESSAGE_ID_REGEX_STRING}?{REACTION_CHANGE_REGEX_STRING}"
)
def completion_slack_workspaces_cb(
data: str, completion_item: str, buffer: str, completion: str
) -> int:
for workspace_name in shared.workspaces:
weechat.completion_list_add(
completion, workspace_name, 0, weechat.WEECHAT_LIST_POS_SORT
)
return weechat.WEECHAT_RC_OK
def completion_list_add_expand(
completion: str, word: str, nick_completion: int, where: str, buffer: str
):
if word == "%(slack_workspaces)":
completion_slack_workspaces_cb("", "slack_workspaces", buffer, completion)
elif word == "%(nicks)":
completion_nicks_cb("", "nicks", buffer, completion)
elif word == "%(threads)":
completion_thread_hashes_cb("", "threads", buffer, completion)
else:
weechat.completion_list_add(completion, word, nick_completion, where)
def completion_slack_workspace_commands_cb(
data: str, completion_item: str, buffer: str, completion: str
) -> int:
base_command = weechat.completion_get_string(completion, "base_command")
base_word = weechat.completion_get_string(completion, "base_word")
args = weechat.completion_get_string(completion, "args")
args_without_base_word = removesuffix(args, base_word)
found_cmd_with_args = find_command(base_command, args_without_base_word)
if found_cmd_with_args:
command = found_cmd_with_args[0]
matching_cmds = [
removeprefix(cmd, command.cmd).lstrip()
for cmd in shared.commands
if cmd.startswith(command.cmd) and cmd != command.cmd
]
if len(matching_cmds) > 1:
for match in matching_cmds:
cmd_arg = match.split(" ")
completion_list_add_expand(
completion, cmd_arg[0], 0, weechat.WEECHAT_LIST_POS_SORT, buffer
)
else:
for arg in command.completion.split("|"):
completion_list_add_expand(
completion, arg, 0, weechat.WEECHAT_LIST_POS_SORT, buffer
)
return weechat.WEECHAT_RC_OK
def completion_slack_channels_cb(
data: str, completion_item: str, buffer: str, completion: str
) -> int:
slack_buffer = shared.buffers.get(buffer)
if slack_buffer is None:
return weechat.WEECHAT_RC_OK
conversations = slack_buffer.workspace.open_conversations.values()
for conversation in conversations:
if conversation.buffer_type == "channel":
weechat.completion_list_add(
completion,
conversation.name_with_prefix("short_name_without_padding"),
0,
weechat.WEECHAT_LIST_POS_SORT,
)
return weechat.WEECHAT_RC_OK
def completion_emojis_cb(
data: str, completion_item: str, buffer: str, completion: str
) -> int:
slack_buffer = shared.buffers.get(buffer)
if slack_buffer is None:
return weechat.WEECHAT_RC_OK
base_word = weechat.completion_get_string(completion, "base_word")
reaction = re.match(REACTION_PREFIX_REGEX_STRING + ":", base_word)
prefix = reaction.group(0) if reaction else ":"
emoji_names = chain(
shared.standard_emojis.keys(), slack_buffer.workspace.custom_emojis.keys()
)
for emoji_name in emoji_names:
if "::skin-tone-" not in emoji_name:
weechat.completion_list_add(
completion,
f"{prefix}{emoji_name}:",
0,
weechat.WEECHAT_LIST_POS_SORT,
)
return weechat.WEECHAT_RC_OK
def completion_nicks_cb(
data: str, completion_item: str, buffer: str, completion: str
) -> int:
slack_buffer = shared.buffers.get(buffer)
if slack_buffer is None:
return weechat.WEECHAT_RC_OK
all_users = get_resolved_futures(slack_buffer.workspace.users.values())
all_nicks = sorted([user.nick.raw_nick for user in all_users], key=str.casefold)
for nick in all_nicks:
weechat.completion_list_add(
completion,
f"@{nick}",
1,
weechat.WEECHAT_LIST_POS_END,
)
weechat.completion_list_add(
completion,
nick,
1,
weechat.WEECHAT_LIST_POS_END,
)
buffer_nicks = sorted(
[nick.raw_nick for nick in slack_buffer.members], key=str.casefold, reverse=True
)
for nick in buffer_nicks:
weechat.completion_list_add(
completion,
nick,
1,
weechat.WEECHAT_LIST_POS_BEGINNING,
)
weechat.completion_list_add(
completion,
f"@{nick}",
1,
weechat.WEECHAT_LIST_POS_BEGINNING,
)
senders = [
m.sender_user_id
for m in slack_buffer.messages.values()
if m.sender_user_id and m.subtype in [None, "me_message", "thread_broadcast"]
]
unique_senders = list(dict.fromkeys(senders))
sender_users = get_resolved_futures(
[slack_buffer.workspace.users[sender] for sender in unique_senders]
)
nicks = [user.nick.raw_nick for user in sender_users]
for nick in nicks:
weechat.completion_list_add(
completion,
nick,
1,
weechat.WEECHAT_LIST_POS_BEGINNING,
)
weechat.completion_list_add(
completion,
f"@{nick}",
1,
weechat.WEECHAT_LIST_POS_BEGINNING,
)
my_user_nick = slack_buffer.workspace.my_user.nick.raw_nick
weechat.completion_list_add(
completion,
f"@{my_user_nick}",
1,
weechat.WEECHAT_LIST_POS_END,
)
weechat.completion_list_add(
completion,
my_user_nick,
1,
weechat.WEECHAT_LIST_POS_END,
)
return weechat.WEECHAT_RC_OK
def completion_thread_hashes_cb(
data: str, completion_item: str, buffer: str, completion: str
) -> int:
slack_buffer = shared.buffers.get(buffer)
if not isinstance(slack_buffer, SlackConversation):
return weechat.WEECHAT_RC_OK
message_tss = sorted(slack_buffer.message_hashes.keys())
messages = [slack_buffer.messages.get(ts) for ts in message_tss]
thread_messages = [
message
for message in messages
if message is not None and message.is_thread_parent
]
for message in thread_messages:
weechat.completion_list_add(
completion, message.hash, 0, weechat.WEECHAT_LIST_POS_BEGINNING
)
for message in thread_messages:
weechat.completion_list_add(
completion, f"${message.hash}", 0, weechat.WEECHAT_LIST_POS_BEGINNING
)
return weechat.WEECHAT_RC_OK
def complete_input(buffer: str, slack_buffer: SlackBuffer, query: str):
if (
slack_buffer.completion_context == "ACTIVE_COMPLETION"
and slack_buffer.completion_values
):
input_value = weechat.buffer_get_string(buffer, "input")
input_pos = weechat.buffer_get_integer(buffer, "input_pos")
result = slack_buffer.completion_values[slack_buffer.completion_index]
input_before = removesuffix(input_value[:input_pos], query)
input_after = input_value[input_pos:]
new_input = input_before + result + input_after
new_pos = input_pos - len(query) + len(result)
with slack_buffer.completing():
weechat.buffer_set(buffer, "input", new_input)
weechat.buffer_set(buffer, "input_pos", str(new_pos))
def nick_suffix():
return weechat.config_string(
weechat.config_get("weechat.completion.nick_completer")
)
async def complete_user_next(
buffer: str, slack_buffer: SlackBuffer, query: str, is_first_word: bool
):
if slack_buffer.completion_context == "NO_COMPLETION":
slack_buffer.completion_context = "PENDING_COMPLETION"
search = await slack_buffer.api.edgeapi.fetch_users_search(query)
if slack_buffer.completion_context != "PENDING_COMPLETION":
return
slack_buffer.completion_context = "ACTIVE_COMPLETION"
suffix = nick_suffix() if is_first_word else " "
slack_buffer.completion_values = [
get_user_nick(name_from_user_info(slack_buffer.workspace, user)).raw_nick
+ suffix
for user in search["results"]
]
slack_buffer.completion_index = 0
elif slack_buffer.completion_context == "ACTIVE_COMPLETION":
slack_buffer.completion_index += 1
if slack_buffer.completion_index >= len(slack_buffer.completion_values):
slack_buffer.completion_index = 0
complete_input(buffer, slack_buffer, query)
def complete_previous(buffer: str, slack_buffer: SlackBuffer, query: str) -> int:
if slack_buffer.completion_context == "ACTIVE_COMPLETION":
slack_buffer.completion_index -= 1
if slack_buffer.completion_index < 0:
slack_buffer.completion_index = len(slack_buffer.completion_values) - 1
complete_input(buffer, slack_buffer, query)
return weechat.WEECHAT_RC_OK_EAT
return weechat.WEECHAT_RC_OK
def input_complete_cb(data: str, buffer: str, command: str) -> int:
slack_buffer = shared.buffers.get(buffer)
if slack_buffer:
input_value = weechat.buffer_get_string(buffer, "input")
input_pos = weechat.buffer_get_integer(buffer, "input_pos")
input_before_cursor = input_value[:input_pos]
word_index = (
-2 if slack_buffer.completion_context == "ACTIVE_COMPLETION" else -1
)
word_until_cursor = " ".join(input_before_cursor.split(" ")[word_index:])
if word_until_cursor.startswith("@"):
query = word_until_cursor[1:]
is_first_word = word_until_cursor == input_before_cursor
if command == "/input complete_next":
run_async(
complete_user_next(buffer, slack_buffer, query, is_first_word)
)
return weechat.WEECHAT_RC_OK_EAT
else:
return complete_previous(buffer, slack_buffer, query)
return weechat.WEECHAT_RC_OK
def register_completions():
if shared.weechat_version < 0x02090000:
weechat.completion_get_string = (
weechat.hook_completion_get_string # pyright: ignore [reportAttributeAccessIssue, reportUnknownMemberType]
)
weechat.completion_list_add = (
weechat.hook_completion_list_add # pyright: ignore [reportAttributeAccessIssue, reportUnknownMemberType]
)
# Disable until working properly
# weechat.hook_command_run(
# "/input complete_*", get_callback_name(input_complete_cb), ""
# )
weechat.hook_completion(
"slack_workspaces",
"Slack workspaces (internal names)",
get_callback_name(completion_slack_workspaces_cb),
"",
)
weechat.hook_completion(
"slack_commands",
"completions for Slack commands",
get_callback_name(completion_slack_workspace_commands_cb),
"",
)
weechat.hook_completion(
"slack_channels",
"conversations in the current Slack workspace",
get_callback_name(completion_slack_channels_cb),
"",
)
weechat.hook_completion(
"slack_emojis",
"Emoji names known to Slack",
get_callback_name(completion_emojis_cb),
"",
)
weechat.hook_completion(
"nicks",
"nicks in the current Slack buffer",
get_callback_name(completion_nicks_cb),
"",
)
weechat.hook_completion(
"threads",
"complete thread ids for slack",
get_callback_name(completion_thread_hashes_cb),
"",
)
