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
|
#--depends-on commands
import time
from src import ModuleManager, utils
class Module(ModuleManager.BaseModule):
def _uptime(self):
return utils.to_pretty_time(int(time.time()-self.bot.start_time))
@utils.hook("received.command.uptime")
def uptime(self, event):
"""
:help: Show my uptime
"""
event["stdout"].write("Uptime: %s" % self._uptime())
@utils.hook("api.get.uptime")
def uptime_api(self, event):
return self._uptime()
def _stats(self):
networks = len(self.bot.servers)
channels = 0
users = 0
for server in self.bot.servers.values():
channels += len(server.channels)
users += len(server.users)
return [networks, channels, users]
@utils.hook("received.command.stats")
def stats(self, event):
"""
:help: Show my network/channel/user stats
"""
networks, channels, users = self._stats()
response = "I currently have %d network" % networks
if networks > 1:
response += "s"
response += ", %d channel" % channels
if channels > 1:
response += "s"
response += " and %d visible user" % users
if users > 1:
response += "s"
event["stdout"].write(response)
@utils.hook("api.get.stats")
def stats_api(self, event):
networks, channels, users = self._stats()
return {"networks": networks, "channels": channels, "users": users}
def _server_stats(self, server):
return {
"hostname": server.connection_params.hostname,
"port": server.connection_params.port,
"tls": server.connection_params.tls,
"alias": server.connection_params.alias,
"hostmask": "%s!%s@%s" % (
server.nickname, server.username, server.hostname),
"users": len(server.users),
"bytes-written": server.socket.bytes_written,
"bytes-read": server.socket.bytes_read,
"connected-since": server.socket.connect_time,
"channels": {
c.name: self._channel_stats(c) for c in server.channels
},
"capabilities": list(server.agreed_capabilities),
"version": server.version
}
@utils.hook("api.get.servers")
def servers_api(self, event):
if event["path"]:
server_id = event["path"][0]
if not server_id.isdigit():
return None
server_id = int(server_id)
server = self.bot.get_server_by_id(server_id)
if not server:
return None
return self._server_stats(server)
else:
servers = {}
for server in self.bot.servers.values():
servers[server.id] = self._server_stats(server)
return servers
def _channel_stats(self, channel):
return {
"users": sorted([user.nickname for user in channel.users],
key=lambda nickname: nickname.lower()),
"topic": channel.topic,
"topic-set-at": channel.topic_time,
"topic-set-by": channel.topic_setter_nickname,
"modes": channel.mode_str()
}
@utils.hook("api.get.channels")
def channels_api(self, event):
if event["path"]:
server_id = event["path"][0]
if not server_id.isdigit():
return None
server_id = int(server_id)
server = self.bot.get_server_by_id(server_id)
if not server:
return None
channels = {}
for channel in server.channels.values():
channels[channel.name] = self._channel_stats(channel)
return channels
else:
channels = {}
for server in self.bot.servers.values():
channels[server.id] = {}
for channel in server.channels.values():
channels[server.id][str(channel)] = self._channel_stats(
channel)
return channels
@utils.hook("api.get.modules")
def modules_api(self, event):
return list(self.bot.modules.modules.keys())
@utils.hook("received.command.caps")
def capabilities(self, event):
"""
:help: List negotiated IRCv3 capabilities
"""
event["stdout"].write("IRCv3 capabilities: %s" %
", ".join(event["server"].agreed_capabilities))
|
