Bot

Complexity

Total Complexity

23

Size/Duplication

Total Lines	94
Duplicated Lines	0 %

Importance

Changes	3
Bugs	0	Features	0

16 Methods

Rating	Name	Duplication	Size	Complexity
A	log_error()	0	12	2
A	handler()	0	2	1
A	change_user()	0	5	3
A	add_user()	0	2	1
A	handle_connect()	0	4	1
A	remove_user()	0	4	3
A	part_user()	0	2	1
A	next_nick()	0	3	1
A	set_nick()	0	3	1
A	collect_incoming_data()	0	2	1
A	say()	0	7	2
A	print_debug()	0	3	2
A	write()	0	3	1
A	found_terminator()	0	5	1
A	__init__()	0	10	1
A	run()	0	13	2

#!/usr/bin/python

from datetime import datetime
from time import time
import random
import signal
import socket
import asyncore
import asynchat

import settings
import logic


class Bot(asynchat.async_chat):

    def __init__(self):
        asynchat.async_chat.__init__(self)
        self.known_users = {}
        self.buffer = ''
        self.set_terminator('\r\n')
        self.nick_num = 0
        self.nick = settings.NICKS[self.nick_num]
        self.logic = logic.BotLogic(self)
        self.ac_in_buffer_size = self.ac_out_buffer_size = 8192   # 2*default
        self.start_time = time()

    def print_debug(self, text):
        if settings.DEBUG:
            print(text)

    def write(self, text):
        self.print_debug('> %s' % text)
        self.push(text + '\r\n')

    def say(self, text, channel):
        if not text:
            return
        line = 'PRIVMSG %s :%s' % (channel, text)
        self.write(line)
        self.logic.self_input(channel, text, line)
        return line

    def set_nick(self):
        self.nick = settings.NICKS[self.nick_num]
        self.write('NICK %s' % self.nick)

    def next_nick(self):
        self.nick_num = (self.nick_num + 1) % len(settings.NICKS)
        self.set_nick()

    def handle_connect(self):
        self.set_nick()
        self.write('USER %s iw 0 :%s' %
                   (settings.IDENT, settings.REAL_NAME))

    def collect_incoming_data(self, data):
        self.buffer += data

    def log_error(self, text):
        with open("error.log", "a") as f:
            f.write(str(datetime.now()) + '\n')
            f.write('Error: ' + text + '\n')
            self.print_debug(text)

            from traceback import format_exception
            from sys import exc_info
            ex_type, ex_val, ex_tb = exc_info()
            ex_text = ''.join(format_exception(ex_type, ex_val, ex_tb, 10))
            f.write(ex_text + '\n')
            self.print_debug(ex_text)

    def found_terminator(self):
        line = self.buffer
        self.buffer = ''
        self.print_debug("< " + line)
        self.logic.new_input(line)

    def run(self):
        def handler(signal, frame):
            pass

        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connect((settings.IRC_SERVER, settings.IRC_PORT))

        random.seed()
        # set the signal handler for shouting random messages
        signal.signal(signal.SIGALRM, handler)
        signal.alarm(20)

        asyncore.loop()

    def part_user(self, channel, nick, msg):
        del self.known_users[channel][nick.lower()]

    def remove_user(self, channel, nick, msg):
        for channel in self.known_users:
            if nick.lower() in self.known_users[channel]:
                self.part_user(channel, nick, msg)

    def add_user(self, channel, nick, msg):
        self.known_users[channel][nick.lower()] = nick

    def change_user(self, channel, old_nick, new_nick):
        for channel in self.known_users:
            if old_nick.lower() in self.known_users[channel]:
                del self.known_users[channel][old_nick.lower()]
                self.known_users[channel][new_nick.lower()] = new_nick