325 lines
16 KiB
Python
325 lines
16 KiB
Python
"""IRC support for managing and executing the IdleRPG game.
|
|
|
|
SPDX-FileCopyrightText: © 2024 Brian S. Stephan <bss@incorporeal.org>
|
|
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<name>\S+)\s+(?P<password>\S+)$'
|
|
LOGOUT_COMMAND_PATTERN = r'^LOGOUT$'
|
|
REGISTER_COMMAND_PATTERN = r'^REGISTER\s+(?P<name>\S+)\s+(?P<password>\S+)\s+(?P<char_class>.*)$'
|
|
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_whoreply_response(self, connection, event):
|
|
"""Track who is in the channel and flag who is online and log them in.
|
|
|
|
We do a /WHO when coming online and joining the game channel, presuming that everyone in it
|
|
*would have* logged themselves in if they were able to, but the bot wasn't online. This may
|
|
log in people who are not, but that doesn't seem like a huge risk to the spirit of the game.
|
|
"""
|
|
user = event.arguments[1]
|
|
host = event.arguments[2]
|
|
nick = event.arguments[4]
|
|
hostmask = f'{nick}!{user}@{host}'
|
|
self.seen_hostmasks.add(hostmask)
|
|
logger.info("Added %s to the set of seen hostmasks.", hostmask)
|
|
for character in Character.objects.filter(enabled=True, status=Character.CHARACTER_STATUS_OFFLINE,
|
|
hostmask=hostmask):
|
|
character.status = Character.CHARACTER_STATUS_LOGGED_IN
|
|
logger.info("Marked %s as logged in.", character)
|
|
character.save()
|
|
|
|
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."""
|
|
logger.debug("kick: %s", event)
|
|
|
|
# somewhat hacky attempt to find the hostmask for the kicked nick
|
|
kicked_nick_hostmask_prefix = f'{event.arguments[0]}!'
|
|
try:
|
|
source = [x for x in self.seen_hostmasks if x.startswith(kicked_nick_hostmask_prefix)][0]
|
|
except IndexError:
|
|
return
|
|
|
|
self.seen_hostmasks.discard(source)
|
|
logger.info("Removed %s from the set of seen hostmasks.", source)
|
|
return self._handle_generic_penalty_and_logout(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."""
|
|
logger.debug("nick change: %s", event)
|
|
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, 30,
|
|
"a nick change",
|
|
'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()
|
|
nick = irc.client.NickMask(hostmask).nick
|
|
self.bot.reply(None, f"{character} has been penalized {penalty} seconds for talking in the game channel.",
|
|
explicit_target=nick)
|
|
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. "
|
|
f"They have been penalized {penalty} seconds.")
|
|
|
|
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()
|
|
nick = irc.client.NickMask(hostmask).nick
|
|
self.bot.reply(None,
|
|
f"{character} has been penalized {penalty} seconds for {reason}, and has been logged out.",
|
|
explicit_target=nick)
|
|
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
|