"""IRC support for managing and executing the IdleRPG game. SPDX-FileCopyrightText: © 2024 Brian S. Stephan SPDX-License-Identifier: AGPL-3.0-or-later """ import logging import threading import time import irc.client from django.db.utils import IntegrityError from ircbot.lib import Plugin from idlerpg.models import Character, Game logger = logging.getLogger(__name__) class IdleRPG(Plugin): """Run a game of IdleRPG.""" SLEEP_BETWEEN_LEVEL_CHECKS = 1 LOGIN_COMMAND_PATTERN = r'^LOGIN\s+(?P\S+)\s+(?P\S+)$' LOGOUT_COMMAND_PATTERN = r'^LOGOUT$' REGISTER_COMMAND_PATTERN = r'^REGISTER\s+(?P\S+)\s+(?P\S+)\s+(?P.*)$' REMOVEME_COMMAND_PATTERN = r'^REMOVEME$' STATUS_COMMAND_PATTERN = r'^(STATUS|WHOAMI)$' def __init__(self, bot, connection, event): """Initialize miscellaneous state stuff.""" self.seen_hostmasks = set() super().__init__(bot, connection, event) def start(self): """Set up the handlers and start processing game state.""" self.connection.add_global_handler('join', self.handle_join, -20) self.connection.add_global_handler('kick', self.handle_kick_penalty, -20) self.connection.add_global_handler('nick', self.handle_nick_penalty, -20) self.connection.add_global_handler('part', self.handle_part_penalty, -20) self.connection.add_global_handler('pubmsg', self.handle_message_penalty, -20) self.connection.add_global_handler('pubnotice', self.handle_message_penalty, -20) self.connection.add_global_handler('quit', self.handle_quit_penalty, -20) self.connection.add_global_handler('whoreply', self.handle_whoreply_response, -20) self.connection.reactor.add_global_regex_handler(['privmsg'], self.LOGIN_COMMAND_PATTERN, self.handle_login, -20) self.connection.reactor.add_global_regex_handler(['privmsg'], self.LOGOUT_COMMAND_PATTERN, self.handle_logout, -20) self.connection.reactor.add_global_regex_handler(['privmsg'], self.REGISTER_COMMAND_PATTERN, self.handle_register, -20) self.connection.reactor.add_global_regex_handler(['privmsg'], self.REMOVEME_COMMAND_PATTERN, self.handle_remove, -20) self.connection.reactor.add_global_regex_handler(['privmsg'], self.STATUS_COMMAND_PATTERN, self.handle_status, -20) # get a list of who is in the channel and auto-log them in logger.info("Automatically logging in users already in the channel(s).") for game in Game.objects.filter(active=True): self.connection.who(game.channel.name) # start the thread to check for level ups self.check_for_level_ups = True level_t = threading.Thread(target=self.level_thread) level_t.daemon = True level_t.start() logger.info("Started up the level up checker thread.") super(IdleRPG, self).start() def stop(self): """Tear down handlers, stop checking for new levels, and return to a stable game state.""" self.check_for_level_ups = False self.connection.remove_global_handler('join', self.handle_join) self.connection.remove_global_handler('kick', self.handle_kick_penalty) self.connection.remove_global_handler('nick', self.handle_nick_penalty) self.connection.remove_global_handler('part', self.handle_part_penalty) self.connection.remove_global_handler('pubmsg', self.handle_message_penalty) self.connection.remove_global_handler('pubnotice', self.handle_message_penalty) self.connection.remove_global_handler('quit', self.handle_quit_penalty) self.connection.remove_global_handler('whoreply', self.handle_whoreply_response) self.connection.reactor.remove_global_regex_handler(['privmsg'], self.handle_login) self.connection.reactor.remove_global_regex_handler(['privmsg'], self.handle_logout) self.connection.reactor.remove_global_regex_handler(['privmsg'], self.handle_register) self.connection.reactor.remove_global_regex_handler(['privmsg'], self.handle_remove) self.connection.reactor.remove_global_regex_handler(['privmsg'], self.handle_status) # log everyone out (no penalty) Character.objects.log_out_everyone() # unset the hostmask tracking self.seen_hostmasks = set() logger.info("Reset the set of seen hostmasks.") super(IdleRPG, self).stop() def handle_join(self, connection, event): """Track a character as online if their player joins the game channel.""" self.seen_hostmasks.add(event.source) logger.info("Added %s to the set of seen hostmasks.", event.source) def handle_kick_penalty(self, connection, event): """Penalize characters for their user getting kicked from the channel.""" self.seen_hostmasks.discard(event.source) logger.info("Removed %s from the set of seen hostmasks.", event.source) return self._handle_generic_penalty_and_logout(event.source, event.target, 250, "getting kicked from the game channel", 'time_penalized_kicked') def handle_nick_penalty(self, connection, event): """Penalize characters for changing their nick while in the channel.""" # TODO: figure out how to update the character and seen hostmasks return self._handle_generic_penalty_and_logout(event.source, event.target, 30, "changing their nick", 'time_penalized_nick_change') def handle_part_penalty(self, connection, event): """Penalize characters for their user parting.""" self.seen_hostmasks.discard(event.source) logger.info("Removed %s from the set of seen hostmasks.", event.source) return self._handle_generic_penalty_and_logout(event.source, event.target, 200, "parting the game channel", 'time_penalized_part') def handle_message_penalty(self, connection, event): """Pay attention to messages in the game channel, and penalize talkers.""" hostmask = event.source logger.debug("looking for %s to try to penalize them", hostmask) try: character = Character.objects.get(enabled=True, hostmask=hostmask) logger.debug("found character %s", character) message = event.arguments[0] penalty = character.penalize(len(message), "sending a privmsg to the game channel") character.time_penalized_privmsg += penalty character.save() except Character.DoesNotExist: logger.debug("no character found for %s", hostmask) return def handle_quit_penalty(self, connection, event): """Penalize characters for their user quitting IRC.""" self.seen_hostmasks.discard(event.source) logger.info("Removed %s from the set of seen hostmasks.", event.source) return self._handle_generic_penalty_and_logout(event.source, event.target, 20, "quitting IRC", 'time_penalized_quit') def handle_login(self, connection, event, match): """Log in a character when requested, assuming they're online.""" hostmask = event.source nick = irc.client.NickMask(hostmask).nick try: character = Character.objects.get(enabled=True, name=match.group('name')) character.check_password(match.group('password')) except (Character.DoesNotExist, ValueError): return self.bot.reply(event, "The requested character does not exist, or has been disabled, " "or your password does not match.") if hostmask not in self.seen_hostmasks: return self.bot.reply(event, f"Please join {character.game.channel.name} before logging in.") if character.status != Character.CHARACTER_STATUS_OFFLINE: return self.bot.reply(event, f"Cannot log in {character.name}, either they already are, " "or they are in a state that cannot be modified.") character.log_in(match.group('password'), hostmask) character.save() self.bot.reply(None, f"{character}, is now online from nickname {nick}. " f"Next level at {character.next_level_str()}.", explicit_target=character.game.channel.name) return self.bot.reply(event, f"{character}, has been successfully logged in.") def handle_logout(self, connection, event, match): """Log out a character when requested.""" hostmask = event.source try: character = Character.objects.get(enabled=True, hostmask=hostmask) except Character.DoesNotExist: return self.bot.reply(event, "You do not have a character in a running game.") if character.status != Character.CHARACTER_STATUS_LOGGED_IN: return self.bot.reply(event, f"Cannot log out {character.name}, either they already are, " "or they are in a state that cannot be modified.") character.log_out() penalty = character.penalize(20, "logging out") character.time_penalized_logout += penalty character.save() return self.bot.reply(event, f"{character}, has been successfully logged out.") def handle_register(self, connection, event, match): """Register a character for a user.""" hostmask = event.source nick = irc.client.NickMask(hostmask).nick try: game = Game.objects.get(active=True) except Game.DoesNotExist: return self.bot.reply(event, "No active game exists.") if hostmask not in self.seen_hostmasks: return self.bot.reply(event, f"Please join {game.channel.name} before registering.") try: character = Character.objects.register(match.group('name'), game, match.group('password'), hostmask, match.group('char_class')) except IntegrityError: return self.bot.reply(event, "Registration failed; either you already have a character, " "or the character name is already taken.") self.bot.reply(None, f"{nick}'s newest character, {character.name}, the {character.character_class}, " f"has been summoned! Next level at {character.next_level_str()}.", explicit_target=game.channel.name) return self.bot.reply(event, f"You have registered {character}, and been automatically logged in.") def handle_remove(self, connection, event, match): """Handle the disabling of a character.""" hostmask = event.source try: character = Character.objects.get(enabled=True, hostmask=hostmask) character.enabled = False character.save() return self.bot.reply(event, f"Character {character.name} has been disabled.") except Character.DoesNotExist: return self.bot.reply(event, "No character associated to your hostmask found (try logging in first).") def handle_status(self, connection, event, match): """Handle the request for character/player status.""" hostmask = event.source try: character = Character.objects.get(enabled=True, hostmask=hostmask) self.bot.reply(event, f"{character}, is {hostmask}.") self.bot.reply(event, f"{character.name} is {Character.CHARACTER_STATUSES[character.status]}.") if character.status == Character.CHARACTER_STATUS_LOGGED_IN: self.bot.reply(event, f"{character.name} will level up on {character.next_level_str()}.") except Character.DoesNotExist: self.bot.reply(event, "No character associated to your hostmask found (try logging in first).") def _handle_generic_penalty_and_logout(self, hostmask: str, channel: str, penalty: int, reason: str, penalty_log_attr: str): """Penalize a character and log them out, for a provided reason. Args: hostmask: the hostmask for the character in question channel: the game channel the event occurred in penalty: the penalty to apply reason: the reason for the penalty """ logger.debug("looking for %s to try to penalize them", hostmask) try: character = Character.objects.get(enabled=True, hostmask=hostmask) logger.debug("character found for %s", hostmask) seconds = character.penalize(penalty, reason) log = getattr(character, penalty_log_attr) setattr(character, penalty_log_attr, log + seconds) try: character.log_out() except ValueError: logger.debug("tried to log out %s but they already were", character) character.save() except Character.DoesNotExist: logger.debug("no character found for %s", hostmask) def level_thread(self): """Check for characters who have leveled up, and log as such in the channel.""" while self.check_for_level_ups: time.sleep(self.SLEEP_BETWEEN_LEVEL_CHECKS) self._level_up_characters() def _level_up_characters(self): """Level up characters in the active game, updating their channel.""" game = Game.objects.get(active=True) logger.debug("checking for level ups in %s", game) for character in Character.objects.levelable(game): logger.debug("going to try to level up %s", character) character.level_up() self.bot.reply(None, f"{character.name}, the {character.character_class}, has attained level " f"{character.level}! Next level at {character.next_level_str()}.", explicit_target=game.channel.name) character.save() plugin = IdleRPG