aboutsummaryrefslogtreecommitdiffstats
path: root/slack/register.py
blob: d8ab7e31fd6fbef0bf223fe4d7e7591145b0c756 (plain) (blame)
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
from __future__ import annotations

import weechat

from slack.commands import register_commands
from slack.config import SlackConfig
from slack.shared import shared
from slack.slack_emoji import load_standard_emojis
from slack.task import run_async, sleep
from slack.util import get_callback_name, with_color

SCRIPT_AUTHOR = "Trygve Aaberge <trygveaa@gmail.com>"
SCRIPT_LICENSE = "MIT"
SCRIPT_DESC = "Extends weechat for typing notification/search/etc on slack.com"


def shutdown_cb():
    shared.script_is_unloading = True
    weechat.config_write(shared.config.weechat_config.pointer)
    return weechat.WEECHAT_RC_OK


def signal_buffer_switch_cb(data: str, signal: str, buffer_pointer: str) -> int:
    prev_buffer_pointer = shared.current_buffer_pointer
    shared.current_buffer_pointer = buffer_pointer

    if prev_buffer_pointer != buffer_pointer:
        prev_slack_buffer = shared.buffers.get(prev_buffer_pointer)
        if prev_slack_buffer:
            run_async(prev_slack_buffer.mark_read())

    slack_buffer = shared.buffers.get(buffer_pointer)
    if slack_buffer:
        run_async(slack_buffer.buffer_switched_to())

    return weechat.WEECHAT_RC_OK


def input_text_changed_cb(data: str, signal: str, buffer_pointer: str) -> int:
    reset_completion_context_on_input(buffer_pointer)
    return weechat.WEECHAT_RC_OK


def input_text_cursor_moved_cb(data: str, signal: str, buffer_pointer: str) -> int:
    reset_completion_context_on_input(buffer_pointer)
    return weechat.WEECHAT_RC_OK


def reset_completion_context_on_input(buffer_pointer: str):
    slack_buffer = shared.buffers.get(buffer_pointer)
    if slack_buffer and slack_buffer.completion_context != "IN_PROGRESS_COMPLETION":
        slack_buffer.completion_context = "NO_COMPLETION"


def modifier_input_text_display_with_cursor_cb(
    data: str, modifier: str, buffer_pointer: str, string: str
) -> str:
    prefix = ""
    slack_buffer = shared.buffers.get(buffer_pointer)
    if slack_buffer:
        input_delim_color = weechat.config_string(
            weechat.config_get("weechat.bar.input.color_delim")
        )
        input_delim_start = with_color(input_delim_color, "[")
        input_delim_end = with_color(input_delim_color, "]")
        if (
            not slack_buffer.workspace.is_connected
            and not slack_buffer.workspace.is_connecting
        ):
            prefix += (
                f"{input_delim_start}"
                f"{with_color(shared.config.color.disconnected.value, 'disconnected')}"
                f"{input_delim_end} "
            )
        if slack_buffer.is_loading or slack_buffer.workspace.is_connecting:
            prefix += (
                f"{input_delim_start}"
                f"{with_color(shared.config.color.loading.value, 'loading')}"
                f"{input_delim_end} "
            )
    return prefix + string


def typing_self_cb(data: str, signal: str, signal_data: str) -> int:
    if not shared.config.look.typing_status_self or signal != "typing_self_typing":
        return weechat.WEECHAT_RC_OK

    slack_buffer = shared.buffers.get(signal_data)
    if slack_buffer:
        slack_buffer.set_typing_self()
    return weechat.WEECHAT_RC_OK


def ws_ping_cb(data: str, remaining_calls: int) -> int:
    for workspace in shared.workspaces.values():
        if workspace.is_connected:
            workspace.ping()
    return weechat.WEECHAT_RC_OK


async def init_async():
    auto_connect = weechat.info_get("auto_connect", "") == "1"
    if auto_connect:
        await sleep(1)  # Defer auto connect to ensure the logger plugin is loaded
        for workspace in shared.workspaces.values():
            if workspace.config.autoconnect:
                run_async(workspace.connect())


def register():
    if weechat.register(
        shared.SCRIPT_NAME,
        SCRIPT_AUTHOR,
        shared.SCRIPT_VERSION,
        SCRIPT_LICENSE,
        SCRIPT_DESC,
        get_callback_name(shutdown_cb),
        "",
    ):
        shared.weechat_version = int(weechat.info_get("version_number", "") or 0)
        shared.current_buffer_pointer = weechat.current_buffer()
        shared.standard_emojis = load_standard_emojis()
        shared.standard_emojis_inverse = {
            value["unicode"]: value for value in shared.standard_emojis.values()
        }
        shared.workspaces = {}
        shared.config = SlackConfig()
        shared.config.config_read()
        register_commands()

        weechat.hook_signal(
            "buffer_switch", get_callback_name(signal_buffer_switch_cb), ""
        )
        weechat.hook_signal(
            "input_text_changed", get_callback_name(input_text_changed_cb), ""
        )
        weechat.hook_signal(
            "input_text_cursor_moved", get_callback_name(input_text_cursor_moved_cb), ""
        )
        weechat.hook_modifier(
            "100|input_text_display_with_cursor",
            get_callback_name(modifier_input_text_display_with_cursor_cb),
            "",
        )
        weechat.hook_signal("typing_self_*", get_callback_name(typing_self_cb), "")
        weechat.hook_timer(5000, 0, 0, get_callback_name(ws_ping_cb), "")

        run_async(init_async())