-
Notifications
You must be signed in to change notification settings - Fork 2
/
Bot.py
103 lines (81 loc) · 3.06 KB
/
Bot.py
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
import time, random, string
class Bot(object):
def __init__(self, s, nickname, username, realname):
self.nickname = nickname
self.username = username
self.realname = realname
self.socket = s
self.write_buffer = b""
self.read_buffer = b""
self.last_read = None
self.ping_sent = False
self._channels = []
self.last_status = 0
self.prefix = "".join([random.choice(string.ascii_uppercase) for n in range(4)])
self.use_prefix = True
def fileno(self):
return self.socket.fileno()
def queue_send(self, data):
encoded_data = data.encode("utf8")
if self.use_prefix:
self.write_buffer += ":{} ".format(self.prefix).encode("utf8")
self.write_buffer += encoded_data + b"\r\n"
def send(self):
self.write_buffer = self.write_buffer[self.socket.send(self.write_buffer):]
def waiting_send(self):
return bool(len(self.write_buffer))
def read(self):
data = b""
try:
data = self.read_buffer+self.socket.recv(2048)
except:
return []
self.read_buffer = b""
data_lines = [line.strip(b"\r") for line in data.split(b"\n")]
if data_lines[-1]:
self.read_buffer = data_lines[-1]
data_lines.pop(-1)
decoded_lines = []
for line in data_lines:
try:
line = line.decode("utf8")
except:
try:
line = line.decode("latin-1")
except:
continue
decoded_lines.append(line)
self.last_read = time.time()
self.ping_sent = False
return decoded_lines
def identify(self):
self.send_identify(self.nickname, self.username, self.realname)
def send_identify(self, nickname, username, realname):
self.send_nick(nickname)
self.send_user(username, realname)
def send_ping(self, text="hello"):
self.queue_send("PING :{0}".format(text))
self.ping_sent = True
def send_pong(self, text):
self.queue_send("PONG :{0}".format(text))
def send_privmsg(self, target, text):
self.queue_send("PRIVMSG {0} :{1}".format(target,text))
def send_join(self, target):
self.queue_send("JOIN :{0}".format(target))
def send_quit(self, text):
self.queue_send("QUIT :{0}".format(text))
def send_nick(self, nick):
self.queue_send("NICK :{}".format(nick))
def send_user(self, ident, real_name):
self.queue_send("USER {} - - {}".format(ident, real_name))
def send_who(self, channel):
self.queue_send("WHO {}".format(channel))
def is_own_nickname(self, nickname):
return self.nickname==nickname
def add_channel(self, channel):
self._channels.append(channel)
def remove_channel(self, channel):
if channel in self._channels:
self._channels.remove(channel)
def summary(self):
return "<{} ({}): {}>".format(self.nickname, self.last_status, ", ".join(self._channels))