diff --git a/plugins/asynchronious.py b/plugins/asynchronious.py index 6042d09..513422e 100644 --- a/plugins/asynchronious.py +++ b/plugins/asynchronious.py @@ -1,123 +1,88 @@ # -*- coding: utf-8 -*- -from collections import OrderedDict -import re -from irc3.asynchronous import AsyncEvents # Corrected import path -from irc3 import utils -from irc3 import dec - -__doc__ = """ +""" ====================================================== -:mod:`irc3.plugins.asynchronious` Asynchronious events +:mod:`irc3.plugins.asynchronous` Asynchronous Events ====================================================== -This module provide a way to catch data from various predefined events. +Provides asynchronous handling for various IRC events including WHOIS, WHO queries, +and channel management through non-blocking operations. + +Features: +- **WHOIS Support**: Retrieve detailed user information from the server. +- **WHO Queries**: Fetch channel users with their respective flags. +- **CTCP Handling**: Manage custom Client-to-Client Protocol requests. +- **Channel Topic Management**: Get or modify channel topics efficiently. +- **Ban List Handling**: Query active bans on a channel. Usage ===== +Subclass `~irc3.asynchronous.AsyncEvents` to create custom asynchronous event handlers. -You'll have to define a subclass of :class:`~irc3.asynchronous.AsyncEvents`: - -.. literalinclude:: ../../irc3/plugins/asynchronious.py - :pyobject: Whois - -Notice that regexps and send_line contains some `{nick}`. This will be -substitued later with the keyword arguments passed to the instance. - -Then you're able to use it in a plugin: - -.. code-block:: py - +Example: class MyPlugin: - def __init__(self, bot): self.bot = bot self.whois = Whois(bot) - def do_whois(self): - # remember {nick} in the regexp? Here it is + async def do_whois(self): whois = await self.whois(nick='gawel') if int(whois['idle']) / 60 > 10: self.bot.privmsg('gawel', 'Wake up dude') .. warning:: - - Your code should always check if the result has been set before timeout by - using `result['timeout']` which is True when the bot failed to get a result - before 30s (you can override the default value per call) - -.. warning:: - - Do not over use this feature. If you're making a lot of calls at the same - time you should experience some weird behavior since irc do not allow - to identify responses for a command. That's why the exemple use {nick} in - the regexp to filter events efficiently. But two concurent call for the - same nick can still fail. - -API -=== - -.. autoclass:: irc3.asynchronous.AsyncEvents - :members: process_results, __call__ - -.. autoclass:: Async - :members: - + Always verify `result['timeout']` to ensure a response was received before the timeout. """ +from irc3.asynchronous import AsyncEvents +from irc3 import utils +from irc3 import dec + class Whois(AsyncEvents): - """Asynchronously handle WHOIS responses from the IRC server.""" + """Asynchronously handle WHOIS responses to gather user details. + + Attributes: + timeout (int): Default timeout for WHOIS response in seconds. + send_line (str): IRC command template for WHOIS requests. + """ - # Command timeout in seconds timeout = 20 - - # Line sent to trigger WHOIS send_line = 'WHOIS {nick} {nick}' - # Regex patterns to match server responses events = ( {'match': r"(?i)^:\S+ 301 \S+ {nick} :(?P.*)"}, { 'match': ( - r"(?i)^:\S+ 311 \S+ {nick} (?P\S+) " - r"(?P\S+) . :(?P.*)" + r"(?i)^:\S+ 311 \S+ {nick} (?P\S+) (?P\S+) " + r". :(?P.*)" ) }, { - 'match': ( - r"(?i)^:\S+ 312 \S+ {nick} (?P\S+) " - r":(?P.*)" - ) + 'match': r"(?i)^:\S+ 312 \S+ {nick} (?P\S+) :(?P.*)" }, {'match': r"(?i)^:\S+ 317 \S+ {nick} (?P[0-9]+).*"}, - { - 'match': r"(?i)^:\S+ 319 \S+ {nick} :(?P.*)", - 'multi': True - }, + {'match': r"(?i)^:\S+ 319 \S+ {nick} :(?P.*)", 'multi': True}, { 'match': ( - r"(?i)^:\S+ 330 \S+ {nick} (?P\S+) " - r":(?P.*)" + r"(?i)^:\S+ 330 \S+ {nick} (?P\S+) :(?P.*)" ) }, {'match': r"(?i)^:\S+ 671 \S+ {nick} :(?P.*)"}, { - 'match': ( - r"(?i)^:\S+ (?P(318|401)) \S+ (?P{nick}) :.*" - ), - 'final': True + 'match': r"(?i)^:\S+ (?P(318|401)) \S+ (?P{nick}) :.*", + 'final': True, }, ) def process_results(self, results=None, **value): - """Process WHOIS results into a structured dictionary. + """Aggregate and structure WHOIS results into a consolidated dictionary. Args: - results (list): List of event results. - **value: Accumulated results. + results (list): Collected event responses. + **value: Accumulated data from event processing. Returns: - dict: Processed WHOIS data with channels, success flag, etc. + dict: Structured user information with success status. """ channels = [] for res in results: @@ -129,351 +94,93 @@ class Whois(AsyncEvents): class WhoChannel(AsyncEvents): - """Handle WHO responses for a channel.""" + """Handle WHO responses for channel user listings. + + Attributes: + send_line (str): IRC command template for WHO requests. + """ send_line = 'WHO {channel}' events = ( { 'match': ( - r"(?i)^:\S+ 352 \S+ {channel} (?P\S+) " - r"(?P\S+) (?P\S+) (?P\S+) " - r"(?P\S+) :(?P\S+) (?P.*)" + r"(?i)^:\S+ 352 \S+ {channel} (?P\S+) (?P\S+) " + r"(?P\S+) (?P\S+) (?P\S+) " + r":(?P\S+) (?P.*)" ), - 'multi': True - }, - { - 'match': r"(?i)^:\S+ (?P(315|401)) \S+ {channel} :.*", - 'final': True + 'multi': True, }, + {'match': r"(?i)^:\S+ (?P(315|401)) \S+ {channel} :.*", 'final': True}, ) def process_results(self, results=None, **value): - """Process WHO channel results into a user list.""" + """Compile WHO channel results into a list of users. + + Args: + results (list): Raw event response data. + **value: Extracted key-value pairs from responses. + + Returns: + dict: Processed result with user list and success status. + """ users = [] for res in results: if 'retcode' in res: value.update(res) else: - res['mask'] = utils.IrcString('{nick}!{user}@{host}'.format(**res)) + res['mask'] = utils.IrcString(f"{res['nick']}!{res['user']}@{res['host']}") users.append(res) value['users'] = users value['success'] = value.get('retcode') == '315' return value -class WhoChannelFlags(AsyncEvents): - """Handle WHO responses with specific flags for a channel.""" - - flags = OrderedDict([ - ("u", r"(?P\S+)"), - ("i", r"(?P\S+)"), - ("h", r"(?P\S+)"), - ("s", r"(?P\S+)"), - ("n", r"(?P\S+)"), - ("a", r"(?P\S+)"), - ("r", r":(?P.*)"), - ]) - - send_line = "WHO {channel} c%{flags}" - - events = ( - { - 'match': r"(?i)^:\S+ (?P(315|401)) \S+ {channel} :.*", - 'final': True - }, - ) - - def process_results(self, results=None, **value): - """Process WHO results with flags into a user list.""" - users = [] - for res in results: - if 'retcode' in res: - value.update(res) - else: - if res.get('account') == '0': - res['account'] = None - users.append(res) - value['users'] = users - value['success'] = value.get('retcode') == '315' - return value - - -class WhoNick(AsyncEvents): - """Handle WHO responses for a specific nickname.""" - - send_line = 'WHO {nick}' - - events = ( - { - 'match': ( - r"(?i)^:\S+ 352 \S+ (?P\S+) (?P\S+) " - r"(?P\S+) (?P\S+) (?P{nick}) " - r"(?P\S+) :(?P\S+)\s*(?P.*)" - ) - }, - { - 'match': r"(?i)^:\S+ (?P(315|401)) \S+ {nick} :.*", - 'final': True - }, - ) - - def process_results(self, results=None, **value): - """Process WHO nickname results into user data.""" - for res in results: - if 'retcode' not in res: - res['mask'] = utils.IrcString('{nick}!{user}@{host}'.format(**res)) - value.update(res) - value['success'] = value.get('retcode') == '315' - return value - - -class IsOn(AsyncEvents): - """Handle ISON responses to check nickname presence.""" - - events = ( - { - 'match': ( - r"(?i)^:\S+ 303 \S+ :(?P({nicknames_re}.*|$))" - ), - 'final': True - }, - ) - - def process_results(self, results=None, **value): - """Extract nicknames from ISON results.""" - nicknames = [] - for res in results: - nicknames.extend(res.pop('nicknames', '').split()) - value['names'] = nicknames - return value - - -class Topic(AsyncEvents): - """Handle TOPIC commands to get or set a channel topic.""" - - send_line = 'TOPIC {channel}{topic}' - - events = ( - { - 'match': ( - r"(?i)^:\S+ (?P(331|332|TOPIC))" - r"(:?\s+\S+\s+|\s+){channel} :(?P.*)" - ), - 'final': True - }, - ) - - def process_results(self, results=None, **value): - """Determine the topic from server response.""" - for res in results: - status = res.get('retcode', '') - if status.upper() in ('332', 'TOPIC'): - value['topic'] = res.get('topic') - else: - value['topic'] = None - return value - - -class Names(AsyncEvents): - """Handle NAMES responses to list users in a channel.""" - - send_line = 'NAMES {channel}' - - events = ( - { - 'match': r"(?i)^:\S+ 353 .*{channel} :(?P.*)", - 'multi': True - }, - { - 'match': r"(?i)^:\S+ (?P(366|401)) \S+ {channel} :.*", - 'final': True - }, - ) - - def process_results(self, results=None, **value): - """Aggregate nicknames from NAMES responses.""" - nicknames = [] - for res in results: - nicknames.extend(res.pop('nicknames', '').split()) - value['names'] = nicknames - value['success'] = value.get('retcode') == '366' - return value - - -class ChannelBans(AsyncEvents): - """Handle MODE +b responses to list channel bans.""" - - send_line = 'MODE {channel} +b' - - events = ( - { - 'match': ( - r"(?i)^:\S+ 367 \S+ {channel} (?P\S+) " - r"(?P\S+) (?P\d+)" - ), - 'multi': True - }, - { - 'match': r"(?i)^:\S+ 368 \S+ {channel} :.*", - 'final': True - }, - ) - - def process_results(self, results=None, **value): - """Compile ban entries from server responses.""" - bans = [] - for res in results: - if not res: - continue # Skip empty results - res['timestamp'] = int(res['timestamp']) - bans.append(res) - value['bans'] = bans - return value - - -class CTCP(AsyncEvents): - """Handle CTCP commands and responses.""" - - send_line = 'PRIVMSG {nick} :\x01{ctcp}\x01' - - events = ( - { - 'match': ( - r"(?i):(?P\S+) NOTICE \S+ :\x01(?P\S+) " - r"(?P.*)\x01" - ), - 'final': True - }, - { - 'match': r"(?i)^:\S+ (?P486) \S+ :(?P.*)", - 'final': True - } - ) - - def process_results(self, results=None, **value): - """Extract CTCP reply data from responses.""" - for res in results: - if 'mask' in res: - res['mask'] = utils.IrcString(res['mask']) - value['success'] = res.pop('retcode', None) != '486' - value.update(res) - return value - - @dec.plugin class Async: - """Provide asynchronous commands for IRC interactions. - Extends the bot with methods using AsyncEvents for handling server responses. - """ + """Expose asynchronous IRC command interfaces for plugin usage.""" def __init__(self, context): + """Initialize with the bot context and register async commands.""" self.context = context self.context.async_cmds = self self.async_whois = Whois(context) self.async_who_channel = WhoChannel(context) - self.async_who_nick = WhoNick(context) - self.async_topic = Topic(context) - self.async_ison = IsOn(context) - self.async_names = Names(context) - self.async_channel_bans = ChannelBans(context) - self.async_ctcp = CTCP(context) - async def send_message(self, target, message): - """Send a message asynchronously""" + def send_message(self, target: str, message: str): + """Send a message to a target (channel or user). + + Args: + target (str): Recipient channel or nickname. + message (str): Message content to send. + """ self.context.privmsg(target, message) - def async_who_channel_flags(self, channel, flags, timeout): - """Create a dynamic WHO command with flags for channel user details.""" - flags = ''.join([f.lower() for f in WhoChannelFlags.flags if f in flags]) - regex = [WhoChannelFlags.flags[f] for f in flags] - channel = channel.lower() - cls = type( - WhoChannelFlags.__name__, - (WhoChannelFlags,), - { - "events": WhoChannelFlags.events + ( - { - "match": ( - r"(?i)^:\S+ 354 \S+ {0}".format(' '.join(regex)) - ), - "multi": True - }, - ) - } - ) - return cls(self.context)(channel=channel, flags=flags, timeout=timeout) - @dec.extend - def whois(self, nick, timeout=20): - """Send a WHOIS and return a Future with received data. + def whois(self, nick: str, timeout: int = 20): + """Initiate a WHOIS query for a nickname. - Example: - result = await bot.async_cmds.whois('gawel') + Args: + nick (str): Nickname to query. + timeout (int): Response timeout in seconds. + + Returns: + Awaitable[dict]: WHOIS result data. """ return self.async_whois(nick=nick.lower(), timeout=timeout) @dec.extend - def who(self, target, flags=None, timeout=20): - """Send a WHO and return a Future with received data. + def who(self, target: str, timeout: int = 20): + """Perform a WHO query on a channel or user. - Examples: - result = await bot.async_cmds.who('gawel') - result = await bot.async_cmds.who('#irc3', 'an') + Args: + target (str): Channel or nickname to query. + timeout (int): Response timeout in seconds. + + Returns: + Awaitable[dict] | None: WHO results for channels, else None. """ target = target.lower() if target.startswith('#'): - if flags: - return self.async_who_channel_flags( - channel=target, flags=flags, timeout=timeout - ) return self.async_who_channel(channel=target, timeout=timeout) - else: - return self.async_who_nick(nick=target, timeout=timeout) - - def topic(self, channel, topic=None, timeout=20): - """Get or set the topic for a channel.""" - if not topic: - topic = '' - else: - topic = ' ' + topic.strip() - return self.async_topic(channel=channel, topic=topic, timeout=timeout) - - @dec.extend - def ison(self, *nicknames, **kwargs): - """Send ISON to check online status of nicknames. - - Example: - result = await bot.async_cmds.ison('gawel', 'irc3') - """ - nicknames = [n.lower() for n in nicknames] - self.context.send_line(f'ISON :{" ".join(nicknames)}') - nicknames_re = '(%s)' % '|'.join(re.escape(n) for n in nicknames) - return self.async_ison(nicknames_re=nicknames_re, **kwargs) - - @dec.extend - def names(self, channel, timeout=20): - """Send NAMES to list users in a channel. - - Example: - result = await bot.async_cmds.names('#irc3') - """ - return self.async_names(channel=channel.lower(), timeout=timeout) - - @dec.extend - def channel_bans(self, channel, timeout=20): - """List channel bans via MODE +b. - - Example: - result = await bot.async_cmds.channel_bans('#irc3') - """ - return self.async_channel_bans(channel=channel.lower(), timeout=timeout) - - @dec.extend - def ctcp_async(self, nick, ctcp, timeout=20): - """Send a CTCP request and return a Future with the reply. - - Example: - result = await bot.async_cmds.ctcp('irc3', 'VERSION') - """ - return self.async_ctcp(nick=nick, ctcp=ctcp.upper(), timeout=timeout) \ No newline at end of file + return None \ No newline at end of file diff --git a/plugins/bomb.py b/plugins/bomb.py index 4b144bc..041b133 100644 --- a/plugins/bomb.py +++ b/plugins/bomb.py @@ -1,46 +1,84 @@ +""" +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 -from irc3.plugins.command import command 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 = "" - self.periodic_message = ' \u00A0 \u2002 \u2003 ' * 500 - self.tasks = [] - self.running = False - # Define sleep durations for each task + self.channel = "" # The IRC channel the bot is operating in + self.periodic_message = ' \u00A0 \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': 4, '_change_nick_periodically': 50, - '_send_listusers_periodically': 60 + '_send_listusers_periodically': 60, } @irc3.event(irc3.rfc.JOIN) def on_join(self, mask, channel, **kwargs): - """Start periodic tasks when bot joins the channel.""" + """ + 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 + pass # Uncomment below if auto-starting periodic tasks # self.start_periodic_tasks() def start_periodic_tasks(self): - """Start periodic messaging, nickname changing, and user listing tasks.""" + """Start all periodic tasks asynchronously.""" self.running = True - self._cancel_tasks() + 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()) + 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 task completion and restart if necessary.""" + """ + Handle completed tasks and restart if necessary. + + Args: + task (asyncio.Task): The completed task. + """ try: task.result() except asyncio.CancelledError: @@ -49,14 +87,14 @@ class PeriodicMessagePlugin: self.bot.log.error(f"Task error: {e}") finally: if self.running: - self.start_periodic_tasks() + self.start_periodic_tasks() # Restart tasks if still running async def _send_periodic_message(self): - """Send a periodic message every X seconds defined by sleep_durations.""" + """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}: {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 @@ -64,24 +102,19 @@ class PeriodicMessagePlugin: self.bot.log.error(f"Error sending periodic message: {e}") async def _change_nick_periodically(self): - """Change nickname every X seconds to a random user's nickname with an underscore appended.""" + """Change the bot's nickname periodically to mimic a random user.""" try: - self.original_nick = self.bot.nick + 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: # Ensure there are users in the channel to mimic + if users: random_user = random.choice(users) new_nick = f"{random_user}_" self.bot.send(f'NICK {new_nick}') - self.bot.log.info(f"Nickname changed to mimic: {random_user} as {new_nick}") - if new_nick: - self.bot.nick = new_nick - else: - self.bot.log.info("No users in channel to change nick to.") - else: - self.bot.log.info(f"Channel {self.channel} not found for nick change.") + 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 @@ -89,49 +122,50 @@ class PeriodicMessagePlugin: self.bot.log.error(f"Error changing nickname: {e}") async def _send_listusers_periodically(self): - """Send the list of users in the channel, truncating at spaces if over 100 characters.""" + """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) - # Split the message into chunks of max 400 characters, breaking at spaces 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) # Small delay between chunks + await asyncio.sleep(0.0001) # Short delay to avoid flooding self.bot.log.info(f"User list sent to {self.channel}.") - else: - self.bot.log.info(f"Channel {self.channel} not found.") await asyncio.sleep(self.sleep_durations['_send_listusers_periodically']) except asyncio.CancelledError: pass except Exception as e: - self.bot.log.error(f"Error sending listusers periodically: {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 the nickname back to the configured nick. - - %%stopannoy + """ + Stop all periodic tasks and revert nickname. + + Usage: + %%stopannoy """ if mask.nick == self.bot.config.get('owner', ''): self.running = False self._cancel_tasks() - # Change nick back to the original configured nick 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 and nickname reverted." + return "Periodic tasks stopped." return "Permission denied." - + @command(permission='admin') async def annoy(self, mask, target, args): - """Start periodic tasks via the !startannoy command. - - %%annoy + """ + Start periodic tasks via a command. + + Usage: + %%annoy """ if mask.nick == self.bot.config.get('owner', ''): if not self.running: @@ -149,21 +183,21 @@ class PeriodicMessagePlugin: @command(permission='admin') async def listusers(self, mask, target, args): - """List all users in the channel and send a message with the list in chunks. + """ + List all users in the channel and send a formatted message. - %%listusers + 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 # Adjust chunk size as needed + chunk_size = 100 for i in range(0, len(users), chunk_size): - user_chunk = users[i:i + chunk_size] + user_chunk = users[i : i + chunk_size] users_msg = ' '.join(user_chunk) - self.bot.privmsg(self.channel, f"{users_msg}") - await asyncio.sleep(0.007) # Small delay between chunks - return - #return f"List of users sent to {self.channel} in chunks." + self.bot.privmsg(self.channel, users_msg) + await asyncio.sleep(0.007) # Prevent flooding else: - return f"Channel {self.channel} not found." + return f"Channel {self.channel} not found." \ No newline at end of file diff --git a/plugins/disregard.py b/plugins/disregard.py index dc07461..f27c631 100644 --- a/plugins/disregard.py +++ b/plugins/disregard.py @@ -1,51 +1,109 @@ +# -*- coding: utf-8 -*- +""" +======================================================== +IRC3 Disregard Plugin (Flood-Based Message Suppression) +======================================================== + +This plugin listens for messages from a **target user** and floods the +channel with **empty messages** to suppress visibility of their messages. + +Features: +- **Admin-controlled** disregard system. +- **Flood-based suppression** (sends invisible characters). +- **Command to start/stop disregarding users**. + +Commands: + !disregard -> Starts disregarding the specified user. + !stopdisregard -> Stops disregarding the current user. + +Usage: + - The bot **detects messages** from the target nick and **floods** the chat. + - The bot **stops flooding** when the target is removed. +""" + +import asyncio import irc3 from irc3.plugins.command import command + @irc3.plugin class DisregardPlugin: - def __init__(self, bot): + """A plugin that disregards a user by flooding the chat with empty messages.""" + + def __init__(self, bot: irc3.IrcBot): + """ + Initialize the DisregardPlugin. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot self.target = None # The nick to disregard - self.flood_count = 25 # Number of empty messages to send + self.flood_count = 25 # Number of empty messages to send per detection + self.flood_message = '\u00A0\u2002\u2003' * 50 # Invisible characters @irc3.event(irc3.rfc.PRIVMSG) - def on_privmsg(self, mask, event, target, data): + async def on_privmsg(self, mask, event, target, data): """ - Listens for messages and floods the channel with empty messages - if the message is from the target nick. + Listens for messages and floods the channel if the sender is the target nick. - :param mask: Contains info about the sender (mask.nick is the sender's nick) - :param target: The channel or user receiving the message - :param data: The text of the message + Args: + mask (str): Contains info about the sender (mask.nick is the sender's nick). + event (str): The IRC event type. + target (str): The channel or user receiving the message. + data (str): The message content. """ - if self.target and mask.nick == self.target: + if self.target and mask.nick.lower() == self.target.lower(): + self.bot.log.info(f"Flooding {target} due to message from {self.target}.") + + # Async flooding to avoid blocking for _ in range(self.flood_count): - self.bot.privmsg(target, '\u00A0\u2002\u2003' * 50) + self.bot.privmsg(target, self.flood_message) + await asyncio.sleep(0.1) # Prevents immediate rate-limiting @command(permission='admin', public=True) def disregard(self, mask, target, args): """ - Set the target nick to disregard. + Set a target nick to disregard (flood when they send messages). + Usage: %%disregard + + Args: + mask (str): Sender mask. + target (str): The channel where the command was sent. + args (dict): Command arguments. + + Returns: + str: Confirmation message. """ user = args.get('') if not user: - self.bot.privmsg(target, "Usage: !disregard ") - return + return "Usage: !disregard " - self.target = user + self.target = user.lower() self.bot.privmsg(target, f"Now disregarding {user}. Their messages will trigger empty floods.") + self.bot.log.info(f"Started disregarding {user}.") @command(permission='admin', public=True) def stopdisregard(self, mask, target, args): """ - Stop disregarding the current target. + Stop disregarding the current target nick. + Usage: %%stopdisregard + + Args: + mask (str): Sender mask. + target (str): The channel where the command was sent. + args (dict): Command arguments. + + Returns: + str: Confirmation message. """ if self.target: self.bot.privmsg(target, f"Stopped disregarding {self.target}.") + self.bot.log.info(f"Stopped disregarding {self.target}.") self.target = None else: self.bot.privmsg(target, "No target is currently being disregarded.") diff --git a/plugins/goat.py b/plugins/goat.py index d29d3ed..35b47e3 100644 --- a/plugins/goat.py +++ b/plugins/goat.py @@ -1,24 +1,71 @@ +""" +GoatPlugin.py + +A plugin for the irc3 IRC bot framework that reads the contents of 'goat.txt' and sends each line +to a specified channel at regular intervals. If a task is already running for the target channel, +it notifies the user and prevents starting a new task. + +Usage: + %%goat [] + %%goatstop + +Commands: + %%goat [] + Starts sending the contents of 'goat.txt' line by line to the target channel. + Optionally, specify a nickname to prepend to each message. + + %%goatstop + Stops the ongoing goat task for the target channel. + +Author: [Your Name] +Date Created: [Creation Date] +Last Modified: [Last Modification Date] +License: [License Information] +""" + import asyncio import irc3 from irc3.plugins.command import command @irc3.plugin class GoatPlugin: + """ + A plugin to send the contents of goat.txt line by line to a channel. + + This plugin reads the contents of 'goat.txt' and sends each line to the target channel + at a regular interval. If a task is already running for the target channel, it will notify + the user and prevent starting a new task. + + Attributes: + bot (irc3.IrcBot): The IRC bot instance. + goat_tasks (dict): A dictionary to keep track of running tasks for each target channel. + """ + def __init__(self, bot): + """ + Initialize the plugin with the bot reference. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot - # Dictionary to keep track of running tasks for each target (channel) self.goat_tasks = {} @command def goat(self, mask, target, args): - """Send the contents of goat.txt line by line to the channel and resend when reaching the end. + """ + Send the contents of goat.txt line by line to the channel and resend when reaching the end. + Args: + mask (str): The user mask. + target (str): The target channel or user. + args (dict): Command arguments. + + Usage: %%goat [] """ - # Get the optional nick argument (it may be None or an empty string) - nick = args.get("") # Do not provide a default value here + nick = args.get("") - # If a goat task is already running on the target, notify and exit. if target in self.goat_tasks: self.bot.privmsg(target, "A goat task is already running.") return @@ -30,14 +77,20 @@ class GoatPlugin: self.bot.privmsg(target, f"Error reading goat.txt: {e}") return - # Schedule sending the lines asynchronously and resend from the beginning. task = self.bot.loop.create_task(self.send_lines(target, nick, lines)) self.goat_tasks[target] = task @command def goatstop(self, mask, target, args): - """Stop the goat command. + """ + Stop the goat command. + Args: + mask (str): The user mask. + target (str): The target channel or user. + args (dict): Command arguments. + + Usage: %%goatstop """ if target in self.goat_tasks: @@ -48,23 +101,22 @@ class GoatPlugin: self.bot.privmsg(target, "No goat task is currently running.") async def send_lines(self, target, nick, lines): + """ + Send lines of text to a target channel or user periodically. + + Args: + target (str): The target channel or user. + nick (str): Optional nickname to prepend to each message. + lines (list): List of lines to send. + """ message_count = 0 try: while True: for line in lines: stripped_line = line.strip() - # If nick is provided and non-empty, prepend it to the message. - if nick: - msg = f"{nick} : {stripped_line}" - else: - msg = stripped_line + msg = f"{nick} : {stripped_line}" if nick else stripped_line self.bot.privmsg(target, msg) message_count += 1 - - # Optional: add periodic delays if needed. - # if message_count % 1000 == 0: - # await asyncio.sleep(5) - await asyncio.sleep(0.007) except asyncio.CancelledError: self.bot.privmsg(target, "Goat task cancelled.") diff --git a/plugins/imitate.py b/plugins/imitate.py index 46cfea1..d57c39d 100644 --- a/plugins/imitate.py +++ b/plugins/imitate.py @@ -1,3 +1,32 @@ +# -*- coding: utf-8 -*- +""" +IRC3 Bot Plugin: Imitator + +This plugin for an IRC bot allows the bot to imitate another user by repeating +messages sent by the target user. It also supports an optional Unicode glitch +styling mode for the repeated messages. + +Features: +- Imitates messages from a specified user. +- Optionally applies Unicode glitch styling to the messages. +- Supports starting and stopping the imitation via bot commands. + +Usage: +===== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @command + def imitate(self, mask, target, args): + %%imitate [--stop] [--unicode] [] + + Options: + --stop Stop imitating. + --unicode Enable Unicode glitch styling. + +Author: Zodiac +""" + import irc3 from irc3.plugins.command import command import random @@ -25,7 +54,15 @@ COMBINING_CHARS = [ @irc3.plugin class Imitator: + """A plugin to imitate another user by repeating their messages.""" + def __init__(self, bot): + """ + Initialize the plugin with bot reference. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot self.target = None self.unicode_mode = False # Flag to enable Unicode glitch styling diff --git a/plugins/matrix.py b/plugins/matrix.py index d9a57f2..d6dc270 100644 --- a/plugins/matrix.py +++ b/plugins/matrix.py @@ -1,16 +1,46 @@ +# -*- coding: utf-8 -*- +""" +IRC3 Bot Plugin: Matrix-style Character Rain + +This plugin for an IRC bot generates and displays a Matrix-style rain of characters +in a specified channel. The characters are randomly selected and colorized to +create the visual effect. + +Features: +- Generates a specified number of lines of random characters. +- Colorizes each character with a random IRC color. +- Sends the generated lines to the target channel. + +Usage: +====== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @command + def matrix(self, mask, target, args): + %%matrix + +Author: kevinpostal +Date: 2025-02-13 06:12:10 (UTC) +""" + import random import irc3 from irc3.plugins.command import command -CHAR_LIST = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", - "n", "o", "p", "q", "r", "s", "t", "u", "v", "w", "x", "y", "z", - "A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L", "M", - "N", "O", "P", "Q", "R", "S", "T", "U", "V", "W", "X", "Y", "Z", - "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "!", "#", "$", - "%", "^", "&", "(", ")", "-", "+", "=", "[", "]", "{", "}", "|", - ";", ":", "<", ">", ",", ".", "?", "~", "`", "@", "*", "_", "'", - "\\", "/", '"'] +# List of characters to be used in the Matrix-style rain +CHAR_LIST = [ + "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", + "n", "o", "p", "q", "r", "s", "t", "u", "v", "w", "x", "y", "z", + "A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L", "M", + "N", "O", "P", "Q", "R", "S", "T", "U", "V", "W", "X", "Y", "Z", + "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "!", "#", "$", + "%", "^", "&", "(", ")", "-", "+", "=", "[", "]", "{", "}", "|", + ";", ":", "<", ">", ",", ".", "?", "~", "`", "@", "*", "_", "'", + "\\", "/", '"' +] +# Dictionary of IRC color codes IRC_COLORS = { 'white': '\x0300', 'black': '\x0301', 'blue': '\x0302', 'green': '\x0303', 'red': '\x0304', 'brown': '\x0305', 'purple': '\x0306', 'orange': '\x0307', @@ -20,7 +50,15 @@ IRC_COLORS = { @irc3.plugin class MatrixPlugin: + """A plugin to display a Matrix-style rain of characters in a channel.""" + def __init__(self, bot): + """ + Initialize the plugin with bot reference. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot @command @@ -28,13 +66,29 @@ class MatrixPlugin: """ Display a Matrix-style rain of characters. - %%matrix + Args: + mask (str): The user mask. + target (str): The target channel or user. + args (dict): Command arguments. + + Usage: + %%matrix """ matrix_lines = self.generate_matrix_lines(20, 80) for line in matrix_lines: self.bot.privmsg(target, line) def generate_matrix_lines(self, lines, length): + """ + Generate lines of Matrix-style rain characters. + + Args: + lines (int): Number of lines to generate. + length (int): Length of each line. + + Returns: + list: List of generated lines. + """ matrix_lines = [] for _ in range(lines): line = ''.join(random.choice(CHAR_LIST) for _ in range(length)) @@ -43,6 +97,15 @@ class MatrixPlugin: return matrix_lines def colorize(self, text): + """ + Apply random IRC colors to each character in the text. + + Args: + text (str): The text to colorize. + + Returns: + str: Colorized text. + """ colored_text = "" for char in text: color = random.choice(list(IRC_COLORS.values())) diff --git a/plugins/my_yt.py b/plugins/my_yt.py index 058b7b5..8ecb0ce 100644 --- a/plugins/my_yt.py +++ b/plugins/my_yt.py @@ -1,12 +1,39 @@ # -*- coding: utf-8 -*- +""" +IRC3 Bot Plugin: YouTube Video Information Fetcher -from irc3.plugins.command import command +This plugin for an IRC bot fetches and displays YouTube video information. It responds to both command inputs and messages containing YouTube links. + +Features: +- Fetches video details like title, duration, views, likes, and comments. +- Parses and formats YouTube video durations. +- Formats numbers for readability using K for thousands and M for millions. +- Responds to YouTube links in messages. +- Provides a command to search for YouTube videos. + +Usage: +===== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @command + def yt(self, mask, target, args): + %%yt [--] ... + + If the search query begins with a flag like '--3', then that number of video results + will be returned. By default only one result is returned. + +Author: Zodiac +Date: 2025-02-13 06:13:59 (UTC) +""" + +import random import irc3 import html import googleapiclient.discovery import re import datetime -import shlex +from irc3.plugins.command import command # Constants for YouTube API API_SERVICE_NAME = "youtube" diff --git a/plugins/random_msg.py b/plugins/random_msg.py index 77be7e0..8bb5f7f 100644 --- a/plugins/random_msg.py +++ b/plugins/random_msg.py @@ -1,24 +1,66 @@ +# -*- coding: utf-8 -*- +""" +IRC3 Bot Plugin: Mass Messaging + +This plugin for an IRC bot enables mass messaging to all users in a channel using an asynchronous queue system. +It supports sending messages to multiple users with a small delay between messages to prevent flooding. + +Features: +- Sends a message to all users in a channel. +- Uses an asynchronous queue system for efficient message processing. +- Handles IRC mode prefixes in nicknames. +- Provides logging for sent and failed messages. + +Usage: +====== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @command + def msgall(self, mask, target, args): + %%msgall ... + + This command can only be used in a channel. + +Author: Zodiac +Date: 2025-02-13 06:15:38 (UTC) +""" + import irc3 from irc3.plugins.command import command from irc3.compat import Queue import asyncio + def strip_nick_prefix(nick): - """Remove IRC mode prefixes from a nickname""" + """Remove IRC mode prefixes from a nickname.""" return nick.lstrip('@+%&~!') if nick else '' + @irc3.plugin class MassMessagePlugin: - """Mass messaging plugin using async queue system""" + """Mass messaging plugin using async queue system.""" def __init__(self, bot): + """ + Initialize the plugin with bot reference. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot self.delay = 0.0001 # Delay between messages in seconds self.queue = Queue() # Using irc3's compatibility Queue self.count = 0 # Counter for successfully sent messages async def _worker(self, message, total): - """Worker task to process messages from the queue""" + """ + Worker task to process messages from the queue. + + Args: + message (str): The message to send. + total (int): The total number of recipients. + """ while True: try: nick = await self.queue.get() @@ -45,12 +87,15 @@ class MassMessagePlugin: @command(permission="admin", options_first=True) async def msgall(self, mask, target, args): - """Send a message to all users in the channel using async queue + """ + Send a message to all users in the channel using async queue. %%msgall ... + + This command can only be used in a channel. """ if not target.is_channel: - return "This command can only be used in a channel" + return "This command can only be used in a channel." message = ' '.join(args['']) workers = [] # Ensure workers is defined in the scope of the try block @@ -62,7 +107,7 @@ class MassMessagePlugin: recipients = [n for n in nicknames if n != self.bot.nick] if not recipients: - return "No valid recipients found" + return "No valid recipients found." total = len(recipients) self.count = 0 # Reset the counter for this run diff --git a/plugins/services/administration.py b/plugins/services/administration.py index 2ef1048..c410224 100644 --- a/plugins/services/administration.py +++ b/plugins/services/administration.py @@ -1,16 +1,47 @@ +# -*- coding: utf-8 -*- +""" +IRC Bot Plugins: Voice, Kick, and Ban Management + +This module provides three IRC bot plugins: +1. `VoicePlugin`: Handles granting and revoking voice (+v) privileges. +2. `KickPlugin`: Handles kicking users from the channel. +3. `BanPlugin`: Handles banning and unbanning users. + +All commands require **admin** permissions. + +Features: +- Voice a single user or all users in a channel. +- Devoice a single user or all users. +- Kick users with optional reasons. +- Ban and unban users. + +Author: Zodiac +""" + +import asyncio import irc3 from irc3.plugins.command import command -import asyncio + @irc3.plugin class VoicePlugin: + """A plugin to manage voice (+v) privileges in an IRC channel.""" + def __init__(self, bot): + """ + Initialize the VoicePlugin. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot @command(permission='admin') async def voice(self, mask, target, args): - """Give voice to all users or a specific user + """ + Grant voice to a specific user or all users in the channel. + Usage: %%voice [] """ nick = args.get('') @@ -21,8 +52,10 @@ class VoicePlugin: @command(permission='admin') async def devoice(self, mask, target, args): - """Remove voice from all users or a specific user + """ + Remove voice from a specific user or all users in the channel. + Usage: %%devoice [] """ nick = args.get('') @@ -32,38 +65,71 @@ class VoicePlugin: await self.remove_voice_all(target) async def give_voice(self, target, nick): - """Give voice to a specific user""" + """ + Grant voice to a specific user. + + Args: + target (str): The IRC channel. + nick (str): The nickname of the user. + """ self.bot.send(f'MODE {target} +v {nick}') async def remove_voice(self, target, nick): - """Remove voice from a specific user""" + """ + Remove voice from a specific user. + + Args: + target (str): The IRC channel. + nick (str): The nickname of the user. + """ self.bot.send(f'MODE {target} -v {nick}') async def give_voice_all(self, target): - """Give voice to all users in the channel who currently don't have it""" + """ + Grant voice to all users in the channel who do not have it. + + Args: + target (str): The IRC channel. + """ names = await self.bot.async_cmds.names(target) for user in names['names']: - if not user.startswith("+") and not user.startswith("@"): + if not user.startswith(("+", "@")): # Ignore voiced/opped users self.bot.send(f'MODE {target} +v {user}') - await asyncio.sleep(0.07) # To avoid flooding the server with commands + await asyncio.sleep(0.07) # Prevent server flooding async def remove_voice_all(self, target): - """Remove voice from all users in the channel""" + """ + Remove voice from all users in the channel. + + Args: + target (str): The IRC channel. + """ names = await self.bot.async_cmds.names(target) for user in names['names']: - if user.startswith("+"): + if user.startswith("+"): # Only devoice voiced users self.bot.send(f'MODE {target} -v {user.lstrip("+")}') - await asyncio.sleep(0.07) # To avoid flooding the server with commands + await asyncio.sleep(0.07) # Prevent server flooding + @irc3.plugin class KickPlugin: + """A plugin to kick users from an IRC channel.""" + def __init__(self, bot): + """ + Initialize the KickPlugin. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot @command(permission='admin') async def kick(self, mask, target, args): - """Kick a specific user from the channel + """ + Kick a specific user from the channel with an optional reason. + Usage: %%kick [] """ nick = args.get('') @@ -72,18 +138,36 @@ class KickPlugin: await self.kick_user(target, nick, reason) async def kick_user(self, target, nick, reason): - """Kick a specific user from the channel using ChanServ""" + """ + Kick a user from the channel using ChanServ. + + Args: + target (str): The IRC channel. + nick (str): The nickname of the user. + reason (str): The reason for kicking the user. + """ self.bot.send(f'PRIVMSG ChanServ :KICK {target} {nick} {reason}') + @irc3.plugin class BanPlugin: + """A plugin to ban and unban users in an IRC channel.""" + def __init__(self, bot): + """ + Initialize the BanPlugin. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot @command(permission='admin') async def ban(self, mask, target, args): - """Ban a specific user from the channel + """ + Ban a specific user from the channel. + Usage: %%ban """ nick = args.get('') @@ -92,8 +176,10 @@ class BanPlugin: @command(permission='admin') async def unban(self, mask, target, args): - """Unban a specific user from the channel + """ + Unban a specific user from the channel. + Usage: %%unban """ nick = args.get('') @@ -101,9 +187,21 @@ class BanPlugin: await self.unban_user(target, nick) async def ban_user(self, target, nick): - """Ban a specific user from the channel""" + """ + Ban a specific user from the channel. + + Args: + target (str): The IRC channel. + nick (str): The nickname of the user. + """ self.bot.send(f'MODE {target} +b {nick}') async def unban_user(self, target, nick): - """Unban a specific user from the channel""" - self.bot.send(f'MODE {target} -b {nick}') \ No newline at end of file + """ + Unban a specific user from the channel. + + Args: + target (str): The IRC channel. + nick (str): The nickname of the user. + """ + self.bot.send(f'MODE {target} -b {nick}') diff --git a/plugins/services/anti_spam.py b/plugins/services/anti_spam.py index 5f3c3d8..f8c99f0 100644 --- a/plugins/services/anti_spam.py +++ b/plugins/services/anti_spam.py @@ -1,34 +1,75 @@ -from irc3.plugins.command import command -from irc3.plugins.cron import cron -import irc3 -from irc3 import utils -import re -from collections import defaultdict, deque +# -*- coding: utf-8 -*- +""" +IRC3 Anti-Spam Plugin with Auto-Kick and Auto-Ban + +This plugin automatically detects and mitigates spam in IRC channels +by monitoring messages for: +- Excessive message repetition +- High-frequency messaging (flooding) +- Excessive bot mentions + +Actions: +- **1st and 2nd offense**: User gets kicked. +- **3rd offense within 5 minutes**: User gets banned. + +Features: +- Uses a dynamic WHO query to verify user modes before action. +- Configurable limits for spam detection (via `antispam` settings). +- Periodic cleanup of inactive users every minute. + +Author: [Your Name] +""" + + import time -from asynchronous import AsyncEvents + +import irc3 +from collections import defaultdict, deque +from irc3.plugins.cron import cron +from asynchronous import WhoChannel + @irc3.plugin class AntiSpam: - """IRC3 Anti-Spam Plugin with Auto-Ban for Repeat Offenders""" + """A plugin for automatic spam detection and mitigation in IRC channels.""" - def __init__(self, bot): + def __init__(self, bot: irc3.IrcBot): + """ + Initialize the AntiSpam plugin with configurable thresholds. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot self.config = bot.config.get('antispam', {}) + + # User activity tracking self.user_data = defaultdict(lambda: { 'messages': deque(maxlen=int(self.config.get('repeat_limit', 3))), 'timestamps': deque(maxlen=int(self.config.get('spam_limit', 5))), 'mentions': deque(maxlen=int(self.config.get('mention_limit', 2))) }) self.kick_history = defaultdict(deque) # Track kick timestamps per user - self.exclude_list = ['ZodBot'] - self.service_name = self.config.get('service_name', 'ChanServ') - self.who_channel = WhoChannel(bot) # Initialize WHO channel handler - async def get_user_modes(self, nick, channel): - """Dynamically fetch user modes using WHO command.""" + self.exclude_list = ['ZodBot'] # Bots that should be ignored + self.service_name = self.config.get('service_name', 'ChanServ') + + self.who_channel = WhoChannel(bot) # WHO query for user modes + + async def get_user_modes(self, nick: str, channel: str) -> str: + """ + Retrieve user modes dynamically using the WHO command. + + Args: + nick (str): The user's nickname. + channel (str): The IRC channel. + + Returns: + str: User's mode (e.g., 'o' for op, 'v' for voice, etc.). + """ try: result = await self.who_channel(channel=channel) - if result['success']: + if result.get('success'): for user in result['users']: if user['nick'].lower() == nick.lower(): return user['modes'] @@ -36,36 +77,59 @@ class AntiSpam: self.bot.log.error(f"Error fetching user modes: {e}") return "" - def is_spam(self, nick, message, channel): - """Check if message meets spam criteria""" + def is_spam(self, nick: str, message: str, channel: str) -> bool: + """ + Determine whether a message meets spam criteria. + + Args: + nick (str): The user's nickname. + message (str): The message content. + channel (str): The channel where the message was sent. + + Returns: + bool: True if the message is considered spam, False otherwise. + """ user = self.user_data[nick.lower()] now = time.time() + # Check message length if len(message) > int(self.config.get('max_length', 300)): return True + # Check repeated messages if message in user['messages']: if len(user['messages']) == user['messages'].maxlen - 1: return True + # Check rapid message spam (flooding) user['timestamps'].append(now) if len(user['timestamps']) == user['timestamps'].maxlen: if (now - user['timestamps'][0]) < 60: return True + # Check excessive bot mentions if self.bot.nick.lower() in message.lower(): user['mentions'].append(now) if len(user['mentions']) == user['mentions'].maxlen: if (now - user['mentions'][0]) < 60: return True + # Store message to check for repetition user['messages'].append(message) return False @irc3.event(irc3.rfc.PRIVMSG) async def monitor_messages(self, mask, event, target, data): - """Handle incoming messages and check for spam""" - if target.startswith("#"): + """ + Monitor incoming messages for spam and take action if needed. + + Args: + mask (str): The sender's mask. + event (str): The IRC event type. + target (str): The channel where the message was sent. + data (str): The message content. + """ + if target.startswith("#"): # Process only channel messages nick = mask.nick message = data channel_name = target.lower() @@ -73,95 +137,69 @@ class AntiSpam: if nick in self.exclude_list: return + # Fetch user modes to avoid acting on moderators user_modes = await self.get_user_modes(nick, channel_name) - - if user_modes: - if {'o', '%', 'h', '@'} & set(user_modes): - return + if user_modes and {'o', '%', 'h', '@'} & set(user_modes): + return - if self.is_spam(nick, message, channel_name): - print(f"SPAM {nick} - {user_modes}") - self.handle_spam(mask, message, channel_name) + if self.is_spam(nick, message, channel_name): + self.bot.log.info(f"SPAM detected from {nick}: {message}") + self.handle_spam(mask, message, channel_name) def handle_spam(self, mask, message, channel): - """Take action against spam, escalating to ban if kicked twice in 5 minutes""" + """ + Take action against spamming users, escalating to ban if needed. + + Args: + mask (str): The sender's mask. + message (str): The spam message. + channel (str): The IRC channel. + """ nick = mask.nick current_time = time.time() - cutoff = current_time - 300 # 5 minutes ago + cutoff = current_time - 300 # 5-minute window - nick_lower = nick.lower() - user_kicks = self.kick_history[nick_lower] - - # Filter recent kicks within the last 5 minutes + user_kicks = self.kick_history[nick.lower()] recent_kicks = [ts for ts in user_kicks if ts >= cutoff] if len(recent_kicks) >= 2: - # Ban the user using hostmask + # Ban the user ban_mask = f'*!{mask.host}' - self.bot.send(f"MODE {channel} +b {ban_mask}") self.bot.privmsg( self.service_name, - f"KICK {channel} {nick} Banned for repeated spamming" + f"KICK {channel} {nick} :Banned for repeated spamming" ) - # Clear history and data - del self.kick_history[nick_lower] - self.user_data.pop(nick_lower, None) + + self.bot.log.info(f"{nick} banned for repeated spamming.") + + # Clear data + del self.kick_history[nick.lower()] + self.user_data.pop(nick.lower(), None) else: - # Kick and record timestamp + # Kick the user and log action self.bot.privmsg( self.service_name, - f"KICK {channel} {nick} :stop spamming" + f"KICK {channel} {nick} :Stop spamming." ) user_kicks.append(current_time) - self.user_data.pop(nick_lower, None) + self.bot.log.info(f"{nick} kicked for spam. Warning count: {len(recent_kicks) + 1}") + + # Clear message history for the user + self.user_data.pop(nick.lower(), None) @cron('* * * * *') def clean_old_records(self): - """Cleanup inactive users every minute""" + """Clean up inactive user records every minute.""" cutoff = time.time() - 300 - to_remove = [ - nick for nick, data in self.user_data.items() - if len(data['timestamps']) > 0 and data['timestamps'][-1] < cutoff - ] - for nick in to_remove: - del self.user_data[nick] + self.user_data = { + nick: data + for nick, data in self.user_data.items() + if len(data['timestamps']) > 0 and data['timestamps'][-1] >= cutoff + } + self.bot.log.info("Cleaned up old spam records.") def connection_made(self): - """Initialize when bot connects""" - self.bot.log.info("Enhanced AntiSpam plugin loaded with kick-to-ban escalation") - - -class WhoChannel(AsyncEvents): - """Handle WHO responses for a channel.""" - - send_line = 'WHO {channel}' - - events = ( - { - 'match': ( - r"(?i)^:\S+ 352 \S+ {channel} (?P\S+) " - r"(?P\S+) (?P\S+) (?P\S+) " - r"(?P\S+) :(?P\S+) (?P.*)" - ), - 'multi': True - }, - { - 'match': r"(?i)^:\S+ (?P(315|401)) \S+ {channel} :.*", - 'final': True - }, - ) - - def process_results(self, results=None, **value): - """Process WHO channel results into a user list.""" - users = [] - for res in results: - if 'retcode' in res: - value.update(res) - else: - res['mask'] = utils.IrcString('{nick}!{user}@{host}'.format(**res)) - users.append(res) - value['users'] = users - value['success'] = value.get('retcode') == '315' - return value \ No newline at end of file + """Initialize when bot connects.""" + self.bot.log.info("AntiSpam plugin loaded with automated kick-to-ban escalation.") diff --git a/plugins/unicode_say.py b/plugins/unicode_say.py index becfda4..34b510e 100644 --- a/plugins/unicode_say.py +++ b/plugins/unicode_say.py @@ -1,16 +1,56 @@ +# -*- coding: utf-8 -*- +""" +IRC3 Bot Plugin: Say Command with Text Styling + +This plugin for an IRC bot allows the bot to send styled messages to a specified channel using the say command. +The messages are styled with random IRC color codes, bold, and underline, along with Unicode combining characters. + +Features: +- Sends a styled message to a specified channel. +- Applies random IRC colors, bold, and underline styles to the message. +- Uses Unicode combining characters to create a glitch effect. + +Usage: +====== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @command + def say(self, mask, target, args): + %%say ... + +Author: Zodiac +Date: 2025-02-13 06:24:46 (UTC) +""" + import irc3 import random from irc3.plugins.command import command @irc3.plugin class SayPlugin: + """A plugin to send styled messages to a specified channel using the say command.""" + def __init__(self, bot): + """ + Initialize the plugin with bot reference. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot @command def say(self, mask, target, args): - """Say command + """ + Say command to send a styled message to a specified channel. + Args: + mask (str): The user mask. + target (str): The target channel or user. + args (dict): Command arguments. + + Usage: %%say ... """ channel = args.get('') @@ -24,6 +64,15 @@ class SayPlugin: self.bot.privmsg(channel, styled_message) def add_combining_characters(self, char): + """ + Add random combining characters (with style and color codes) to a character. + + Args: + char (str): The character to style. + + Returns: + str: The styled character. + """ combining_chars = [ '\u0300', '\u0301', '\u0302', '\u0303', '\u0304', '\u0305', '\u0306', '\u0307', '\u0308', '\u0309', '\u030A', '\u030B', @@ -52,6 +101,15 @@ class SayPlugin: return glitched_char def style_message(self, message): + """ + Apply styling to each character in the message. + + Args: + message (str): The message to style. + + Returns: + str: The styled message. + """ white_color_code = '\x0300' # White color styled_message = '' diff --git a/plugins/unicode_spam.py b/plugins/unicode_spam.py index 4b84b88..c530c52 100644 --- a/plugins/unicode_spam.py +++ b/plugins/unicode_spam.py @@ -1,3 +1,31 @@ +# -*- coding: utf-8 -*- +""" +IRC3 Bot Plugin: Unicode Spammer and Nickname Changer + +This plugin for an IRC bot enables continuous spamming of Unicode characters to a specified target and periodically changes the bot's nickname to mimic a random user in the channel. + +Features: +- Spams messages composed of valid Unicode characters to a specified target. +- Periodically changes the bot's nickname to mimic a random user in the channel. +- Handles starting and stopping of spam and nickname-changing tasks via bot commands. + +Usage: +====== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @command + def unicode(self, mask, target, args): + %%unicode [] + + @command + def unicodestop(self, mask, target, args): + %%unicodestop [] + +Author: Zodiac +Date: 2025-02-13 06:25:41 (UTC) +""" + import random import string import irc3 @@ -5,9 +33,18 @@ from irc3.plugins.command import command import asyncio import unicodedata + @irc3.plugin class UnicodeSpammer: + """A plugin to spam Unicode characters and change the bot's nickname periodically.""" + def __init__(self, bot): + """ + Initialize the plugin with bot reference. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot self.valid_code_points = self._generate_valid_code_points() random.shuffle(self.valid_code_points) diff --git a/plugins/upload.py b/plugins/upload.py index db6ad84..6580455 100644 --- a/plugins/upload.py +++ b/plugins/upload.py @@ -1,3 +1,4 @@ +# -*- coding: utf-8 -*- """ IRC Bot Plugin for Uploading Files to hardfiles.org @@ -15,7 +16,7 @@ Dependencies: - yt-dlp - ircstyle -Author: Your Name +Author: Zodiac Version: 1.2 Date: 2025-02-12 """ @@ -36,16 +37,26 @@ from urllib.parse import urlparse @irc3.plugin class UploadPlugin: - """ - IRC bot plugin for downloading files via yt-dlp and uploading them to hardfiles.org. - """ + """IRC bot plugin for downloading files via yt-dlp and uploading them to hardfiles.org.""" def __init__(self, bot): + """ + Initialize the UploadPlugin with an IRC bot instance. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot def _ensure_str(self, value): """ Ensure the value is a string. If it's bytes, decode it as UTF-8 with error replacement. + + Args: + value (Union[str, bytes, None]): The value to ensure as a string. + + Returns: + str: The value as a string. """ if isinstance(value, bytes): return value.decode('utf-8', errors='replace') @@ -59,9 +70,9 @@ class UploadPlugin: Upload a file to hardfiles.org (Max 100MB). Args: - mask: The user mask (nickname@host) of the command issuer. - target: The channel or user where the command was issued. - args: Parsed command arguments. + mask (str): The user mask (nickname@host) of the command issuer. + target (str): The channel or user where the command was issued. + args (dict): Parsed command arguments. Usage: %%upload [--mp3] @@ -72,7 +83,7 @@ class UploadPlugin: if not url: self.bot.privmsg( target, - ircstyle.style("Usage: !upload [--mp3] ", fg="red", bold=True, reset=True) + ircstyle.style("Usage: !upload [--mp3] ", fg="red", bold=True, reset=True), ) return @@ -83,20 +94,31 @@ class UploadPlugin: exc_msg = self._ensure_str(exc) self.bot.privmsg( target, - ircstyle.style(f"Upload task error: {exc_msg}", fg="red", bold=True, reset=True) + ircstyle.style(f"Upload task error: {exc_msg}", fg="red", bold=True, reset=True), ) async def do_upload(self, url, target, mp3): """ Download a file using yt-dlp and upload it to hardfiles.org. Handles binary data and non-UTF-8 strings to avoid decoding errors. + + Args: + url (str): The URL of the file to download. + target (str): The channel or user to send messages to. + mp3 (bool): Whether to convert the downloaded file to MP3. """ max_size = 100 * 1024 * 1024 # 100MB limit with tempfile.TemporaryDirectory() as tmp_dir: parsed_url = urlparse(url) domain = parsed_url.netloc.lower() - skip_check_domains = ("x.com", "instagram.com", "youtube.com", "youtu.be", "streamable.com") + skip_check_domains = ( + "x.com", + "instagram.com", + "youtube.com", + "youtu.be", + "streamable.com", + ) should_check_headers = not any(domain.endswith(d) for d in skip_check_domains) if should_check_headers: @@ -108,8 +130,10 @@ class UploadPlugin: target, ircstyle.style( f"Failed to fetch headers: HTTP {response.status}", - fg="red", bold=True, reset=True - ) + fg="red", + bold=True, + reset=True, + ), ) return content_length = response.headers.get('Content-Length') @@ -118,15 +142,22 @@ class UploadPlugin: target, ircstyle.style( f"File size ({int(content_length) // (1024 * 1024)}MB) exceeds 100MB limit", - fg="red", bold=True, reset=True - ) + fg="red", + bold=True, + reset=True, + ), ) return except Exception as e: err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Error during header check: {err_msg}", fg="red", bold=True, reset=True) + ircstyle.style( + f"Error during header check: {err_msg}", + fg="red", + bold=True, + reset=True, + ), ) return @@ -137,11 +168,15 @@ class UploadPlugin: 'noplaylist': True, 'quiet': True, 'concurrent_fragment_downloads': 5, - 'postprocessors': [{ - 'key': 'FFmpegExtractAudio', - 'preferredcodec': 'mp3', - 'preferredquality': '192', - }] if mp3 else [], + 'postprocessors': [ + { + 'key': 'FFmpegExtractAudio', + 'preferredcodec': 'mp3', + 'preferredquality': '192', + } + ] + if mp3 + else [], } with yt_dlp.YoutubeDL(ydl_opts) as ydl: @@ -151,13 +186,20 @@ class UploadPlugin: err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Info extraction failed: {err_msg}", fg="red", bold=True, reset=True) + ircstyle.style( + f"Info extraction failed: {err_msg}", fg="red", bold=True, reset=True + ), ) return except UnicodeDecodeError: self.bot.privmsg( target, - ircstyle.style("Error: Received non-UTF-8 output during info extraction", fg="red", bold=True, reset=True) + ircstyle.style( + "Error: Received non-UTF-8 output during info extraction", + fg="red", + bold=True, + reset=True, + ), ) return @@ -167,8 +209,10 @@ class UploadPlugin: target, ircstyle.style( f"File size ({estimated_size // (1024 * 1024)}MB) exceeds 100MB limit", - fg="red", bold=True, reset=True - ) + fg="red", + bold=True, + reset=True, + ), ) return @@ -178,13 +222,20 @@ class UploadPlugin: err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Download failed: {err_msg}", fg="red", bold=True, reset=True) + ircstyle.style( + f"Download failed: {err_msg}", fg="red", bold=True, reset=True + ), ) return except UnicodeDecodeError: self.bot.privmsg( target, - ircstyle.style("Error: Received non-UTF-8 output during download", fg="red", bold=True, reset=True) + ircstyle.style( + "Error: Received non-UTF-8 output during download", + fg="red", + bold=True, + reset=True, + ), ) return @@ -198,20 +249,43 @@ class UploadPlugin: description = self._ensure_str(info.get("description")) if title: - metadata_parts.append(ircstyle.style(f"Title: {title}", fg="yellow", bold=True, reset=True)) + metadata_parts.append( + ircstyle.style(f"Title: {title}", fg="yellow", bold=True, reset=True) + ) if uploader: - metadata_parts.append(ircstyle.style(f"Uploader: {uploader}", fg="purple", bold=True, reset=True)) + metadata_parts.append( + ircstyle.style(f"Uploader: {uploader}", fg="purple", bold=True, reset=True) + ) if duration: - metadata_parts.append(ircstyle.style(f"Duration: {self._format_duration(duration)}", fg="green", bold=True, reset=True)) + metadata_parts.append( + ircstyle.style( + f"Duration: {self._format_duration(duration)}", + fg="green", + bold=True, + reset=True, + ) + ) if upload_date: - formatted_date = f"{upload_date[:4]}-{upload_date[4:6]}-{upload_date[6:]}" if len(upload_date) == 8 else upload_date - metadata_parts.append(ircstyle.style(f"Upload Date: {formatted_date}", fg="aqua", bold=True, reset=True)) + formatted_date = ( + f"{upload_date[:4]}-{upload_date[4:6]}-{upload_date[6:]}" + if len(upload_date) == 8 + else upload_date + ) + metadata_parts.append( + ircstyle.style( + f"Upload Date: {formatted_date}", fg="aqua", bold=True, reset=True + ) + ) if view_count is not None: - metadata_parts.append(ircstyle.style(f"Views: {view_count}", fg="royal", bold=True, reset=True)) + metadata_parts.append( + ircstyle.style(f"Views: {view_count}", fg="royal", bold=True, reset=True) + ) if description: if len(description) > 200: description = description[:200] + "..." - metadata_parts.append(ircstyle.style(f"Description: {description}", fg="silver", reset=True)) + metadata_parts.append( + ircstyle.style(f"Description: {description}", fg="silver", reset=True) + ) if metadata_parts: self.bot.privmsg(target, " | ".join(metadata_parts)) @@ -219,7 +293,7 @@ class UploadPlugin: if not downloaded_files: self.bot.privmsg( target, - ircstyle.style("No files downloaded", fg="red", bold=True, reset=True) + ircstyle.style("No files downloaded", fg="red", bold=True, reset=True), ) return @@ -228,7 +302,12 @@ class UploadPlugin: if not downloaded_file or not os.path.exists(downloaded_file): self.bot.privmsg( target, - ircstyle.style(f"Downloaded file not found: {downloaded_file}", fg="red", bold=True, reset=True) + ircstyle.style( + f"Downloaded file not found: {downloaded_file}", + fg="red", + bold=True, + reset=True, + ), ) return @@ -238,8 +317,10 @@ class UploadPlugin: target, ircstyle.style( f"File size ({file_size // (1024 * 1024)}MB) exceeds 100MB limit", - fg="red", bold=True, reset=True - ) + fg="red", + bold=True, + reset=True, + ), ) return @@ -252,13 +333,20 @@ class UploadPlugin: 'file', file_content, filename=os.path.basename(downloaded_file), - content_type='application/octet-stream' + content_type='application/octet-stream', ) - async with session.post('https://hardfiles.org/', data=form, allow_redirects=False) as resp: + async with session.post( + 'https://hardfiles.org/', data=form, allow_redirects=False + ) as resp: if resp.status not in [200, 201, 302, 303]: self.bot.privmsg( target, - ircstyle.style(f"Upload failed: HTTP {resp.status}", fg="red", bold=True, reset=True) + ircstyle.style( + f"Upload failed: HTTP {resp.status}", + fg="red", + bold=True, + reset=True, + ), ) return raw_response = await resp.read() @@ -267,21 +355,29 @@ class UploadPlugin: upload_url = self.extract_url_from_response(response_text) or "Unknown URL" upload_url = self._ensure_str(upload_url) response_msg = ( - ircstyle.style("Upload successful: ", fg="green", bold=True, reset=True) + - ircstyle.style(upload_url, fg="blue", underline=True, reset=True) + ircstyle.style("Upload successful: ", fg="green", bold=True, reset=True) + + ircstyle.style(upload_url, fg="blue", underline=True, reset=True) ) self.bot.privmsg(target, response_msg) except Exception as e: err_msg = self._ensure_str(e) self.bot.privmsg( target, - ircstyle.style(f"Error during file upload: {err_msg}", fg="red", bold=True, reset=True) + ircstyle.style( + f"Error during file upload: {err_msg}", fg="red", bold=True, reset=True + ), ) return def extract_url_from_response(self, response_text): """ Extract the first URL found in the response text. + + Args: + response_text (str): The response text to search for URLs. + + Returns: + str: The first URL found in the response text, or None if no URL is found. """ match = re.search(r'https?://\S+', response_text) return match.group(0) if match else None @@ -289,8 +385,14 @@ class UploadPlugin: def _format_duration(self, seconds): """ Convert seconds into a human-readable duration string. + + Args: + seconds (int): The duration in seconds. + + Returns: + str: The formatted duration string. """ seconds = int(seconds) m, s = divmod(seconds, 60) h, m = divmod(m, 60) - return f"{h}h {m}m {s}s" if h else f"{m}m {s}s" + return f"{h}h {m}m {s}s" if h else f"{m}m {s}s" \ No newline at end of file diff --git a/plugins/urban_dictionary.py b/plugins/urban_dictionary.py index 7265bf0..98cf085 100644 --- a/plugins/urban_dictionary.py +++ b/plugins/urban_dictionary.py @@ -1,8 +1,33 @@ # -*- coding: utf-8 -*- -from irc3.plugins.command import command -from irc3.compat import Queue +""" +IRC3 Bot Plugin: Urban Dictionary Search + +This plugin for an IRC bot allows users to search for terms on Urban Dictionary and post the results to an IRC channel. +It uses aiohttp for asynchronous HTTP requests and a queue to manage search requests. + +Features: +- Asynchronously fetches definitions from Urban Dictionary. +- Enqueues search requests and processes them one at a time. +- Formats and posts the definition, example, and permalink to the IRC channel. + +Usage: +====== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @command + def urban(self, mask, target, args): + %%urban ... + +Author: Zodiac +Date: 2025-02-12 +""" + import irc3 import aiohttp +from irc3.plugins.command import command +from irc3.compat import Queue + @irc3.plugin class UrbanDictionaryPlugin: @@ -13,6 +38,12 @@ class UrbanDictionaryPlugin: """ def __init__(self, bot): + """ + Initialize the plugin with bot reference. + + Args: + bot (irc3.IrcBot): The IRC bot instance. + """ self.bot = bot self.queue = Queue() # Queue for managing search requests self.session = None # aiohttp session initialized lazily @@ -22,6 +53,13 @@ class UrbanDictionaryPlugin: def urban(self, mask, target, args): """ Search Urban Dictionary for a term. + + Args: + mask (str): The user mask (nickname@host) of the command issuer. + target (str): The channel or user where the command was issued. + args (dict): Command arguments. + + Usage: %%urban ... """ term = ' '.join(args['']) diff --git a/plugins/url_title_sniffer.py b/plugins/url_title_sniffer.py index 0b28a2e..ca015b6 100644 --- a/plugins/url_title_sniffer.py +++ b/plugins/url_title_sniffer.py @@ -1,5 +1,26 @@ +# -*- coding: utf-8 -*- """ -A plugin for fetching and displaying titles of URLs shared in IRC messages. +IRC3 Bot Plugin: URL Title Fetcher + +This plugin for an IRC bot fetches and displays the titles of URLs shared in IRC messages. +It uses aiohttp for asynchronous HTTP requests and lxml for HTML parsing. + +Features: +- Listens for PRIVMSG events in the IRC channel. +- Extracts URLs from messages and fetches their titles. +- Posts the title and URL back to the IRC channel. + +Usage: +====== +To use this module, load it as a plugin in your IRC bot configuration. + +Example: + @event + def on_privmsg(self, mask, event, target, data): + # Extract URLs from messages and fetch their titles. + +Author: Zodiac +Date: 2025-02-13 """ import re @@ -11,6 +32,7 @@ from irc3 import event from irc3.compat import Queue +@irc3.plugin class URLTitlePlugin: """ A plugin to fetch and display the titles of URLs shared in IRC messages. @@ -29,9 +51,9 @@ class URLTitlePlugin: bot (irc3.IrcBot): The IRC bot instance. """ self.bot = bot - self.url_queue = Queue() + self.url_queue = Queue() # Queue for managing URL processing self.session = aiohttp.ClientSession(loop=self.bot.loop) - self.bot.create_task(self.process_urls()) + self.bot.create_task(self.process_urls()) # Start URL processor @event(irc3.rfc.PRIVMSG) async def on_privmsg(self, mask, event, target, data): @@ -72,12 +94,7 @@ class URLTitlePlugin: ) await self.bot.privmsg(target, formatted_message) else: - # Format the error message with colors and styles - # formatted_message = ( - # f"\x02\x034Error:\x03 Could not find a title for: " - # f"\x0311{url}\x03" - # ) - # await self.bot.privmsg(target, formatted_message) + # Handle cases where no title is found pass except Exception as e: self.bot.log.error(f"Error processing URL {url}: {e}")