-
Notifications
You must be signed in to change notification settings - Fork 7
Expand file tree
/
Copy pathplugin.py
More file actions
254 lines (210 loc) · 8.24 KB
/
plugin.py
File metadata and controls
254 lines (210 loc) · 8.24 KB
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
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
import logging
import re
import sys
# noinspection PyUnresolvedReferences
import utils
from utils import irc_nickname
from pybot import pybot
# noinspection PyUnresolvedReferences
from color import color
from functools import wraps
class plugin:
def __init__(self, bot: pybot):
self.bot = bot
self.config = self.bot.config[self._get_class_name()] if self._get_class_name() in self.bot.config else None
self.logger = logging.getLogger(self._get_class_name())
if self.logger.level > logging.DEBUG and 'debug' in self.config and self.config['debug']:
self.logger.setLevel(logging.DEBUG)
self.assert_config()
def _get_class_name(self):
return self.__class__.__name__
def unload_plugin(self):
"""
called when plugin needs to be disabled / reloaded
"""
pass
def assert_config(self):
"""
called by plugin superclass to assert config values
should throw if config is invalid
in such case, plugin won't be loaded
"""
pass
# see https://www.alien.net.au/irc/irc2numerics.html
# for deep explanation
def on_welcome(self, raw_msg, server, port, nickname, **kwargs):
"""
called by bot when connected to server
:param raw_msg : raw IRC msg
:param server : server address
:param port : server's port
:param nickname : bot's nickname
"""
def on_disconnect(self, raw_msg, server, port, **kwargs):
"""
called by bot when disconnected from server
:param raw_msg : raw IRC msg
:param server : server address
:param port : server's port
"""
def on_join(self, raw_msg, source, **kwargs):
"""
called by bot when somebody joins channel
:param raw_msg : raw IRC msg
:param source : source of joined user
"""
def on_me_joined(self, raw_msg, **kwargs):
"""
called by bot when joined channel
:param raw_msg : raw IRC msg
"""
def on_mode(self, raw_msg, source, mode_change, **kwargs):
"""
called by bot when someone's mode changed
:param raw_msg : raw IRC msg
:param source : changer's source
:param mode_change : e.g. ['+o', 'some_nick'], ['-v', 'some_nick'], ['+r'], ['+f', '#some_channel']
"""
def on_pubmsg(self, raw_msg, source, msg, **kwargs):
"""
called by bot when public msg received
:param raw_msg : raw IRC msg
:param source : pubmsg source
:param msg : full message
"""
def on_kick(self, raw_msg, who, source, **kwargs):
"""
called by bot when somebody got kicked
:param raw_msg : raw IRC msg
:param who : user who gets kicked
:param source : kicker's source
"""
def on_me_kicked(self, raw_msg, source, **kwargs):
"""
called by bot when kicked from channel
:param raw_msg : raw IRC msg
:param source : kicker's source
"""
pass
def on_privmsg(self, raw_msg, msg, source, **kwargs):
"""
called by bot when private msg received
:param raw_msg : raw IRC msg
:param msg : full message
:param source : privmsg source
"""
def on_whoisuser(self, raw_msg, nick, user, host, **kwargs):
"""
called by bot when /whois response received
:param raw_msg : raw IRC msg
:param nick : requested nickname
:param user : requested username
:param host : requested hostname
"""
def on_nicknameinuse(self, raw_msg, busy_nickname, **kwargs):
"""
called by bot when given nickname is reserved
:param raw_msg : raw IRC msg
:param busy_nickname : nickname bot was trying to use
"""
def on_nick(self, raw_msg, source, old_nickname, new_nickname, **kwargs):
"""
called by bot when somebody changes nickname
:param raw_msg : raw IRC msg
:param source : changer's source (with old nickname!)
:param old_nickname : old nickname
:param new_nickname : new nickname
"""
def on_part(self, raw_msg, source, **kwargs):
"""
called by bot when somebody lefts channel
:param raw_msg : raw IRC msg
:param source : source of left user
"""
def on_quit(self, raw_msg, source, **kwargs):
"""
called by bot when somebody disconnects from IRC server
:param raw_msg : raw IRC msg
:param source : source of disconnected user
"""
def on_ctcp(self, raw_msg, source, msg, **kwargs):
"""
called by bot when ctcp arrives (/me ...)
:param raw_msg : raw IRC msg
:param source : source of ctcped user
:param msg : ctcp message
"""
def on_namreply(self, raw_msg, nicknames, **kwargs):
"""
called by bot when names response arrives
:param raw_msg : raw IRC msg
:param nicknames : nicknames in channel
"""
pass
def command_alias(*args):
def command_alias_decorator(func):
if any(len(arg.split()) > 1 for arg in args):
print(f'invalid aliases for {func.__qualname__}')
sys.exit(9)
func.__aliases = [*args]
return func
return command_alias_decorator
def command(_cls=None, admin=False, superadmin=False, channel_op=False):
def command_impl(func):
@wraps(func)
def command_impl_impl(self, sender_nick, **kwargs):
sender_nick = irc_nickname(sender_nick)
if hasattr(command_impl_impl, '__superadmin') and sender_nick != self.bot.config['superop']:
self.logger.info(f'{sender_nick} is not superop, skipping command')
self.bot.say(f"{sender_nick}: you're not bot owner, sorry!")
return
if hasattr(command_impl_impl, '__admin') and not self.bot.is_user_op(sender_nick):
self.logger.info(f'{sender_nick} is not op, skipping command')
self.bot.say(f"{sender_nick}: you're not bot operator, sorry!")
return
if hasattr(command_impl_impl, '__channel_op') and not self.bot.get_channel().is_oper(sender_nick):
self.logger.info(f'{sender_nick} is not channel operator, skipping command')
self.bot.say(f"{sender_nick}: you're not channel operator, sorry!")
return
try:
func(self, sender_nick=sender_nick, **kwargs)
except Exception as e:
self.logger.error(f'exception caught calling {func.__qualname__}: {type(e).__name__}: {e}')
self.bot.say('internal error, sorry :(')
utils.report_error()
if hasattr(command_impl_impl, '__regex'):
print(f'function {func.__qualname__} already registered as regex handler')
sys.exit(8)
command_impl_impl.__command = True
if admin: command_impl_impl.__admin = True
if superadmin: command_impl_impl.__admin = command_impl_impl.__superadmin = True
if channel_op: command_impl_impl.__channel_op = True
return command_impl_impl
if _cls is None:
return command_impl
else:
return command_impl(_cls)
def on_message(regex_str):
def on_message_impl(func):
@wraps(func)
def on_message_impl_impl(self, **kwargs):
try:
func(self, **kwargs)
except Exception as e:
self.logger.error(f'exception caught calling {func.__qualname__}: {type(e).__name__}: {e}')
utils.report_error()
if hasattr(on_message_impl_impl, '__command'):
print(f'function {func.__qualname__} already registered as command')
sys.exit(7)
try:
on_message_impl_impl.__regex = re.compile(regex_str)
return on_message_impl_impl
except Exception as e:
print(f'invalid regex for regex handler {func.__qualname__}: {type(e).__name__}: {e}')
sys.exit(5)
return on_message_impl
def doc(doc_string):
def doc_impl(obj):
obj.__doc_string = doc_string.strip()
return obj
return doc_impl