diff options
Diffstat (limited to 'slack/completions.py')
| -rw-r--r-- | slack/completions.py | 359 |
1 files changed, 359 insertions, 0 deletions
diff --git a/slack/completions.py b/slack/completions.py new file mode 100644 index 0000000..93d4c78 --- /dev/null +++ b/slack/completions.py @@ -0,0 +1,359 @@ +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.workspace.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 [reportUnknownMemberType, reportGeneralTypeIssues] + ) + weechat.completion_list_add = ( + weechat.hook_completion_list_add # pyright: ignore [reportUnknownMemberType, reportGeneralTypeIssues] + ) + + # 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), + "", + ) |