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
|
#--depends-on commands
#--depends-on permissions
from src import IRCLine, ModuleManager, utils
class Module(ModuleManager.BaseModule):
@utils.hook("received.command.nick")
@utils.kwarg("help", "Change my nickname")
@utils.kwarg("permission", "changenickname")
@utils.spec("!<nickname>word")
def change_nickname(self, event):
event["server"].send_nick(event["spec"][0])
@utils.hook("received.command.raw")
@utils.kwarg("help", "Send a line of raw IRC data")
@utils.kwarg("permission", "raw")
@utils.spec("!<line>string")
def raw(self, event):
if IRCLine.is_human(event["spec"][0]):
line = IRCLine.parse_human(event["spec"][0])
else:
line = IRCLine.parse_line(event["spec"][0])
line = event["server"].send(line)
if not line == None:
event["stdout"].write("Sent: %s" % line.parsed_line.format())
else:
event["stderr"].write("Line was filtered")
@utils.hook("received.command.part")
@utils.kwarg("help", "Part from the current or given channel")
@utils.kwarg("permission", "part")
@utils.kwarg("require_mode", "high")
@utils.kwarg("require_access", "high,part")
@utils.spec("!-privateonly !<channel>word")
@utils.spec("!-channelonly ?<channel>word")
def part(self, event):
event["server"].send_part(event["spec"][0] or event["target"].name)
@utils.hook("received.command.join")
@utils.kwarg("help", "Join a given channel")
@utils.kwarg("permission", "join")
@utils.spec("!<channel>word")
def join(self, event):
event["server"].send_join(event["spec"][0])
def _id_from_alias(self, alias):
return self.bot.database.servers.get_by_alias(alias)
def _server_from_alias(self, alias):
id, server = self._both_from_alias(alias)
return server
def _both_from_alias(self, alias):
id = self._id_from_alias(alias)
if id == None:
raise utils.EventError("Unknown server alias")
return id, self.bot.get_server_by_id(id)
@utils.hook("received.command.reconnect")
@utils.kwarg("help", "Reconnect to the current, or provided, server")
@utils.kwarg("permission", "reconnect")
@utils.spec("?<server>word")
def reconnect(self, event):
alias = event["spec"][0] or str(event["server"])
server = self._server_from_alias(alias)
if server:
line = server.send_quit("Reconnecting")
line.events.on("send").hook(lambda e: self.bot.reconnect(
server.id, server.connection_params))
if not server == event["server"]:
event["stdout"].write("Reconnecting to %s" % alias)
else:
event["stdout"].write("Not connected to %s" % alias)
@utils.hook("received.command.connect", min_args=1)
@utils.kwarg("help", "Connect to a given server")
@utils.kwarg("permission", "connect")
@utils.spec("!<server>word")
def connect(self, event):
alias = event["spec"][0]
server = self._server_from_alias(alias)
if server:
raise utils.EventError("Already connected to %s" % str(server))
server = self.bot.add_server(self._id_from_alias(alias))
event["stdout"].write("Connecting to %s" % str(server))
@utils.hook("received.command.disconnect")
@utils.kwarg("help", "Disconnect from the current or provided server")
@utils.kwarg("permission", "disconnect")
@utils.spec("?<server>word")
def disconnect(self, event):
alias = event["spec"][0] or str(event["server"])
id, server = self._both_from_alias(alias)
if not server == None:
server.disconnect()
self.bot.disconnect(server)
elif id in self.bot.reconnections:
self.bot.reconnections[id].cancel()
del self.bot.reconnections[id]
else:
raise utils.EventError("Server not connected")
if not server == event["server"]:
event["stdout"].write("Disconnected from %s" % alias)
@utils.hook("received.command.shutdown")
@utils.kwarg("help", "Shutdown the bot")
@utils.kwarg("permission", "shutdown")
@utils.spec("?<reason>string")
def shutdown(self, event):
reason = event["spec"][0] or "Shutting down"
for server in self.bot.servers.values():
line = server.send_quit(reason)
line.events.on("send").hook(self._shutdown_hook(server))
def _shutdown_hook(self, server):
def shutdown(e):
server.disconnect()
self.bot.disconnect(server)
return shutdown
@utils.hook("received.command.addserver")
@utils.kwarg("help", "Add a new server")
@utils.kwarg("pemission", "addserver")
@utils.spec(
"!<alias>word !<hostname:port>word !<nickname!username@bindhost>word")
def add_server(self, event):
alias = event["spec"][0]
hostname, sep, port = event["spec"][1].partition(":")
tls = port.startswith("+")
port = port.lstrip("+")
if not hostname or not port or not port.isdigit():
raise utils.EventError("Please provide <hostname>:[+]<port>")
port = int(port)
hostmask = IRCLine.parse_hostmask(event["spec"][2])
nickname = hostmask.nickname
username = hostmask.username or nickname
realname = nickname
bindhost = hostmask.hostname or None
try:
server_id = self.bot.database.servers.add(alias, hostname, port, "",
tls, bindhost, nickname, username, realname)
except Exception as e:
event["stderr"].write("Failed to add server")
self.log.error("failed to add server \"%s\"", [alias],
exc_info=True)
return
event["stdout"].write("Added server '%s'" % alias)
@utils.hook("received.command.editserver")
@utils.kwarg("help", "Edit server details")
@utils.kwarg("permission", "editserver")
@utils.spec("!<alias>word !<option>word !<value>string")
def edit_server(self, event):
alias = event["spec"][0]
server_id = self._id_from_alias(alias)
if server_id == None:
raise utils.EventError("Unknown server '%s'" % alias)
option = event["spec"][1].lower()
value = event["spec"][2]
value_parsed = None
if option == "hostname":
value_parsed = value
elif option == "port":
if not value.isdigit():
raise utils.EventError("Invalid port")
value_parsed = int(value.lstrip("0"))
elif option == "tls":
value_lower = value.lower()
if not value_lower in ["yes", "no"]:
raise utils.EventError("TLS should be either 'yes' or 'no'")
value_parsed = value_lower == "yes"
elif option == "password":
value_parsed = value
elif option == "bindhost":
value_parsed = value
elif option in ["nickname", "username", "realname"]:
value_parsed = value
else:
raise utils.EventError("Unknown option '%s'" % option)
self.bot.database.servers.edit(server_id, option, value_parsed)
event["stdout"].write("Set %s for %s" % (option, alias))
|
