ref: c199809be6ad8db9c451f0575fa7d525ce4cce7f dotfiles/.weechat/python/talkative.py -rw-r--r-- 8.4 KiB View raw
                                                                                
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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
#!/usr/bin/env python3
# Portions of this script are from
# https://weechat.org/scripts/source/anotify.py.html/
import shlex
import subprocess
import weechat

# TODO:
# - Seek to last line you were highlighted on
# - Queue of lines to speak, instead of always stopping the last thing
# - Remember a history of things we've said, so they can be repeated (more than
#   just the last message)

weechat.register("talkative", "Drew DeVault <sir@cmpwn.com>",
        "1.0", "MIT", "Makes weechat speak aloud", "", "")

last_msg = ""
disabled = False
subscriptions = set()

if weechat.config_is_set_plugin("subscriptions"):
    subs = weechat.config_string(weechat.config_get_plugin("subscriptions"))
    subscriptions = set(subs.split(";"))

def espeak(text, blocking=False):
    global last_msg
    last_msg = text
    if disabled:
        return
    text = weechat.string_remove_color(text, "")
    subprocess.run(["pkill", "espeak"])
    subprocess.Popen(["espeak", "--", text],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

def get_buffer(search):
    infolist = weechat.infolist_get("hotlist", "", "")
    while weechat.infolist_next(infolist):
        buf = weechat.infolist_pointer(infolist, "buffer_pointer")
        short_name = weechat.buffer_get_string(buf, "short_name")
        if search.lower() in short_name.lower():
            weechat.infolist_free(infolist)
            return buf
    weechat.infolist_free(infolist)
    return None

def update_subscriptions():
    subs = ";".join(subscriptions)
    weechat.config_set_plugin("subscriptions", subs)

def cb_irc_server_connected(data, signal, signal_data):
    espeak(f"Connected to {signal_data}")
    return weechat.WEECHAT_RC_OK

def cb_irc_server_disconnected(data, signal, signal_data):
    espeak(f"Disconnected from {signal_data}")
    return weechat.WEECHAT_RC_OK

def cb_buffer_switch(data, signal, signal_data):
    name = weechat.buffer_get_string(signal_data, "short_name")
    espeak(f"buffer {name}")
    return weechat.WEECHAT_RC_OK

TAGGED_MESSAGES = {
    'public message or action': set(['irc_privmsg', 'notify_message']),
    'private message or action': set(['irc_privmsg', 'notify_private']),
    'notice message': set(['irc_notice', 'notify_private']),
    'invite message': set(['irc_invite', 'notify_highlight']),
    'channel topic': set(['irc_topic', ]),
}

def cb_process_message(data, wbuffer, date, tags, displayed,
        highlight, prefix, message):
    tags = set(tags.split(','))
    is_public_message = tags.issuperset(
            TAGGED_MESSAGES['public message or action'])
    is_private_message = tags.issuperset(
            TAGGED_MESSAGES['private message or action'])
    if wbuffer == weechat.current_buffer() and is_public_message:
        espeak(f"{prefix} said... {message}")
        return weechat.WEECHAT_RC_OK
    if is_private_message:
        espeak(f"PM from {prefix} said... {message}")
        return weechat.WEECHAT_RC_OK
    bufname = weechat.buffer_get_string(wbuffer, "name")
    if wbuffer != weechat.current_buffer() and bufname in subscriptions:
        short_name = weechat.buffer_get_string(wbuffer, "short_name")
        espeak(f"Activity in {short_name}")
        return weechat.WEECHAT_RC_OK
    return weechat.WEECHAT_RC_OK

def summarize():
    infolist = weechat.infolist_get("hotlist", "", "")
    summary = "summary of channels with activity\n"
    while weechat.infolist_next(infolist):
        priority = weechat.infolist_integer(infolist, "priority")
        buf = weechat.infolist_pointer(infolist, "buffer_pointer")
        short_name = weechat.buffer_get_string(buf, "short_name")
        if priority == 1:
            summary += f"new message in {short_name}\n"
        if priority == 2:
            summary += f"private message from {short_name}\n"
        if priority == 3:
            summary += f"highlighted in {short_name}\n"
    weechat.infolist_free(infolist)
    espeak(summary)

line = None
last_buf = None
hdata_line = weechat.hdata_get('line')
hdata_line_data = weechat.hdata_get('line_data')

def seek_line(which):
    global line, last_buf
    last_buf = weechat.current_buffer()
    own_lines = weechat.hdata_pointer(
            weechat.hdata_get("buffer"),
            weechat.current_buffer(),
            "own_lines")
    if not own_lines:
        return weechat.WEECHAT_RC_OK
    line = weechat.hdata_pointer(
            weechat.hdata_get('lines'), own_lines, which)

def repeat_line(direction):
    global line, last_buf
    if weechat.current_buffer() != last_buf:
        seek_line('last_line')
    else:
        if direction != 0:
            line = weechat.hdata_move(hdata_line, line, direction)
    while line:
        data = weechat.hdata_pointer(hdata_line, line, 'data')
        prefix = weechat.hdata_string(hdata_line_data, data, "prefix")
        message = weechat.hdata_string(hdata_line_data, data, "message")
        displayed = weechat.hdata_char(hdata_line_data, data, "displayed")
        if displayed == 1:
            espeak(f"{prefix} said... {message}")
            break
    else:
        espeak("no more lines to read")

def repeat_previous():
    repeat_line(-1)

def repeat_next():
    repeat_line(1)

def repeat_last():
    seek_line('last_line')
    repeat_line(0)

def repeat_first():
    seek_line('first_line')
    repeat_line(0)

def talkative_cmd_cb(data, buf, args):
    args = shlex.split(args)
    if args[0] == "repeat":
        espeak(last_msg)
    elif args[0] == "disable":
        espeak("talkative disabled")
        disable = True
    elif args[0] == "enable":
        disable = False
        espeak("talkative enabled")
    elif args[0] == "cancel":
        subprocess.run(["pkill", "espeak"])
    elif args[0] == "speak":
        espeak(" ".join(args[1:]))
    elif args[0] == "input":
        buf = weechat.current_buffer()
        pending = weechat.buffer_get_string(buf, "input")
        espeak(f"pending text: {pending}")
    elif args[0] == "summarize":
        summarize()
    elif args[0] == "previous":
        repeat_previous()
    elif args[0] == "next":
        repeat_next()
    elif args[0] == "last":
        repeat_last()
    elif args[0] == "first":
        repeat_first()
    elif args[0] == "whereami":
        buf = weechat.current_buffer()
        espeak(f"buffer {weechat.buffer_get_string(buf, 'short_name')}")
    elif args[0] == "subscribe":
        if len(args) == 2:
            buf = get_buffer(args[1])
        else:
            buf = weechat.current_buffer()
        if buf is None:
            espeak("Could not find buffer")
            return weechat.WEECHAT_RC_OK
        name = weechat.buffer_get_string(buf, "name")
        subscriptions.add(name)
        short_name = weechat.buffer_get_string(buf, "short_name")
        espeak(f"Subscribed to {short_name}")
        update_subscriptions()
    elif args[0] == "unsubscribe":
        if len(args) > 1:
            buf = get_buffer(args[1])
        else:
            buf = weechat.current_buffer()
        if buf is None:
            espeak("Could not find buffer")
            return weechat.WEECHAT_RC_OK
        name = weechat.buffer_get_string(buf, "name")
        subscriptions.remove(name)
        short_name = weechat.buffer_get_string(buf, "short_name")
        espeak(f"Unsubscribed from {short_name}")
        update_subscriptions()
    else:
        espeak("unknown talkative command")
    return weechat.WEECHAT_RC_OK

weechat.hook_signal('irc_server_connected', 'cb_irc_server_connected', '')
weechat.hook_signal('irc_server_disconnected',
        'cb_irc_server_disconnected', '')
weechat.hook_signal('buffer_switch',
        'cb_buffer_switch', '')
weechat.hook_print('', '', '', 1, 'cb_process_message', '')
weechat.hook_command("talk", "controls the talkative plugin",
        "/talk [action] [args...]",
        "disable: turn off talkative\n" +
        "enable: turn on talkative\n" +
        "first: read the first line in the current buffer\n" +
        "input: read out the text pending in the input buffer\n" +
        "last: read the last line in the current buffer\n",
        "next: read the next line in the current buffer\n" +
        "previous: read the previous line in the current buffer\n" +
        "repeat: repeat the last thing talkative said\n" +
        "speak [text]: say 'text' aloud\n" +
        "subscribe: subscribes to noficiations for buffer activity\n" +
        "summarize: list the buffers which have unread messages\n" +
        "unsubscribe: unsubscribes to noficiations for buffer activity\n" +
        "", "talkative_cmd_cb", "")
weechat.prnt("", "talkative ready")