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
|
# Part of rabbitears See LICENSE for permissions
# Copyright (C) 2022 Matt Arnold
import socket
import sys
import irctokens
import time
class IRCBadMessage(Exception):
pass
class IRCError(Exception):
pass
def printred(s):
t = f"\033[1;31m {s} \033[0m\n"
print(t)
def parsemsg(s):
"""Breaks a message from an IRC server into its prefix, command, and arguments."""
prefix = ""
trailing = []
if not s:
raise IRCBadMessage("Empty line.")
if s[0] == ":":
prefix, s = s[1:].split(" ", 1)
if s.find(" :") != -1:
s, trailing = s.split(" :", 1)
args = s.split()
args.append(trailing)
else:
args = s.split()
command = args.pop(0)
return prefix, command, args
LINEEND = "\r\n"
class IRCBot:
irc = None
def __init__(self, sock, config=None):
self.irc = sock
self.connected = False
self.config = config
def send_cmd(self, line):
"""Send an IRC Command, takes an IRC command string without the CRLF
Returns encoded msg on success raises IRCError on failure"""
if not self.connected:
raise IRCError("Not Connected")
encmsg = bytes(line.format() + LINEEND, "UTF-8")
expected = len(encmsg)
if self.irc.send(encmsg) == expected:
return str(encmsg)
else:
raise IRCError("Unexpected Send Length")
def on_welcome(self, *args, **kwargs):
authmsg = irctokens.build("NICKSERV", ["IDENTIFY", self.config["nickpass"]])
self.send_cmd(authmsg)
joinmsg = irctokens.build("JOIN", [self.config["channel"]])
self.send_cmd(joinmsg)
def send_privmsg(self, dst, msg):
msg = irctokens.build("PRIVMSG", [dst, msg])
self.send_cmd(msg)
def send_quit(self, quitmsg):
msg = irctokens.build("QUIT", [quitmsg])
print(msg)
self.send_cmd(msg)
def send_action(self, action_msg, dst):
pass
def connect(self, server, port, channel, botnick, botnickpass):
if self.config is None:
self.config = {}
self.config["hostname"] = server
self.config["port"] = port
self.config["nick"] = botnick
self.config["channel"] = channel
self.config["nickpass"] = botnickpass
print("Connecting to: " + server)
self.irc.connect((self.config["hostname"], self.config["port"]))
self.connected = True
# Perform user registration
password_msg = irctokens.build("PASS", [self.config["nickpass"]]).format()
print(password_msg)
usermsg = irctokens.build("USER", [botnick, "0", "*", "frog"]).format()
print(usermsg)
self.send_cmd(password_msg)
nickmsg = irctokens.build("NICK", [botnick]).format()
print(nickmsg)
self.send_cmd(nickmsg)
self.send_cmd(usermsg)
def get_response(self):
# Get the response
resp = self.irc.recv(4096).decode("UTF-8")
msg = parsemsg(resp)
nwmsg = irctokens.tokenise(resp)
printred(nwmsg.command)
if nwmsg.command == "001":
self.on_welcome(nwmsg.params)
if nwmsg.command == "ERROR":
raise IRCError(str(nwmsg.params[0]))
if nwmsg.command == "PING":
print("Sending pong")
self.irc.send(bytes("PONG " + LINEEND, "UTF-8"))
return msg
|
