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
|
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_conversation import get_conversation_from_buffer_pointer
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():
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:
conversation = get_conversation_from_buffer_pointer(buffer_pointer)
if conversation:
run_async(conversation.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):
conversation = get_conversation_from_buffer_pointer(buffer_pointer)
if conversation and conversation.completion_context != "IN_PROGRESS_COMPLETION":
conversation.completion_context = "NO_COMPLETION"
def modifier_input_text_display_with_cursor_cb(
data: str, modifier: str, buffer_pointer: str, string: str
) -> str:
prefix = ""
conversation = get_conversation_from_buffer_pointer(buffer_pointer)
if conversation:
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 conversation.workspace.is_connected:
prefix += (
f"{input_delim_start}"
f"{with_color(shared.config.color.disconnected.value, 'disconnected')}"
f"{input_delim_end} "
)
if conversation.is_loading:
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.value:
return weechat.WEECHAT_RC_OK
conversation = get_conversation_from_buffer_pointer(signal_data)
if conversation:
conversation.typing_update_self(signal)
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.value:
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.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(
"window_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(
"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())
|
