weitersager.irc

Complexity

Total Complexity

22

Size/Duplication

Total Lines	152
Duplicated Lines	0 %

Test Coverage

Coverage

85.71%

Importance

Changes

0

14 Methods

Rating	Name	Duplication	Size	Complexity
A	Bot.start()	0	3	1
A	Bot.on_nicknameinuse()	0	4	1
A	Bot.get_version()	0	3	1
A	DummyBot.__init__()	0	2	1
A	DummyBot.start()	0	4	2
A	Bot.__init__()	0	34	3
A	Bot.on_welcome()	0	8	1
A	DummyBot.disconnect()	0	3	1
A	Bot.on_badchannelkey()	0	4	1
A	Bot.on_join()	0	8	2
A	Bot._send_custom_commands_after_welcome()	0	4	2
A	DummyBot.say()	0	2	1
A	Bot.say()	0	3	1
A	Bot._join_channels()	0	8	2

"""
weitersager.irc
~~~~~~~~~~~~~~~

Internet Relay Chat

:Copyright: 2007-2021 Jochen Kupperschmidt
:License: MIT, see LICENSE for details.
"""

from __future__ import annotations
import logging
import ssl
from typing import Union

from irc.bot import ServerSpec, SingleServerIRCBot
from irc.connection import Factory
from jaraco.stream.buffer import LenientDecodingLineBuffer

from .config import IrcChannel, IrcConfig, IrcServer
from .signals import irc_channel_joined
from .util import start_thread


logger = logging.getLogger(__name__)


class Bot(SingleServerIRCBot):
    """An IRC bot to forward messages to IRC channels."""

    def __init__(
        self,
        server: IrcServer,
        nickname: str,
        realname: str,
        commands: list[str],
        channels: set[IrcChannel],
    ) -> None:
        logger.info(
            'Connecting to IRC server %s:%d ...', server.host, server.port
        )

        server_spec = ServerSpec(server.host, server.port, server.password)
        factory = Factory(wrapper=ssl.wrap_socket) if server.ssl else Factory()
        SingleServerIRCBot.__init__(
            self, [server_spec], nickname, realname, connect_factory=factory
        )

        if server.rate_limit is not None:
            logger.info(
                'IRC send rate limit set to %.2f messages per second.',
                server.rate_limit,
            )
            self.connection.set_rate_limit(server.rate_limit)
        else:
            logger.info('No IRC send rate limit set.')

        self.commands = commands

        # Avoid `UnicodeDecodeError` on non-UTF-8 messages.
        self.connection.buffer_class = LenientDecodingLineBuffer

        # Note: `self.channels` already exists in super class.
        self.channels_to_join = channels

    def start(self) -> None:
        """Connect to the server, in a separate thread."""
        start_thread(super().start, self.__class__.__name__)

    def get_version(self) -> str:
        """Return this on CTCP VERSION requests."""
        return 'Weitersager'

    def on_welcome(self, conn, event) -> None:
        """Join channels after connect."""
        logger.info(
            'Connected to IRC server %s:%d.', *conn.socket.getpeername()
        )

        self._send_custom_commands_after_welcome(conn)
        self._join_channels(conn)

    def _send_custom_commands_after_welcome(self, conn):
        """Send custom commands after having been welcomed by the server."""
        for command in self.commands:
            conn.send_raw(command)

    def _join_channels(self, conn):
        """Join the configured channels."""
        channels = sorted(self.channels_to_join)
        logger.info('Channels to join: %s', ', '.join(c.name for c in channels))

        for channel in channels:
            logger.info('Joining channel %s ...', channel.name)
            conn.join(channel.name, channel.password or '')

    def on_nicknameinuse(self, conn, event) -> None:
        """Choose another nickname if conflicting."""
        self._nickname += '_'
        conn.nick(self._nickname)

    def on_join(self, conn, event) -> None:
        """Successfully joined channel."""
        joined_nick = event.source.nick
        channel_name = event.target

        if joined_nick == self._nickname:
            logger.info('Joined IRC channel: %s', channel_name)
            irc_channel_joined.send(channel_name=channel_name)

    def on_badchannelkey(self, conn, event) -> None:
        """Channel could not be joined due to wrong password."""
        channel_name = event.arguments[0]
        logger.warning('Cannot join channel %s (bad key).', channel_name)

    def say(self, channel_name: str, text: str) -> None:
        """Say message on channel."""
        self.connection.privmsg(channel_name, text)


class DummyBot:
    """A fake bot that writes messages to STDOUT."""

    def __init__(self, channels: set[IrcChannel]) -> None:
        self.channels = channels

    def start(self) -> None:
        # Fake channel joins.
        for channel in sorted(self.channels):
            irc_channel_joined.send(channel_name=channel.name)

    def say(self, channel_name: str, text: str) -> None:
        logger.debug('%s> %s', channel_name, text)

    def disconnect(self, msg: str) -> None:
        # Mimics `irc.bot.SingleServerIRCBot.disconnect`.
        logger.info('Shutting down bot ...')


def create_bot(config: IrcConfig) -> Union[Bot, DummyBot]:
    """Create and return an IRC bot according to the configuration."""
    if config.server is None:
        logger.info('No IRC server specified; will write to STDOUT instead.')
        return DummyBot(config.channels)

    return Bot(
        config.server,
        config.nickname,
        config.realname,
        config.commands,
        config.channels,
    )