g1mp/plugins/unicode_spam.py

# -*- 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 [<target>]

    @command
    def unicodestop(self, mask, target, args):
        %%unicodestop [<target>]

Author: Zodiac
Date: 2025-02-13 06:25:41 (UTC)
"""

import random
import string
import irc3
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)
        # Dictionary to keep track of running spam tasks: {target_nick: asyncio.Task, ...}
        self.spam_tasks = {}
        # Task for periodically changing nick
        self.nick_changer_task = None
        # Store the original nickname to restore later.
        self.original_nick = self.bot.nick
        # Nick change interval in seconds; adjust as desired.
        self.nick_change_interval = 10

    def _generate_valid_code_points(self):
        """Generate a list of valid Unicode code points that are renderable and not Chinese."""
        excluded_code_points = {
            # Already excluded code points
            0x1F95F,  # 🮕
            0x18F33,  # 𘏳
            0x18F34,  # 𘏴
            0x18F35,  # 𘏵
            *range(0x1FB8, 0x1FCD),  # Range for the mentioned characters (🭨 to 🭿)
            *range(0x1FD0, 0x1FF0),  # Range for the mentioned characters (🯀 to 🯹)
            *range(0x1DF20, 0x1DF33),  # Range for 𐹠 to 𐹲

            # New characters to exclude
            0x1F200, 0x1F201, 0x1F202, 0x1F203, 0x1F204, 0x1F205, 0x1F206, 0x1F207,
            0x1F208, 0x1F209, 0x1F20A, 0x1F20B, 0x1F20C, 0x1F20D, 0x1F20E, 0x1F20F,
            0x1F210, 0x1F211, 0x1F212, 0x1F213, 0x1F214, 0x1F215, 0x1F216, 0x1F217,
            0x1F218, 0x1F219, 0x1F21A, 0x1F21B, 0x1F21C, 0x1F21D, 0x1F21E, 0x1F21F,
            0x1F220, 0x1F221, 0x1F222, 0x1F223, 0x1F224, 0x1F225, 0x1F226, 0x1F227,
            0x1F228, 0x1F229, 0x1F22A, 0x1F22B, 0x1F22C, 0x1F22D, 0x1F22E, 0x1F22F,
            0x1F230, 0x1F231, 0x1F232, 0x1F233, 0x1F234, 0x1F235, 0x1F236, 0x1F237,
            0x1F238, 0x1F239, 0x1F23A, 0x1F23B, 0x1F23C, 0x1F23D, 0x1F23E, 0x1F23F,
            0x1F240, 0x1F241, 0x1F242, 0x1F243, 0x1F244, 0x1F245, 0x1F246, 0x1F247,
            0x1F248, 0x1F249, 0x1F24A, 0x1F24B, 0x1F24C, 0x1F24D, 0x1F24E, 0x1F24F,
            0x1F250, 0x1F251, 0x1F252, 0x1F253, 0x1F254, 0x1F255, 0x1F256, 0x1F257,
            0x1F258, 0x1F259, 0x1F25A, 0x1F25B, 0x1F25C, 0x1F25D, 0x1F25E, 0x1F25F,

            # Additional code points to remove as per the original list
            0x1F600, 0x1F601, 0x1F602, 0x1F603, 0x1F604, 0x1F605, 0x1F606, 0x1F607,
            # ... (continue for any other exclusions you need) ...

            # Unicode ranges to exclude (example ranges, adjust as needed)
            *range(0x17004, 0x1707F),
            *range(0x17154, 0x1717F),
            *range(0x171D4, 0x171FF),
            # ... many more ranges as in your original code ...
        }

        return [
            cp for cp in range(0x110000)
            if not self._is_chinese(cp)      # Exclude Chinese characters
            and self._is_renderable(cp)       # Exclude non-renderable characters
            and cp not in excluded_code_points  # Exclude the specified characters
        ]

    def _is_chinese(self, code_point):
        """Check if the Unicode code point is a Chinese character."""
        try:
            char = chr(code_point)
            return unicodedata.name(char).startswith("CJK")
        except ValueError:
            return False

    def _is_renderable(self, code_point):
        """Check if a Unicode code point is renderable in IRC."""
        try:
            char = chr(code_point)
        except ValueError:
            return False
        category = unicodedata.category(char)
        excluded_categories = {'Cc', 'Cf', 'Cs', 'Co', 'Cn', 'Zl', 'Zp', 'Mn', 'Mc', 'Me'}
        return category not in excluded_categories

    def _random_nick(self):
        """Generate a random nickname if no users are found."""
        return 'User' + ''.join(random.choices(string.ascii_letters + string.digits, k=5))

    async def _change_nick_periodically(self, channel):
        """
        Every self.nick_change_interval seconds change the bot's nickname to mimic
        a random user's nickname from the channel (with an underscore appended).
        If no users are found, generate a random nickname.
        """
        self.original_nick = self.bot.nick  # store the original nick
        try:
            while True:
                channel_key = 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}_"
                    else:
                        new_nick = self._random_nick()
                    # Send the NICK command
                    self.bot.send(f'NICK {new_nick}')
                    self.bot.log.info(f"Nickname changed to mimic: {new_nick}")
                    self.bot.nick = new_nick
                else:
                    self.bot.log.info(f"Channel {channel} not found for nick change.")
                await asyncio.sleep(self.nick_change_interval)
        except asyncio.CancelledError:
            # Restore the original nickname upon cancellation.
            self.bot.send(f'NICK {self.original_nick}')
            self.bot.nick = self.original_nick
            self.bot.log.info("Nickname reverted to original.")
            raise
        except Exception as e:
            self.bot.log.error(f"Error changing nickname: {e}")

    @command(permission='admin')
    async def unicode(self, mask, target, args):
        """
        Handle the !unicode command:
          - !unicode <target>: start spamming Unicode messages to <target> and start periodic nick changes.
        
        %%unicode [<target>]
        """
        target_nick = args['<target>']
        if target_nick in self.spam_tasks:
            await self._send_message(target, f"Unicode spam is already running for {target_nick}.")
        else:
            if not target_nick:
                target_nick = target
            # Start the Unicode spam task for the target.
            spam_task = asyncio.create_task(self._unicode_spam_loop(target_nick))
            self.spam_tasks[target_nick] = spam_task
            await self._send_message(target, f"Started Unicode spam for {target_nick}.")

            # Start the nickname changer if not already running.
            # We assume that if the command was issued in a channel,
            # then `target` is the channel name.
            if self.nick_changer_task is None:
                self.nick_changer_task = asyncio.create_task(self._change_nick_periodically(target))

    @command(permission='admin')
    async def unicodestop(self, mask, target, args):
        """
        Handle the !unicodestop command:
          - !unicodestop: stop all running Unicode spamming and nick-changing tasks.

        %%unicodestop [<target>]
        """
        stop_msgs = []
        # Cancel all running spam tasks.
        if self.spam_tasks:
            for nick, task in list(self.spam_tasks.items()):
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
                del self.spam_tasks[nick]
            stop_msgs.append("Stopped all Unicode spam tasks.")
        else:
            stop_msgs.append("No Unicode spam tasks are running.")

        # Cancel the nickname changer task if it is running.
        if self.nick_changer_task is not None:
            self.nick_changer_task.cancel()
            try:
                await self.nick_changer_task
            except asyncio.CancelledError:
                pass
            self.nick_changer_task = None
            stop_msgs.append("Nickname changer stopped and nick reverted.")

        await self._send_message(target, " ".join(stop_msgs))

    async def _unicode_spam_loop(self, target_nick):
        """
        Continuously send messages composed of Unicode characters to the given target.
        This loop is cancelled when the !unicodestop command is issued.
        """
        message_count = 0
        try:
            while True:
                buffer = []
                # Cycle through valid code points.
                for code_point in self.valid_code_points:
                    try:
                        buffer.append(chr(code_point))
                        # When the buffer gets large, send a message.
                        if len(buffer) >= 400:
                            await self._send_message(target_nick, ''.join(buffer))
                            buffer.clear()
                            message_count += 1
                            await self._throttle_messages(message_count)
                    except Exception as e:
                        self.bot.log.error(f"Error processing character {code_point}: {e}")

                # Send any remaining characters.
                if buffer:
                    await self._send_message(target_nick, ''.join(buffer))
        except asyncio.CancelledError:
            self.bot.log.info(f"Unicode spam for {target_nick} cancelled.")
            await self._send_message(target_nick, "Unicode spam stopped.")
            raise

    async def _send_message(self, target, message):
        """Send a message to the target with error handling."""
        try:
            self.bot.privmsg(target, message)
        except Exception as e:
            self.bot.log.error(f"Error sending message to {target}: {e}")

    async def _throttle_messages(self, message_count):
        """Throttle messages to prevent flooding."""
        await asyncio.sleep(0.007)
        if message_count % 5 == 0:
            await asyncio.sleep(0.07)
        if message_count % 25 == 0:
            await asyncio.sleep(0.07)
        if message_count % 50 == 0:
            await asyncio.sleep(0.07)