Zodiac/g1mp
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

clean up code

Browse Source
This commit is contained in:
Zodiac 2025-02-12 22:35:15 -08:00
parent 75d830b260
commit 3d82132425
15 changed files with 1043 additions and 632 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

457
plugins/asynchronious.py
View File

@ -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<away>.*)"},
{
'match': (
r"(?i)^:\S+ 311 \S+ {nick} (?P<username>\S+) "
r"(?P<host>\S+) . :(?P<realname>.*)"
r"(?i)^:\S+ 311 \S+ {nick} (?P<username>\S+) (?P<host>\S+) "
r". :(?P<realname>.*)"
)
},
{
'match': (
r"(?i)^:\S+ 312 \S+ {nick} (?P<server>\S+) "
r":(?P<server_desc>.*)"
)
'match': r"(?i)^:\S+ 312 \S+ {nick} (?P<server>\S+) :(?P<server_desc>.*)"
},
{'match': r"(?i)^:\S+ 317 \S+ {nick} (?P<idle>[0-9]+).*"},
{
'match': r"(?i)^:\S+ 319 \S+ {nick} :(?P<channels>.*)",
'multi': True
},
{'match': r"(?i)^:\S+ 319 \S+ {nick} :(?P<channels>.*)", 'multi': True},
{
'match': (
r"(?i)^:\S+ 330 \S+ {nick} (?P<account>\S+) "
r":(?P<account_desc>.*)"
r"(?i)^:\S+ 330 \S+ {nick} (?P<account>\S+) :(?P<account_desc>.*)"
)
},
{'match': r"(?i)^:\S+ 671 \S+ {nick} :(?P<connection>.*)"},
{
'match': (
r"(?i)^:\S+ (?P<retcode>(318|401)) \S+ (?P<nick>{nick}) :.*"
),
'final': True
'match': r"(?i)^:\S+ (?P<retcode>(318|401)) \S+ (?P<nick>{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<user>\S+) "
r"(?P<host>\S+) (?P<server>\S+) (?P<nick>\S+) "
r"(?P<modes>\S+) :(?P<hopcount>\S+) (?P<realname>.*)"
r"(?i)^:\S+ 352 \S+ {channel} (?P<user>\S+) (?P<host>\S+) "
r"(?P<server>\S+) (?P<nick>\S+) (?P<modes>\S+) "
r":(?P<hopcount>\S+) (?P<realname>.*)"
),
'multi': True
},
{
'match': r"(?i)^:\S+ (?P<retcode>(315|401)) \S+ {channel} :.*",
'final': True
'multi': True,
},
{'match': r"(?i)^:\S+ (?P<retcode>(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<user>\S+)"),
("i", r"(?P<ip>\S+)"),
("h", r"(?P<host>\S+)"),
("s", r"(?P<server>\S+)"),
("n", r"(?P<nick>\S+)"),
("a", r"(?P<account>\S+)"),
("r", r":(?P<realname>.*)"),
])
send_line = "WHO {channel} c%{flags}"
events = (
{
'match': r"(?i)^:\S+ (?P<retcode>(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<channel>\S+) (?P<user>\S+) "
r"(?P<host>\S+) (?P<server>\S+) (?P<nick>{nick}) "
r"(?P<modes>\S+) :(?P<hopcount>\S+)\s*(?P<realname>.*)"
)
},
{
'match': r"(?i)^:\S+ (?P<retcode>(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>({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<retcode>(331|332|TOPIC))"
r"(:?\s+\S+\s+|\s+){channel} :(?P<topic>.*)"
),
'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<nicknames>.*)",
'multi': True
},
{
'match': r"(?i)^:\S+ (?P<retcode>(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<mask>\S+) "
r"(?P<user>\S+) (?P<timestamp>\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<mask>\S+) NOTICE \S+ :\x01(?P<ctcp>\S+) "
r"(?P<reply>.*)\x01"
),
'final': True
},
{
'match': r"(?i)^:\S+ (?P<retcode>486) \S+ :(?P<reply>.*)",
'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)
return None

118
plugins/bomb.py
View File

@ -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.log.info(f"Nickname changed to: {new_nick}")
await asyncio.sleep(self.sleep_durations['_change_nick_periodically'])
except asyncio.CancelledError:
pass
@ -89,48 +122,49 @@ 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.
"""
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.
"""
Start periodic tasks via a command.
Usage:
%%annoy
"""
if mask.nick == self.bot.config.get('owner', ''):
@ -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.
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."

88
plugins/disregard.py
View File

@ -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 <nick> -> 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 <nick>
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('<nick>')
if not user:
self.bot.privmsg(target, "Usage: !disregard <nick>")
return
return "Usage: !disregard <nick>"
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.")

86
plugins/goat.py
View File

@ -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 [<nick>]
%%goatstop
Commands:
%%goat [<nick>]
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 [<nick>]
"""
# Get the optional nick argument (it may be None or an empty string)
nick = args.get("<nick>") # Do not provide a default value here
nick = args.get("<nick>")
# 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.")

37
plugins/imitate.py
View File

@ -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] [<nick>]
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

67
plugins/matrix.py
View File

@ -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",
# 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,6 +66,12 @@ class MatrixPlugin:
"""
Display a Matrix-style rain of characters.
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)
@ -35,6 +79,16 @@ class MatrixPlugin:
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()))

31
plugins/my_yt.py
View File

@ -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 [--<amount>] <search_query>...
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"

57
plugins/random_msg.py
View File

@ -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 <message>...
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 <message>...
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['<message>'])
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

132
plugins/services/administration.py
View File

@ -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>]
"""
nick = args.get('<nick>')
@ -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>]
"""
nick = args.get('<nick>')
@ -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> [<reason>]
"""
nick = args.get('<nick>')
@ -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>
"""
nick = args.get('<nick>')
@ -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>
"""
nick = args.get('<nick>')
@ -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"""
"""
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}')

200
plugins/services/anti_spam.py
View File

@ -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):
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.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<user>\S+) "
r"(?P<host>\S+) (?P<server>\S+) (?P<nick>\S+) "
r"(?P<modes>\S+) :(?P<hopcount>\S+) (?P<realname>.*)"
),
'multi': True
},
{
'match': r"(?i)^:\S+ (?P<retcode>(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
"""Initialize when bot connects."""
self.bot.log.info("AntiSpam plugin loaded with automated kick-to-ban escalation.")

60
plugins/unicode_say.py
View File

@ -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 <channel> <message>...
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> <message>...
"""
channel = args.get('<channel>')
@ -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 = ''

37
plugins/unicode_spam.py
View File

@ -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 [<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
@ -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)

182
plugins/upload.py
View File

@ -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] <url>
@ -72,7 +83,7 @@ class UploadPlugin:
if not url:
self.bot.privmsg(
target,
ircstyle.style("Usage: !upload [--mp3] <url>", fg="red", bold=True, reset=True)
ircstyle.style("Usage: !upload [--mp3] <url>", 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': [{
'postprocessors': [
{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '192',
}] if mp3 else [],
}
]
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,6 +385,12 @@ 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)

42
plugins/urban_dictionary.py
View File

@ -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 <term>...
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>...
"""
term = ' '.join(args['<term>'])

35
plugins/url_title_sniffer.py
View File

@ -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}")