""" IRC3 Bot Plugin: Periodic Messaging and Nickname Manipulation This plugin for an IRC bot automates periodic messages, nickname changes, and user listing within a channel. It supports starting and stopping these tasks dynamically via bot commands. Features: - Sends a periodic empty message (spam prevention technique). - Changes the bot's nickname periodically to mimic users. - Lists users in the channel periodically in manageable chunks. Author: Zodiac """ import asyncio import irc3 import random import textwrap from irc3.plugins.command import command @irc3.plugin class PeriodicMessagePlugin: """A plugin to periodically send messages, change nicknames, and list users.""" def __init__(self, bot): """ Initialize the plugin with bot reference and default parameters. Args: bot (irc3.IrcBot): The IRC bot instance. """ self.bot = bot self.channel = "" # The IRC channel the bot is operating in self.periodic_message = ' \u2002 \u2003 ' * 500 # Empty message trick self.tasks = [] # Stores running asyncio tasks self.running = False # Flag to control task execution self.original_nick = "" # Store the original bot nickname # Sleep durations for various tasks (in seconds) self.sleep_durations = { '_send_periodic_message': 17, '_change_nick_periodically': 50, '_send_listusers_periodically': 60, } @irc3.event(irc3.rfc.JOIN) def on_join(self, mask, channel, **kwargs): """ Handle the bot joining a channel. Args: mask (str): The user mask. channel (str): The channel name. kwargs (dict): Additional keyword arguments. """ self.channel = channel if not self.running: pass # Uncomment below if auto-starting periodic tasks # self.start_periodic_tasks() def start_periodic_tasks(self): """Start all periodic tasks asynchronously.""" self.running = True self._cancel_tasks() # Ensure no duplicate tasks exist self.tasks = [ asyncio.create_task(self._send_periodic_message()), asyncio.create_task(self._change_nick_periodically()), asyncio.create_task(self._send_listusers_periodically()), ] for task in self.tasks: task.add_done_callback(self._handle_task_done) def _handle_task_done(self, task): """ Handle completed tasks and restart if necessary. Args: task (asyncio.Task): The completed task. """ try: task.result() except asyncio.CancelledError: self.bot.log.info("Task cancelled as expected.") except Exception as e: self.bot.log.error(f"Task error: {e}") finally: if self.running: self.start_periodic_tasks() # Restart tasks if still running async def _send_periodic_message(self): """Send an empty periodic message to the channel.""" try: while self.running: self.bot.privmsg(self.channel, self.periodic_message) self.bot.log.info(f"Message sent to {self.channel}.") await asyncio.sleep(self.sleep_durations['_send_periodic_message']) except asyncio.CancelledError: pass except Exception as e: self.bot.log.error(f"Error sending periodic message: {e}") async def _change_nick_periodically(self): """Change the bot's nickname periodically to mimic a random user.""" try: self.original_nick = self.bot.nick # Store original nickname while self.running: channel_key = self.channel.lower() if channel_key in self.bot.channels: users = list(self.bot.channels[channel_key]) if users: random_user = random.choice(users) new_nick = f"{random_user}_" self.bot.send(f'NICK {new_nick}') self.bot.nick = new_nick self.bot.log.info(f"Nickname changed to: {new_nick}") await asyncio.sleep(self.sleep_durations['_change_nick_periodically']) except asyncio.CancelledError: pass except Exception as e: self.bot.log.error(f"Error changing nickname: {e}") async def _send_listusers_periodically(self): """Send a list of users in the channel periodically.""" try: while self.running: channel_key = self.channel.lower() if channel_key in self.bot.channels: users = list(self.bot.channels[channel_key]) users_msg = ' '.join(users) chunks = textwrap.wrap(users_msg, width=400, break_long_words=False) for chunk in chunks: self.bot.privmsg(self.channel, chunk) await asyncio.sleep(0.0001) # Short delay to avoid flooding self.bot.log.info(f"User list sent to {self.channel}.") await asyncio.sleep(self.sleep_durations['_send_listusers_periodically']) except asyncio.CancelledError: pass except Exception as e: self.bot.log.error(f"Error sending user list: {e}") @command(permission='admin') def stopannoy(self, mask, target, args): """ Stop all periodic tasks and revert nickname. Usage: %%stopannoy """ if mask.nick == self.bot.config.get('owner', ''): self.running = False self._cancel_tasks() if self.original_nick: self.bot.send(f'NICK {self.original_nick}') self.bot.nick = self.original_nick self.bot.log.info(f"Nickname reverted to: {self.original_nick}") return "Periodic tasks stopped." return "Permission denied." @command(permission='admin') async def annoy(self, mask, target, args): """ Start periodic tasks via a command. Usage: %%annoy """ if mask.nick == self.bot.config.get('owner', ''): if not self.running: self.channel = target self.start_periodic_tasks() return "Annoy tasks started." return "Permission denied." def _cancel_tasks(self): """Cancel all running tasks.""" for task in self.tasks: if task and not task.done(): task.cancel() self.tasks = [] @command(permission='admin') async def listusers(self, mask, target, args): """ List all users in the channel and send a formatted message. Usage: %%listusers """ self.channel = target channel_key = self.channel.lower() if channel_key in self.bot.channels: users = list(self.bot.channels[channel_key]) chunk_size = 100 for i in range(0, len(users), chunk_size): user_chunk = users[i : i + chunk_size] users_msg = ' '.join(user_chunk) self.bot.privmsg(self.channel, users_msg) await asyncio.sleep(0.007) # Prevent flooding else: return f"Channel {self.channel} not found."