# Command line IRC client

I made this IRC client ages ago in Python, and thought I'd revisit it for Python 3.5 (I'd like to play with asyncio). Before I start, I'd like to have an overall design review about this.

Known issues, although feel free to comment on them:

• I do too much in one class
• There is a fair bit of repetition that could probably be pulled out

I don't think there are any obvious bugs, but I never had a rigorous test suite for this (read as: there was never a test suite, I wrote this before I discovered unit testing) so its very possible I've missed something.

I'm also not an expert in the IRC spec - this was mostly googling and trial and error. If I've misunderstood/misused/missed something please comment on that as well.

from __future__ import absolute_import, print_function, unicode_literals

from functools import partial
from multiprocessing import dummy
import datetime
import enum
import logging
import select
import socket
import time

now = datetime.datetime.now()
logging.basicConfig(
filename=''.join(["Logs/", str(datetime.datetime.now()), ".log"]),
level=logging.INFO
)

ErrorCodes = enum(
'ErrorCodes',
'UNKNOWN_HOST UNKNOWN_FAILURE UNKNOWN_CHANNEL UNKOWN_USER '
'MESSAGE_SENT PONG_SUCCESS SERVER_CONNECTED HOSTNAME_NOT_RESOLVED '
'SERVER_DISCONNECTED CHANNEL_JOINED CHANNEL_NAME_MALFORMED CHANNEL_LEFT '
)

class IrcMember(object):
"""Represents an individual who uses the IRC client.

Only stores non-sensitive information.

Attributes
----------
nickname : str
User nickname.
real_name : str
User's "real" name.
ident : str
User's id
servers: dict
Server name to socket mapping for connected servers.
server_channels: dict
Server name to a list of channel names.
server_data: dict
Server name to dict of user information if it differs from the
default values.
Socket lock.
replies: dict
Pending replies.
DEFAULT_PORT: int
The default port to use.
"""

DEFAULT_PORT = 6667

def __init__(self, nick, **kwargs):
"""Creates a new member.

Parameters
----------
nick : str
The user's nickname.
real : str, optional
The user's "real" name, defaults to the nickname.
ident : str, optional
The user's id, defaults to the nickname.
"""

self.nickname = nick
self.real_name = nick
self.ident = nick

for key, value in kwargs.iteritems():
self.__dict__[key] = value

self.servers = {}
self.server_channels = {}
self.server_data = {}

self.replies = {}

def send_server_message(self, hostname, message):
"""Send a message to a server.

Parameters
----------
hostname : str
Name of the server to send to.
message : str
Message to send.
"""

if hostname not in self.servers:
logging.warning("No such server {}".format(hostname))
logging.warning("Failed to send message {}".format(message))
return ErrorCodes.UNKNOWN_HOST

sock = self.servers[hostname]
try:
sock.send("{} \r\n".format(message.rstrip()))
except socket.error as e:
logging.exception(e)
logging.warning("Failed to send message {}".format(message))
return ErrorCodes.UNKNOWN_FAILURE
else:
return ErrorCodes.MESSAGE_SENT

def send_channel_message(self, hostname, chan_name, message):
"""Sends a message to a channel.

Parameters
----------
hostname : str
Name of the server to send to
chan_name : str
Name of the channel
message : str
Message to send
"""

if hostname not in self.servers:
logging.warning("Not connected to server {}".format(hostname))
logging.warning("Failed to send message {}".format(message))
return ErrorCodes.UNKNOWN_HOST
elif chan_name not in self.server_channels[hostname]:
logging.warning("Not in channel {}".format(chan_name))
logging.warning("Failed to send message {}".format(message))
return ErrorCodes.UNKNOWN_CHANNEL
else:
return self.send_private_message(
hostname, chan_name, message, channel=True
)

def send_private_message(self, hostname, name, message, channel=False):
"""Sends a private message.

Parameters
----------
hostname: str
Name of the server to send to
name: str
Name of the user or channel to send to
message: str
Message to send
channel: bool, optional
Whether or not this is a channel message
"""

if hostname not in self.servers:
logging.warning("No such server {}".format(hostname))
logging.warning("Failed to send message {}".format(message))
return ErrorCodes.UNKNOWN_HOST

if not (channel or self.user_exists(name)):
return ErrorCodes.UNKNOWN_USER

message = "PRIVMSG {}: {}".format(username, message.rstrip())
return self.send_server_message(hostname, message)

"""Validate a user exists.

Parameters
----------
Name of the user.
"""

## TODO: implement this
return True

def ping_pong(self, sock, data):
"""Pong a server.

Parameters
----------
sock : socket.socket
Socket to pong on.
data : str
Data to send in pong.
"""

try:
sock.send("PONG {}\r\n".format(data))
except socket.error as e:
logging.exception(e)
logging.warn("Couldn't pong the server")
return ErrorCodes.UNKNOWN_FAILURE
else:
return ErrorCodes.PONG_SUCCESS

def join_server(self, hostname, port=None, **kwargs):
"""Join a server.

Parameters
----------
hostname: str
Name of the server to join.
port: int, optional
Port to connect on - defaults to IrcMember.DEFAULT_PORT.
nickname: str, optional
Nickname to use on this server
real_name: str, optional
'Real' name ot use on this server
ident: str, optional
Identity to use on this server.
"""

if port is None:
port = IrcMember.DEFAULT_PORT

if hostname in self.servers:
return ErrorCodes.SERVER_CONNECTED

nick = self.nickname
ident = self.ident
realname = self.real_name

## Checking if the data for this server is different from the defaults
if kwargs:
self.serv_to_data[hostname] = {}
for key, value in kwargs.items():
if key in ['nickname', 'real_name', 'ident']:
self.server_data[hostname][key] = value
locals()[key] = value
else:
logging.info(
"key-value pair {}: {} unusued".format(key, value)
)
if not self.server_data[hostname]:
del self.server_data[hostname]

try:
ip = socket.gethostbyname(hostname)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((ip, port))
self.servers[hostname] = sock
self.serv_to_chan[hostname] = []
sock.settimeout(2)
self.send_server_message(
hostname, "NICK {}\r\n".format(nick))
self.send_server_message(hostname,
"USER {} {} bla: {}\r\n".format(nick, ident, realname))

except socket.gaierror as e:
logging.exception(e)
return ErrorCodes.HOSTNAME_NOT_RESOLVED

except socket.error as e:
logging.exception(e)
if port != IrcMember.DEFAULT_PORT:
logging.warning(
"Consider using port %s (the defacto IRC port) not of %s"
% (IrcMember.DEFAULT_PORT, port)
)
return ErrorCodes.UNKNOWN_FAILURE

else:
logging.info("Connected to {} on {}".format(hostname, port))
return ErrorCodes.SERVER_CONNECTED

def leave_server(self, hostname):
"""Leave a server.

Parameters
----------
hostname: str
Server to disconnect from.
"""

if hostname not in self.servers:
logging.warning("Not connected to {}".format(hostname))
return ErrorCodes.SERVER_DISCONNECTED

try:
self.send_server_message(hostname, "QUIT\r\n")
self.servers[hostname].close()
except socket.error as e:
logging.exception(e)
logging.warning("Failed to leave server {}".format(hostname))
return 1
else:
for attr_name in ['servers', 'server_channels', 'server_data']:
try:
del getattr(self, attr_name)[hostname]
except KeyError:
# This host doesn't have any data about that
pass
logging.info("Left server {}".format(hostname))
return ErrorCodes.SERVER_DISCONNECTED

def join_channel(self, hostname, channel_name):
"""Join a channel.

Parameters
----------
hostname: str
Server the channel is on.
channel_name: str
Name of the channel.
"""

if channel_name in self.serv_to_chan[hostname]:
logging.warning(
"Already connected to {} on {}".format(hostname, channel_name))
return ErrorCodes.CHANNEL_JOINED

if chan_name.startswith("#"):
try:
self.send_server_message(
hostname, "JOIN {}\r\n".format(chan_name))
except socket.error as e:
logging.exception(e)
logging.warning("Failed to connect to {}".format(chan_name))
return ErrorCodes.UNKNOWN_FAILURE
else:
self.serv_to_chan[hostname].append(chan_name)
logging.info("Connected to {}".format(chan_name))
return ErrorCodes.CHANNEL_JOINED
else:
logging.warning("Channel names should look like #<channel_name>")
return ErrorCodes.CHANNEL_NAME_MALFORMED

def leave_channel(self, hostname, chan_name):
"""Leave a channel.

Parameters
----------
hostname: str
Server the channel is on.
channel_name: str
Name of the channel.
"""

if hostname not in self.servers:
logging.warning("No such server {}".format(hostname))
return ErrorCodes.UNKNOWN_HOST
elif chan_name not in self.serv_to_chan[hostname]:
logging.warning("No such channel {}".format(chan_name))
return ErrorCodes.CHANNEL_LEFT
else:
try:
self.send_server_message(
hostname, "PART {}\r\n".format(chan_name)
)
except socket.error as e:
logging.exception(e)
logging.warning("Failed to leave {}".format(chan_name))
return ErrorCodes.UNKNOWN_FAILURE
else:
self.serv_to_chan[hostname].remove(chan_name)
logging.info("Left channel {}".format(chan_name))
return ErrorCodes.CHANNEL_LEFT

"""Display all messages waiting to be received.

Parameters
----------
buff_size: int, optional
How large of a buffer to receive with.
"""

ready, _, _ = select.select(self.servers.values(), [], [], 5)

for host, sock in self.servers.iteritems():
try:
pool = dummy.Pool()
buff_size=buff_size),

with self.lock:
replies, self.replies = self.replies, {}

print("{} :\n\n".format(server))
print(" {}".format(message))
except socket.error as e:
logging.exception(e)
logging.warning("Failed to get messages")
return ErrorCodes.UNKNOWN_FAILURE

"""Receive a message from a single server.

Parameters
----------
hostname: tuple
buff_size: int, optional
How large of a buffer to receive with.

Notes
-----
Has already checked that there is a message waiting.
"""

hostname = hostname[0]
sock = self.servers[hostname]

while True:
try:
break

for line in temp:
line = line.rstrip().split()
if (line[0] == "PING"):
self.ping_pong(sock, line[1])
else:
line = " ".join(line)
except socket.error:
break

with self.lock:
try:
except KeyError:

def __del__(self):
for host, sock in self.servers.items():
self.leave_server(host)

if __name__ == "__main__":

me = IRC_member(NICK)
me.join_server(HOST, nickname='test', ident='test', real_name='test')
time.sleep(1)
me.join_channel(HOST, CHAN)
time.sleep(1)
i = 0

while i < 100:
start = time.time()
msg = raw_input("Would you like to say something? ")
if msg == 'n':
break
if msg.rstrip():
me.send_channel_message(HOST, CHAN, msg)
end = time.time()
if (end-start) < 5:
time.sleep(int(5-(end-start)))
i += 1


Okay, since bounties are worth giving people a run for their money, and I love Python, and I miss my days of making IRC bots... I'll give this a go.. I'll try to "dig into how [you] design and implement the IRC client as a whole", particularly with regards to the spec (Which I have actually (somewhat, as much as is possible with things like RFC's) read!).

But before I do that, there are a couple of issues I'd like to point out overall.

## Oh the OOPmanity

Firstly, as you mentioned - you are doing too much in one class! You're dealing with a domain where we have the benefit of some pretty well discretized concepts (read: classes). The main one I think you should consider abstracting out of IrcMember, is the idea of a Connection. Make a separate class for connections, that manages the socket, stores the (list of?) room(s) you're in, etc. This way, IrcMember can just represent the user, i.e. a real name, ident, nickname (unless that is per-server).

However, don't go OOverboard and make abstractions for channels, messages, or any thing else.

## Interactive Command-line (oh my)

Secondly, as I'm sure you're aware (As would be anyone having run the code), a command-line interface for something interactive like IRC isn't the best way, particularly without ncurses, etc. As raw_input blocks, you're unable to check for new messages (or server PING's) - which will eventually (5 - 20 minutes depending on which network) timeout your socket as idle on the server-side. Of course, if you're chatting actively you should have something to say at least every 5 minutes, so this isn't a a huge deal. But it does mean your client won't be capable of idling online (not 100% sure of your use-case for it here).

Okay, now into the good stuff...

## Channel names

From RFC1459 Section 1.3:

Channels names are strings (beginning with a '&' or '#' character) of length up to 200 characters. Apart from the the requirement that the first character being either '&' or '#'; the only restriction on a channel name is that it may not contain any spaces (' '), a control G (^G or ASCII 7), or a comma (',' which is used as a list item separator by the protocol).

(Note that there are also extensions, both standard and non-standard, to allow channel names starting with % and probably other characters as well)

So, your join_channel code which checks channel names start with '#':

   if chan_name.startswith("#"):
...
else:
logging.warning("Channel names should look like #<channel_name>")
return ErrorCodes.CHANNEL_NAME_MALFORMED


is not strictly by the book.

## No MODE's

Perhaps not a game-changer (depending on what you want/need to do in your client), but your script has no method to allow for the sending of an IRC MODE command (either for the user or for a channel). Some (not all) networks use user-modes to handle name registration, etc - but if your not concerned about that then leave it out as it is.

## KICKing, Banning, and generally staying on TOPIC

The KICK and TOPIC commands, as well as the channel MODE 'b' (ban) are used for room administration (keeping out the rif-raf, setting the channel topic, etc). If your client is just for chatting, then again don't worry about them - but they are needed in real life (unfortunately). (Also, if you plan on using IRCX - then look into the ACCESS command.)

## Make it all a bit more INVITEing

IRC supports invitation-only chat rooms, so if you plan on joining/inviting people to them then you'll need the INVITE command. Otherwise, it can (probably) be ignored.

## Channels are great... If you know their name

This one was a bit more of a 'shocker' to me, to LIST support? Is your user expected to know the exact name of the channel they desire? If not, I'd suggest adding some logic for room lists (most likely associated with the Connection instance for that server).

## Channels are great... Unless you like talking to people

Along the same lines as above, the channel functionality is somewhat limited without the NAMES command. At present, the client can join a room and send it a message - but not know if there's anybody else there or not or who they are! NAMES tells the server to send you a list of what nicknames are registered (joined) on a channel (usually, a channel that you have also joined). Most servers send a NAMES list along with a successful JOIN notification. I would recommend that you store a list of the nicknames in a room once the client has joined it.

## Miscellaneous other things missing

Your client at present has no way to WHOIS or WHO anyone else, or to send anyone (person, i.e. a nickname) a message (PRIVMSG or NOTICE), and also cannot send NOTICE's to channels. (None of which are big deals, at all, so I just wrapped them all up here).

## Some code smells I noticed

Just to finish up, a bit more on your coding... As you mentioned, "There is a fair bit of repetition that could probably be pulled out". Particularly things like send_channel_message, which just ends up calling send_private_message anyway (I do understand why, because they both use PRIVMSG, and channel messages need some additional checks first). But it just doesn't feel right reading through the code.

There (mostly) seems to be a one-to-one mapping of a lot of your functions to IRC command verbs. For example, leave_server -> QUIT, join_channel -> JOIN, send_private_message -> PRIVMSG, etc. Part of me just feels that (as you say) there is just a tad too much repetition in that architecture. (Perhaps some more abstract encapsulation of a server command and arguments?)

The leave_server function also doesn't seem to support QUIT messages, not the end of the world - but easily implemented.

Also, as much as I love Easter Eggs, what is this:

def user_exists(self, username):
"""Validate a user exists.
Parameters
----------
Name of the user.
"""
## TODO: implement this
return True


(PS, I love ## TODO: implement this) Was this going to be (one day, I'm sure) some sort of logic around eight the WHO or WHOIS commands to prevent nickname collision?

In send_server_message you should be using EAFP (Easier to Ask Forgiveness than Permission). You test if hostname not in self.servers, but surely this would normally not be the case? You should instead use a try except, and catch a KeyError on the rare occasion this condition arises:

    try:
sock = self.servers[hostname]
except KeyError:
logging.warning("No such server {}".format(hostname))
logging.warning("Failed to send message {}".format(message))
return ErrorCodes.UNKNOWN_HOST


Also semantically, I would prefer not to return from inside the else, just have the return outside of a block since there's no other way it would reach that point.

It may be worth having this as a get_host or validate_host function too. You repeatedly use a similar pattern. Confusingly though, it seems like you don't always give the same error back? Having a single function prevents duplicate code and makes errors more clear.

I also take issue with returning MESSAGE_SENT as an apparent error code. Surely that's a successful result? It could be clearer to either have a separate enum for successful results, or to rename it to something like ResultCodes.

• I'd upvote twice just because of paragraph about EAFP. I'm learning python and I always feel weird when it's abused (top often, IMO) making code a convoluted mess (for me) – Adriano Repetti Jan 30 '16 at 11:46
• The other reason for EAFP, in the general case, is if the resource exists when you check for it, but for some reason ceases to exist in the split second between checking for and actually using it. With file systems it can plausibly happen, and then you won't have nice error handling to deal with the exception. – Ogaday Feb 5 '16 at 15:45