irc2phpbb.irc_bot.IrcBot.checkMarvinActions() - Code Metrics - Inspection of "Migration to uv and some other stuff" - mosbth/irc2phpbb - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#86)

by Daniel

created 2025-10-29 18:50 UTC

irc2phpbb.irc_bot.IrcBot.checkMarvinActions() C

↳ Parent: irc2phpbb.irc_bot

Complexity

Conditions

Size

Total Lines	24
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	20
dl	0
loc	24
rs	6.6666
c	0
b	0
f	0
cc	9
nop	2

#! /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 irc2phpbb.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			import logging
12			import os
13			import shutil
14			import socket
15
16			import chardet
17
18			from irc2phpbb.bot import Bot
19
20			LOG = logging.getLogger("bot")
21
22			class 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
216

mosbth / irc2phpbb

Pull Request — master (#86)

irc2phpbb.irc_bot.IrcBot.checkMarvinActions() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like