irc_bot

  1#! /usr/bin/env python3
  2# -*- coding: utf-8 -*-
  3
  4"""
  5Module for the IRC bot.
  6
  7Connecting, sending and receiving messages and doing custom actions.
  8
  9Keeping a log and reading incoming material.
 10"""
 11import logging
 12import os
 13import shutil
 14import socket
 15
 16import chardet
 17
 18from bot import Bot
 19
 20LOG = logging.getLogger("bot")
 21
 22class IrcBot(Bot):
 23    """Bot implementing the IRC protocol"""
 24    def __init__(self):
 25        super().__init__()
 26        self.CONFIG = {
 27            "server": None,
 28            "port": 6667,
 29            "channel": None,
 30            "nick": "marvin",
 31            "realname": "Marvin The All Mighty dbwebb-bot",
 32            "ident": None,
 33            "dirIncoming": "incoming",
 34            "dirDone": "done",
 35            "lastfm": None,
 36        }
 37
 38        # Socket for IRC server
 39        self.SOCKET = None
 40
 41    def connectToServer(self):
 42        """Connect to the IRC Server"""
 43
 44        # Create the socket  & Connect to the server
 45        server = self.CONFIG["server"]
 46        port = self.CONFIG["port"]
 47
 48        if server and port:
 49            self.SOCKET = socket.socket()
 50            LOG.info("Connecting: %s:%d", server, port)
 51            self.SOCKET.connect((server, port))
 52        else:
 53            LOG.error("Failed to connect, missing server or port in configuration.")
 54            return
 55
 56        # Send the nick to server
 57        nick = self.CONFIG["nick"]
 58        if nick:
 59            msg = f'NICK {nick}\r\n'
 60            self.sendMsg(msg)
 61        else:
 62            LOG.info("Ignore sending nick, missing nick in configuration.")
 63
 64        # Present yourself
 65        realname = self.CONFIG["realname"]
 66        self.sendMsg(f'USER  {nick} 0 * :{realname}\r\n')
 67
 68        # This is my nick, i promise!
 69        ident = self.CONFIG["ident"]
 70        if ident:
 71            self.sendMsg(f'PRIVMSG nick IDENTIFY {ident}\r\n')
 72        else:
 73            LOG.info("Ignore identifying with password, ident is not set.")
 74
 75        # Join a channel
 76        channel = self.CONFIG["channel"]
 77        if channel:
 78            self.sendMsg(f'JOIN {channel}\r\n')
 79        else:
 80            LOG.info("Ignore joining channel, missing channel name in configuration.")
 81
 82    def sendPrivMsg(self, message, channel):
 83        """Send and log a PRIV message"""
 84        if channel == self.CONFIG["channel"]:
 85            self.MSG_LOG.debug("%s <%s>  %s", channel, self.CONFIG["nick"], message)
 86
 87        msg = f"PRIVMSG {channel} :{message}\r\n"
 88        self.sendMsg(msg)
 89
 90    def sendMsg(self, msg):
 91        """Send and occasionally print the message sent"""
 92        LOG.debug("SEND: %s", msg.rstrip("\r\n"))
 93        self.SOCKET.send(msg.encode())
 94
 95    def decode_irc(self, raw, preferred_encs=None):
 96        """
 97        Do character detection.
 98        You can send preferred encodings as a list through preferred_encs.
 99        http://stackoverflow.com/questions/938870/python-irc-bot-and-encoding-issue
100        """
101        if preferred_encs is None:
102            preferred_encs = ["UTF-8", "CP1252", "ISO-8859-1"]
103
104        changed = False
105        enc = None
106        res = None
107        for enc in preferred_encs:
108            try:
109                res = raw.decode(enc)
110                changed = True
111                break
112            except Exception:
113                pass
114
115        if not changed:
116            try:
117                enc = chardet.detect(raw)['encoding']
118                res = raw.decode(enc)
119            except Exception:
120                res = raw.decode(enc, 'ignore')
121
122        return res
123
124    def receive(self):
125        """Read incoming message and guess encoding"""
126        lines = None
127        try:
128            buf = self.SOCKET.recv(2048)
129            lines = self.decode_irc(buf)
130            lines = lines.split("\n")
131            buf = lines.pop()
132        except Exception as err:
133            LOG.error("Error reading incoming message %s", err)
134
135        return lines
136
137    def readincoming(self):
138        """
139        Read all files in the directory incoming, send them as a message if
140        they exists and then move the file to directory done.
141        """
142        if not os.path.isdir(self.CONFIG["dirIncoming"]):
143            return
144
145        listing = os.listdir(self.CONFIG["dirIncoming"])
146
147        for infile in listing:
148            filename = os.path.join(self.CONFIG["dirIncoming"], infile)
149
150            with open(filename, "r", encoding="UTF-8") as f:
151                for msg in f:
152                    self.sendPrivMsg(msg, self.CONFIG["channel"])
153
154            try:
155                shutil.move(filename, self.CONFIG["dirDone"])
156            except Exception:
157                LOG.warning("Failed to move %s to %s. Deleting.", filename, self.CONFIG["dirDone"])
158                os.remove(filename)
159
160    def mainLoop(self):
161        """For ever, listen and answer to incoming chats"""
162        while 1:
163            # Check in any in the incoming directory
164            self.readincoming()
165
166            for line in self.receive():
167                LOG.debug(line)
168                words = line.strip().split()
169
170                if not words:
171                    continue
172
173                self.checkIrcActions(words)
174                self.checkMarvinActions(words)
175
176    def begin(self):
177        """Start the bot"""
178        self.connectToServer()
179        self.mainLoop()
180
181    def checkIrcActions(self, words):
182        """
183        Check if Marvin should take action on any messages defined in the
184        IRC protocol.
185        """
186        if words[0] == "PING":
187            self.sendMsg(f"PONG {words[1]}\r\n")
188
189        if words[1] == 'INVITE':
190            self.sendMsg(f'JOIN {words[3]}\r\n')
191
192    def checkMarvinActions(self, words):
193        """Check if Marvin should perform any actions"""
194        if words[1] == 'PRIVMSG' and words[2] == self.CONFIG["channel"]:
195            self.MSG_LOG.debug("%s <%s>  %s",
196                               words[2],
197                               words[0].split(":")[1].split("!")[0],
198                               " ".join(words[3:]))
199
200        if words[1] == 'PRIVMSG':
201            raw = ' '.join(words[3:])
202            row = self.tokenize(raw)
203
204            if self.CONFIG["nick"] in row:
205                for action in self.ACTIONS:
206                    msg = action(row)
207                    if msg:
208                        self.sendPrivMsg(msg, words[2])
209                        break
210            else:
211                for action in self.GENERAL_ACTIONS:
212                    msg = action(row)
213                    if msg:
214                        self.sendPrivMsg(msg, words[2])
215                        break