irc_bot.IrcBot.__init__() - Code Metrics - mosbth/irc2phpbb - Measure and Improve Code Quality continuously with Scrutinizer

irc_bot.IrcBot.init() A
last analyzed 2025-08-06 17:11 UTC

↳ Parent: irc_bot

Complexity

Conditions

Size

Total Lines	16
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	13
nop	1
dl	0
loc	16
ccs	4
cts	4
cp	1
crap	1
rs	9.75
c	0
b	0
f	0

#! /usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Module for the IRC bot.

Connecting, sending and receiving messages and doing custom actions.

Keeping a log and reading incoming material.
"""
import logging
import os
import shutil
import socket

import chardet

from bot import Bot

LOG = logging.getLogger("bot")

class IrcBot(Bot):
    """Bot implementing the IRC protocol"""
    def __init__(self):
        super().__init__()
        self.CONFIG = {
            "server": None,
            "port": 6667,
            "channel": None,
            "nick": "marvin",
            "realname": "Marvin The All Mighty dbwebb-bot",
            "ident": None,
            "dirIncoming": "incoming",
            "dirDone": "done",
            "lastfm": None,
        }

        # Socket for IRC server
        self.SOCKET = None

    def connectToServer(self):
        """Connect to the IRC Server"""

        # Create the socket  & Connect to the server
        server = self.CONFIG["server"]
        port = self.CONFIG["port"]

        if server and port:
            self.SOCKET = socket.socket()
            LOG.info("Connecting: %s:%d", server, port)
            self.SOCKET.connect((server, port))
        else:
            LOG.error("Failed to connect, missing server or port in configuration.")
            return

        # Send the nick to server
        nick = self.CONFIG["nick"]
        if nick:
            msg = f'NICK {nick}\r\n'
            self.sendMsg(msg)
        else:
            LOG.info("Ignore sending nick, missing nick in configuration.")

        # Present yourself
        realname = self.CONFIG["realname"]
        self.sendMsg(f'USER  {nick} 0 * :{realname}\r\n')

        # This is my nick, i promise!
        ident = self.CONFIG["ident"]
        if ident:
            self.sendMsg(f'PRIVMSG nick IDENTIFY {ident}\r\n')
        else:
            LOG.info("Ignore identifying with password, ident is not set.")

        # Join a channel
        channel = self.CONFIG["channel"]
        if channel:
            self.sendMsg(f'JOIN {channel}\r\n')
        else:
            LOG.info("Ignore joining channel, missing channel name in configuration.")

    def sendPrivMsg(self, message, channel):
        """Send and log a PRIV message"""
        if channel == self.CONFIG["channel"]:
            self.MSG_LOG.debug("%s <%s>  %s", channel, self.CONFIG["nick"], message)

        msg = f"PRIVMSG {channel} :{message}\r\n"
        self.sendMsg(msg)

    def sendMsg(self, msg):
        """Send and occasionally print the message sent"""
        LOG.debug("SEND: %s", msg.rstrip("\r\n"))
        self.SOCKET.send(msg.encode())

    def decode_irc(self, raw, preferred_encs=None):
        """
        Do character detection.
        You can send preferred encodings as a list through preferred_encs.
        http://stackoverflow.com/questions/938870/python-irc-bot-and-encoding-issue
        """
        if preferred_encs is None:
            preferred_encs = ["UTF-8", "CP1252", "ISO-8859-1"]

        changed = False
        enc = None
        res = None
        for enc in preferred_encs:
            try:
                res = raw.decode(enc)
                changed = True
                break
            except Exception:
                pass

        if not changed:
            try:
                enc = chardet.detect(raw)['encoding']
                res = raw.decode(enc)
            except Exception:
                res = raw.decode(enc, 'ignore')

        return res

    def receive(self):
        """Read incoming message and guess encoding"""
        lines = None
        try:
            buf = self.SOCKET.recv(2048)
            lines = self.decode_irc(buf)
            lines = lines.split("\n")
            buf = lines.pop()
        except Exception as err:
            LOG.error("Error reading incoming message %s", err)

        return lines

    def readincoming(self):
        """
        Read all files in the directory incoming, send them as a message if
        they exists and then move the file to directory done.
        """
        if not os.path.isdir(self.CONFIG["dirIncoming"]):
            return

        listing = os.listdir(self.CONFIG["dirIncoming"])

        for infile in listing:
            filename = os.path.join(self.CONFIG["dirIncoming"], infile)

            with open(filename, "r", encoding="UTF-8") as f:
                for msg in f:
                    self.sendPrivMsg(msg, self.CONFIG["channel"])

            try:
                shutil.move(filename, self.CONFIG["dirDone"])
            except Exception:
                LOG.warning("Failed to move %s to %s. Deleting.", filename, self.CONFIG["dirDone"])
                os.remove(filename)

    def mainLoop(self):
        """For ever, listen and answer to incoming chats"""
        while 1:
            # Check in any in the incoming directory
            self.readincoming()

            for line in self.receive():
                LOG.debug(line)
                words = line.strip().split()

                if not words:
                    continue

                self.checkIrcActions(words)
                self.checkMarvinActions(words)

    def begin(self):
        """Start the bot"""
        self.connectToServer()
        self.mainLoop()

    def checkIrcActions(self, words):
        """
        Check if Marvin should take action on any messages defined in the
        IRC protocol.
        """
        if words[0] == "PING":
            self.sendMsg(f"PONG {words[1]}\r\n")

        if words[1] == 'INVITE':
            self.sendMsg(f'JOIN {words[3]}\r\n')

    def checkMarvinActions(self, words):
        """Check if Marvin should perform any actions"""
        if words[1] == 'PRIVMSG' and words[2] == self.CONFIG["channel"]:
            self.MSG_LOG.debug("%s <%s>  %s",
                               words[2],
                               words[0].split(":")[1].split("!")[0],
                               " ".join(words[3:]))

        if words[1] == 'PRIVMSG':
            raw = ' '.join(words[3:])
            row = self.tokenize(raw)

            if self.CONFIG["nick"] in row:
                for action in self.ACTIONS:
                    msg = action(row)
                    if msg:
                        self.sendPrivMsg(msg, words[2])
                        break
            else:
                for action in self.GENERAL_ACTIONS:
                    msg = action(row)
                    if msg:
                        self.sendPrivMsg(msg, words[2])
                        break


1		#! /usr/bin/env python3
2		# -- coding: utf-8 --
3
4		"""
5		Module for the IRC bot.
6
7		Connecting, sending and receiving messages and doing custom actions.
8
9		Keeping a log and reading incoming material.
10		"""
11	1	import logging
12	1	import os
13	1	import shutil
14	1	import socket
15
16	1	import chardet
17
18	1	from bot import Bot
19
20	1	LOG = logging.getLogger("bot")
21
22	1	class IrcBot(Bot):
23		"""Bot implementing the IRC protocol"""
24	1	def __init__(self):
25	1	super().__init__()
26	1	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	1	self.SOCKET = None
40
41	1	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	1	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	1	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	1	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	1	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	1	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	1	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	1	def begin(self):
177		"""Start the bot"""
178		self.connectToServer()
179		self.mainLoop()
180
181	1	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	1	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
216

mosbth / irc2phpbb

irc_bot.IrcBot.__init__() A last analyzed 2025-08-06 17:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

irc_bot.IrcBot.init() A
last analyzed 2025-08-06 17:11 UTC