Compare commits

..

No commits in common. "master" and "main" have entirely different histories.
master ... main

28 changed files with 3 additions and 4772 deletions

2
.gitignore vendored
View File

@ -1,2 +0,0 @@
*.pyc
config.ini

3
README.md Normal file
View File

@ -0,0 +1,3 @@
# g1mp
IRC bot using https://github.com/gawel/irc3

View File

@ -1,123 +0,0 @@
# -*- coding: utf-8 -*-
from irc3.compat import asyncio
import re
class event:
iotype = 'in'
iscoroutine = True
def __init__(self, **kwargs):
# kwargs get interpolated into the regex.
# Any kwargs not ending in _re get escaped
self.meta = kwargs.get('meta')
regexp = self.meta['match'].format(**{
k: v if k.endswith('_re') else re.escape(v)
for (k, v) in kwargs.items()
if k != 'meta'
})
self.regexp = regexp
regexp = getattr(self.regexp, 're', self.regexp)
self.cregexp = re.compile(regexp).match
def compile(self, *args, **kwargs):
return self.cregexp
def __repr__(self):
s = getattr(self.regexp, 'name', self.regexp)
name = self.__class__.__name__
return '<temp_event {0} {1}>'.format(name, s)
def __call__(self, callback):
async def wrapper(*args, **kwargs):
# Ensure callback is awaited if it's an async function
if asyncio.iscoroutinefunction(callback):
return await callback(self, *args, **kwargs)
else:
return callback(self, *args, **kwargs)
self.callback = wrapper
return self
def default_result_processor(self, results=None, **value): # pragma: no cover
value['results'] = results
if len(results) == 1:
value.update(results[0])
return value
def async_events(context, events, send_line=None,
process_results=default_result_processor,
timeout=30, **params):
loop = context.loop
task = loop.create_future() # async result
results = [] # store events results
events_ = [] # reference registered events
# async timeout
timeout = asyncio.ensure_future(
asyncio.sleep(timeout, loop=loop), loop=loop)
def end(t=None):
"""t can be a future (timeout done) or False (result success)"""
if not task.done():
# cancel timeout if needed
if t is False:
timeout.cancel()
# detach events
context.detach_events(*events_)
# clean refs
events_[:] = []
# set results
task.set_result(process_results(results=results, timeout=bool(t)))
# end on timeout
timeout.add_done_callback(end)
def callback(e, **kw):
"""common callback for all events"""
results.append(kw)
if e.meta.get('multi') is not True:
context.detach_events(e)
if e in events_:
events_.remove(e)
if e.meta.get('final') is True:
# end on success
end(False)
events_.extend([event(meta=kw, **params)(callback) for kw in events])
context.attach_events(*events_, insert=True)
if send_line:
context.send_line(send_line.format(**params))
return task
class AsyncEvents:
"""Asynchronious events"""
timeout = 30
send_line = None
events = []
def __init__(self, context):
self.context = context
def process_results(self, results=None, **value): # pragma: no cover
"""Process results.
results is a list of dict catched during event.
value is a dict containing some metadata (like timeout=(True/False).
"""
return default_result_processor(results=results, **value)
def __call__(self, **kwargs):
"""Register events; and callbacks then return a `asyncio.Future`.
Events regexp are compiled with `params`"""
kwargs.setdefault('timeout', self.timeout)
kwargs.setdefault('send_line', self.send_line)
kwargs['process_results'] = self.process_results
return async_events(self.context, self.events, **kwargs)

View File

@ -1,23 +0,0 @@
88██████████17██16███88█89,16▀16█29█41███7█41█16,29▀17,29▀29█41███41,7▀▀7█7,53▀▀53████7█41██7,53▀53█████7,53▀41,53▀41,7▀41█30█41,30▀30,29▀29,16▀17█16,29▀29██
88█████████88,89▀17██16██16,89▀88█89,88▀16█17█29█41████29,41▀29█29,41▀41█41,7▀7,53▀53██53,7▀7█53██████7,53▀7,41▀53███53,65▀▀53█7█53,7▀53█41,7▀41█29█16,29▀29,41▀41█41,29▀29██
88█████████89█17██16██89█88██16,89▀16█17,16▀29█41███41,29▀29█41██7█53██53,41▀7,41▀▀53██53,41▀7,41▀53██53,7▀41█7,41▀53███65,53▀53█53,7▀41█7,41▀7█41█41,30▀29█41,29▀29██29,17▀29,16▀
88█████████89█17██16,89▀89██88██89,16▀16██28,16▀29,28▀41█41,7▀▀28,29▀41█41,29▀41█7,41▀41████41,7▀41,53▀▀7,53▀53,65▀▀7,65▀▀▀53,65▀▀▀▀▀▀7,65▀41,53▀▀41█29,41▀29,17▀▀18,17▀29,17▀17,88▀88█
88█████████89█17█17,18▀16,89▀89,88▀89█88███16,89▀▀16██41█7,41▀53,7▀41,7▀29█29,17▀29█29,7▀41,65▀7,65▀53,65▀65███65,66▀66,78▀▀▀66█66,65▀▀▀▀66████65,66▀65███53,65▀29,53▀16,41▀88,17▀▀▀88,16▀
88█████████89█17█29,30▀17█88,89▀88███████16█28,16▀29,41▀41,7▀41,53▀29,7▀41,53▀65█████65,66▀65,78▀66,78▀▀▀66██66,65▀65███65,53▀65██65,77▀65,66▀66█66,78▀▀66,77▀77,65▀65█41,77▀65,78▀▀65,77▀41,64▀
88█████████89█17█31,18▀18█17█89,16▀88█89,88▀88████89,16▀28,16▀29,40▀7█41,7▀41██29,41▀7,41▀65,53▀65█66,65▀78,66▀78█78,0▀▀97,0▀78,0▀78█66,78▀65██████65,77▀65████65,77▀77,0▀78,0▀0,78▀0,77▀77,0▀97,0▀65,77▀
88█████████89█17█18██17██88,89▀88█████16█28█41,28▀41,40▀41█41,7▀7██7,53▀65██77,78▀78██77,66▀97,66▀97,77▀77,65▀65█65,41▀▀7,4▀▀▀7,41▀7,4▀64,4▀64,40▀40,41▀64,77▀65,77▀77,97▀97,0▀78,0▀97█0,78▀▀0,97▀97,77▀
88█████████89█17█18██17██89█88████88,16▀88,89▀28,16▀28█40█41██53,7▀53█7█7,65▀65,78▀78,0▀78█78,77▀78█66,65▀77,66▀77█65█64,77▀64█64,4▀▀▀▀4██40█65,64▀77,65▀77,76▀97,78▀97,77▀▀77██77,97▀78,0▀77,0▀
88█████████89█17█18██17██89,88▀88███88,16▀41,65▀65,77▀17,65▀16,93▀29,16▀41█41,7▀41█53,7▀7,41▀65█0,97▀97,78▀78█78,77▀▀77,65▀77█65,77▀65,64▀64,4▀4,40▀▀4██4,40▀4███40,28▀64█77,97▀0,97▀▀▀97,77▀78,77▀97,76▀77,64▀76,64▀
88█████████89█17█18█18,17▀17█89█88████29,41▀77,65▀65█97,0▀▀17,92▀28█41,29▀41█41,40▀7,41▀65,7▀78,77▀78█77█65,77▀▀65█77,66▀▀64,77▀40,4▀▀4███████40█64,7▀77,65▀77█76,97▀65,97▀65,0▀77,97▀▀▀65,77▀
88█████████89█17█18█17█17,89▀88████88,89▀41█65█77█0,97▀0█14,94▀29,17▀41█41,7▀41█40,41▀7█65████████77,65▀64█4████4,64▀4█4,64▀4█40█41,29▀76,64▀0,77▀▀0,65▀77,64▀77,7▀77,64▀65,41▀40,64▀
88█████████89█17█18█17█89█89,88▀88███89█41█65█77█97,78▀0,97▀41,64▀28,7▀29,41▀7,41▀▀41█7,41▀77,64▀97,65▀78,65▀66,65▀77█78,77▀78,97▀▀77█64,41▀4,41▀64█64,77▀▀▀▀65,64▀7█64,7▀7█41,7▀7█7,65▀64,77▀65,77▀77█65,64▀64,7▀64█
88█████████89█17█18█17█89█88████89█41,7▀65█77██78█65,77▀64,41▀77,65▀65,97▀65█7,64▀41,65▀▀▀▀7█65,41▀▀▀65,29▀41,29▀7,29▀7,41▀▀65,41▀▀64,41▀7,41▀7██41,7▀7█53█65█77,65▀▀▀64,41▀64██64,41▀
88█████████89█17█19█17█89█88████89,16▀7,41▀65██77,65▀77█78,77▀95,77▀41█77,65▀77█77,76▀76,65▀65█7██7,41▀41█29█29,17▀29,16▀29,17▀41,29▀41███40,41▀41███41,29▀29█41,29▀7,41▀▀7███41,7▀7,65▀65,77▀64,77▀
88█████████89,88▀17█19█89█88████88,89▀16█29█7█65,7▀65█77,65▀77██65,77▀41,65▀64,65▀7,65▀7███53,7▀41█40,29▀16█16,17▀▀16█29,17▀41██41,40▀41███29,17▀16██17,16▀29,17▀41,29▀7,41▀7███65█77,65▀▀
88██████████17█19,17▀88█████16██29█41█7,41▀53,7▀65,7▀65,53▀65█████65,7▀7█7,41▀41,29▀29,16▀16,89▀17,89▀17,88▀▀88█89█17█29█40,29▀40█40,29▀29,16▀88██16,88▀16,89▀17,16▀29█41█7,41▀7██53█7██
88██████████17██88████88,89▀16,89▀16█29█29,41▀41██7,41▀7██53,7▀▀53,41▀7,41▀41█41,29▀29,17▀17█16,89▀88█████89█16█28,16▀28█28,17▀16█89,16▀88████89,88▀17█29,17▀41,29▀▀▀41█7,41▀7█
88██████████89██88████89,88▀16,88▀16,89▀29,17▀41█████7,41▀▀41,29▀29█29,17▀17████18,19▀89,88▀88████89,16▀28,29▀28,40▀40█29,40▀29█17,28▀89,18▀88,89▀88███89,88▀17,89▀17██29██41██
88██████████89█88,89▀88██████88,89▀17█29█41,29▀▀▀41█41,29▀29█17█17,89▀▀16,17▀89,18▀17█18█19█88█████16,89▀28█40████28█18██89,18▀88████89,88▀17,16▀17█29,17▀▀41,29▀
88██████████89█89,88▀88██████89,88▀16█29,17▀28,17▀29██29,17▀29,16▀17,16▀88█16,88▀89,88▀18,88▀19,88▀17,88▀19,88▀17,88▀88██████16,88▀28,16▀▀40,28▀40,16▀17,16▀18,17▀18██89,19▀88█████89,88▀▀17,89▀29,17▀
88██████████89█88,89▀88██████88,89▀16,89▀16█28,16▀▀17,88▀89,88▀▀88███████████████88,89▀88,16▀88,89▀88██████89,88▀▀88████████
42,1▀▀30,1▀19,1▀88,1▀▀▀▀▀▀89,1▀▀88,1▀▀▀▀▀▀▀89,1▀88,1▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀16,1▀88,1▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀

View File

@ -1 +0,0 @@
{"_default": {"2": {"mask": "*!uid677411@2266CC43:5D1E0B1F:A5B0466C:IP", "permission": "admin"}, "3": {"mask": "sad!*@*", "permission": "", "timestamp": "2025-02-14T16:05:44.177666"}, "4": {"mask": "Day!~Day@194.44.50.9", "permission": "ignore", "timestamp": "2025-02-14T16:05:53.633575"}}}

View File

View File

@ -1,194 +0,0 @@
# -*- coding: utf-8 -*-
"""
======================================================
:mod:`irc3.plugins.asynchronous` Asynchronous 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.
Example:
class MyPlugin:
def __init__(self, bot):
self.bot = bot
self.whois = Whois(bot)
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::
Always verify `result['timeout']` to ensure a response was received before the timeout.
"""
from asynchronous import AsyncEvents
from irc3 import utils
from irc3 import dec
class Whois(AsyncEvents):
"""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.
"""
timeout = 20
send_line = 'WHOIS {nick} {nick}'
events = (
{'match': r"(?i)^:\S+ 301 \S+ {nick} :(?P<away>.*)"},
{
'match': (
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+) :(?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+ 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,
},
)
def process_results(self, results=None, **value):
"""Aggregate and structure WHOIS results into a consolidated dictionary.
Args:
results (list): Collected event responses.
**value: Accumulated data from event processing.
Returns:
dict: Structured user information with success status.
"""
channels = []
for res in results:
channels.extend(res.pop('channels', '').split())
value.update(res)
value['channels'] = channels
value['success'] = value.get('retcode') == '318'
return value
class WhoChannel(AsyncEvents):
"""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+) (?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},
)
def process_results(self, results=None, **value):
"""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(f"{res['nick']}!{res['user']}@{res['host']}")
users.append(res)
value['users'] = users
value['success'] = value.get('retcode') == '315'
return value
@dec.plugin
class Async:
"""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)
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)
@dec.extend
def whois(self, nick: str, timeout: int = 20):
"""Initiate a WHOIS query for a nickname.
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: str, timeout: int = 20):
"""Perform a WHO query on a channel or user.
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('#'):
return self.async_who_channel(channel=target, timeout=timeout)
return None
@dec.extend
async def names(self, target: str):
"""Retrieve the list of users in a channel."""
result = await self.who(target)
if result and 'users' in result:
return {'names': result['users']}
return {'names': []}

View File

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

View File

@ -1,109 +0,0 @@
# -*- 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:
"""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 per detection
self.flood_message = '\u00A0\u2002\u2003' * 50 # Invisible characters
@irc3.event(irc3.rfc.PRIVMSG)
async def on_privmsg(self, mask, event, target, data):
"""
Listens for messages and floods the channel if the sender is the target nick.
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.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, self.flood_message)
await asyncio.sleep(0.1) # Prevents immediate rate-limiting
@command(permission='admin', public=True)
def disregard(self, mask, target, args):
"""
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:
return "Usage: !disregard <nick>"
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 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.")

View File

@ -1,123 +0,0 @@
"""
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
self.goat_tasks = {}
@command(permission='admin')
def goat(self, mask, target, args):
"""
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>]
"""
nick = args.get("<nick>")
if target in self.goat_tasks:
self.bot.privmsg(target, "A goat task is already running.")
return
try:
with open('goat.txt', 'r', encoding='utf-8') as file:
lines = file.readlines()
except Exception as e:
self.bot.privmsg(target, f"Error reading goat.txt: {e}")
return
task = self.bot.loop.create_task(self.send_lines(target, nick, lines))
self.goat_tasks[target] = task
@command(permission='admin')
def goatstop(self, mask, target, args):
"""
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:
task = self.goat_tasks.pop(target)
task.cancel()
self.bot.privmsg(target, "Goat task stopped.")
else:
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()
msg = f"{nick} : {stripped_line}" if nick else stripped_line
self.bot.privmsg(target, msg)
message_count += 1
await asyncio.sleep(0.007)
except asyncio.CancelledError:
self.bot.privmsg(target, "Goat task cancelled.")
raise

View File

@ -1,127 +0,0 @@
# -*- 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
# Pre-define the list of Unicode combining characters.
COMBINING_CHARS = [
'\u0300', '\u0301', '\u0302', '\u0303', '\u0304', '\u0305',
'\u0306', '\u0307', '\u0308', '\u0309', '\u030A', '\u030B',
'\u030C', '\u030D', '\u030E', '\u030F', '\u0310', '\u0311',
'\u0312', '\u0313', '\u0314', '\u0315', '\u0316', '\u0317',
'\u0318', '\u0319', '\u031A', '\u031B', '\u031C', '\u031D',
'\u031E', '\u031F', '\u0320', '\u0321', '\u0322', '\u0323',
'\u0324', '\u0325', '\u0326', '\u0327', '\u0328', '\u0329',
'\u032A', '\u032B', '\u032C', '\u032D', '\u032E', '\u032F',
'\u0330', '\u0331', '\u0332', '\u0333', '\u0334', '\u0335',
'\u0336', '\u0337', '\u0338', '\u0339', '\u033A', '\u033B',
'\u033C', '\u033D', '\u033E', '\u033F', '\u0340', '\u0341',
'\u0342', '\u0343', '\u0344', '\u0345', '\u0346', '\u0347',
'\u0348', '\u0349', '\u034A', '\u034B', '\u034C', '\u034D',
'\u034E', '\u034F', '\u0350', '\u0351', '\u0352', '\u0353',
'\u0354', '\u0355', '\u0356', '\u0357', '\u0358', '\u0359',
'\u035A', '\u035B', '\u035C', '\u035D', '\u035E', '\u035F',
'\u0360', '\u0361', '\u0362'
]
@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
def add_combining_characters(self, char):
"""Add random combining characters (with style and color codes) to a character."""
glitched_char = char
# Append between 1 and 3 randomly styled combining characters.
for _ in range(random.randint(1, 3)):
color = random.randint(0, 15)
style = random.choice(['\x02', '\x1F']) # Bold or Underline.
combining_char = random.choice(COMBINING_CHARS)
glitched_char += f'\x03{color}{style}{combining_char}\x0F'
return glitched_char
def style_message(self, message):
"""Apply glitch styling to each character in the message."""
white_color_code = '\x0300' # IRC color code for white.
styled_chars = [
f'{white_color_code}{self.add_combining_characters(c)}'
for c in message
]
return ''.join(styled_chars).strip()
@irc3.event(irc3.rfc.PRIVMSG)
def on_privmsg(self, mask, event, target, data):
"""If a message is received from the target user, repeat it with optional glitch styling."""
if self.target and mask.nick == self.target:
if self.unicode_mode:
data = self.style_message(data)
self.bot.privmsg(target, data)
@command(permission='admin', public=True)
def imitate(self, mask, target, args):
"""
%%imitate [--stop] [--unicode] [<nick>]
Options:
--stop Stop imitating.
--unicode Enable Unicode glitch styling.
"""
stop = args.get('--stop')
nick = args.get('<nick>')
unicode = args.get('--unicode')
if stop:
self.target = None
self.unicode_mode = False
self.bot.privmsg(target, "Stopped imitating.")
return
if not nick:
self.bot.privmsg(target, "Error: You must specify a nick to imitate.")
return
self.target = nick
self.unicode_mode = unicode
self.bot.send_line(f'NICK {nick}_')
if self.unicode_mode:
self.bot.privmsg(target, f"Now imitating {nick} with Unicode glitches!")
else:
self.bot.privmsg(target, f"Now imitating {nick}!")

View File

@ -1,113 +0,0 @@
# -*- 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
# 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',
'yellow': '\x0308', 'light_green': '\x0309', 'cyan': '\x0310', 'light_cyan': '\x0311',
'light_blue': '\x0312', 'pink': '\x0313', 'grey': '\x0314', 'light_grey': '\x0315'
}
@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
def matrix(self, mask, target, args):
"""
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(80, 120)
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))
line = self.colorize(line)
matrix_lines.append(line)
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()))
colored_text += f"{color}{char}"
return colored_text

View File

@ -1,249 +0,0 @@
# -*- coding: utf-8 -*-
"""
IRC3 Bot Plugin: YouTube Video Information Fetcher
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
from irc3.plugins.command import command
from plugins.services.permissions import check_ignore
# Constants for YouTube API
API_SERVICE_NAME = "youtube"
API_VERSION = "v3"
DEVELOPER_KEY = "AIzaSyBNrqOA0ZIziUVLYm0K5W76n9ndqz6zTxI"
# Initialize YouTube API client
youtube = googleapiclient.discovery.build(
API_SERVICE_NAME, API_VERSION, developerKey=DEVELOPER_KEY
)
# Regular expression for matching YouTube video URLs
YT_URL_REGEX = (
r"((?:https?:)?\/\/)?((?:www|m)\.)?((?:youtube(-nocookie)?\.com|youtu\.be))"
r"(\/(?:[\w\-]+\?v=|embed\/|live\/|v\/)?)([\w\-]+)(\S+)?"
)
@irc3.plugin
class YouTubePlugin:
"""
An IRC plugin to fetch and display YouTube video information.
This plugin responds to both command inputs and messages containing YouTube links.
"""
def __init__(self, bot):
"""
Initialize the YouTubePlugin with an IRC bot instance.
:param bot: IRC bot instance
"""
self.bot = bot
self.yt_reg = re.compile(YT_URL_REGEX)
def parse_duration(self, duration):
"""
Parse ISO 8601 duration format into a human-readable string.
:param duration: ISO 8601 duration string
:return: Formatted duration string
"""
duration = duration[2:] # Remove 'PT' prefix
hours = minutes = seconds = 0
if 'H' in duration:
hours, duration = duration.split('H', 1)
hours = int(hours)
if 'M' in duration:
minutes, duration = duration.split('M', 1)
minutes = int(minutes)
if 'S' in duration:
seconds = int(duration.split('S', 1)[0])
parts = []
if hours > 0:
parts.extend([str(hours), f"{minutes:02}", f"{seconds:02}"])
elif minutes > 0:
parts.extend([str(minutes), f"{seconds:02}"])
else:
parts.append(str(seconds))
return ":".join(parts)
def format_number(self, num):
"""
Format numbers to use K for thousands and M for millions.
:param num: Integer number to format
:return: Formatted string
"""
if num >= 1_000_000:
return f"{num/1_000_000:.1f}M"
elif num >= 1_000:
return f"{num/1_000:.1f}k"
return str(num)
async def get_video_info(self, video_id):
"""
Retrieve video information from YouTube API.
:param video_id: ID of the video
:return: Dictionary with video info or None if data couldn't be fetched
"""
try:
request = youtube.videos().list(
part="contentDetails,statistics,snippet",
id=video_id
)
response = await self.bot.loop.run_in_executor(None, request.execute)
if not response.get("items"):
return None
item = response["items"][0]
stats = item["statistics"]
snippet = item["snippet"]
details = item["contentDetails"]
info = {
"title": html.unescape(snippet["title"]).title(),
"duration": self.parse_duration(details["duration"]),
"views": int(stats.get("viewCount", 0)),
"likes": int(stats.get("likeCount", 0)),
"comments": int(stats.get("commentCount", 0)),
"channel": snippet.get("channelTitle", "Unknown Channel"),
"videoId": video_id,
}
if published := snippet.get("publishedAt"):
dt = datetime.datetime.fromisoformat(published.replace("Z", "+00:00"))
info["date"] = dt.strftime("%b %d, %Y")
else:
info["date"] = "Unknown Date"
return info
except Exception as e:
self.bot.log.error(f"YouTube API error: {e}")
return None
def format_message(self, info):
"""
Format video information into an attractive mIRC color-coded message.
:param info: Dictionary containing video information
:return: Formatted message string
"""
return (
f"\x0300,04\x0315Youtube\x03\x03\x1F{info['title']}\x0F\x03 | "
f"\x0308Duration:\x03 \x0307\x02{info['duration']}\x0F\x03 | "
f"\x02Views:\x02 \x0309{self.format_number(info['views'])}\x0F\x03 | "
f"\x0306Published:\x03 \x0314{info['date']}\x0F\x03 | "
f"\x0312https://youtu.be/{info['videoId']}\x0F\x03"
)
@command(options_first=True, use_shlex=True, aliases=['y'],)
async 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.
"""
tokens = args["<search_query>"]
# Default to one result if no flag is provided.
amount = 1
# Separate amount flag from the rest of the tokens
new_tokens = []
for token in tokens:
if token.startswith("--"):
try:
amount = int(token[2:])
except ValueError:
self.bot.privmsg(target, "Invalid amount specified. Using default of 1 result.")
else:
new_tokens.append(token)
query = " ".join(new_tokens)
if not query:
self.bot.privmsg(target, "You must specify a search query.")
return
try:
request = youtube.search().list(
part="id,snippet",
q=query,
type="video",
maxResults=amount
)
result = await self.bot.loop.run_in_executor(None, request.execute)
if not result.get("items"):
self.bot.privmsg(target, "No results found.")
return
# If more than one result is requested, show a header.
if amount > 1:
self.bot.privmsg(target, f"\x02Search Results For:\x03 \x0311{query}\x03")
self.bot.privmsg(target, "" * 99)
count = 0
for item in result["items"]:
count += 1
video_id = item["id"]["videoId"]
if info := await self.get_video_info(video_id):
# Only number the results if more than one is returned.
if amount == 1:
msg = self.format_message(info)
else:
msg = f" {count}. {self.format_message(info)}"
self.bot.privmsg(target, msg)
if amount > 1:
self.bot.privmsg(target, "" * 99)
except Exception as e:
self.bot.privmsg(target, f"Search error: {e}")
@irc3.event(irc3.rfc.PRIVMSG)
@check_ignore
async def on_message(self, mask, event, target, data):
"""
Event handler for messages in the channel to detect and respond to YouTube links.
:param mask: Mask of the user who sent the message
:param event: Event object from IRC
:param target: Target channel or user where the message was sent
:param data: Content of the message
"""
if mask.nick == self.bot.nick:
return
for match in self.yt_reg.finditer(data):
video_id = match.group(6)
if info := await self.get_video_info(video_id):
self.bot.privmsg(target, self.format_message(info))

View File

@ -1,670 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
IRC Bot Plugin for Note Taking using TinyDB.
This plugin provides a single !note command with subcommands to manage both
daily and general notes. It supports the following subcommands:
today [--add|--replace] [<content>...]
View or update today's daily note. If content is provided and a note for
today already exists, a warning is shown unless you specify --add (to append)
or --replace (to replace).
daily [--date <date>] [--add|--replace] [<content>...]
View or update a daily note for the specified date (default is today). If
content is provided and a note already exists, a warning is shown unless you
specify --add (to append) or --replace (to replace).
general <title> [--tags <tags>] [<content>...]
View or update a general note with the given title.
Optionally specify tags.
list <type>
List all notes of the specified type (either 'daily' or 'general'). The
content of each note is displayed truncated.
search <type> <keyword>
Search for a keyword in notes of the specified type.
summary
Display a summary of note counts.
Notes are stored persistently in the TinyDB database file "notes.json". Daily
notes are keyed by date (YYYY-MM-DD) and general notes by a slugified title.
Dependencies:
- irc3
- ircstyle
- tinydb
Author: Your Name
Version: 1.3
Date: 2025-02-18
"""
import re
from datetime import datetime, date
from tinydb import TinyDB, Query
import irc3
from irc3.plugins.command import command
import ircstyle
import logging
# -----------------------------------------------------------------------------
# Helper Functions
# -----------------------------------------------------------------------------
def slugify(text):
"""
Convert text to a slug (lowercase, underscores, alphanumerics only).
Args:
text (str): The text to slugify.
Returns:
str: A slugified version of the text.
"""
text = text.lower()
text = re.sub(r'\s+', '_', text)
text = re.sub(r'[^a-z0-9_-]', '', text)
return text
def now_str():
"""
Return the current datetime as an ISO formatted string.
Returns:
str: The current datetime.
"""
return datetime.now().isoformat()
def today_str():
"""
Return today's date in ISO format (YYYY-MM-DD).
Returns:
str: Today's date.
"""
return date.today().isoformat()
def default_daily_template(day):
"""
Generate a default daily note template.
Args:
day (str): The date string (YYYY-MM-DD).
Returns:
str: Default content for a daily note.
"""
return f"""# Daily Note for {day}
## To do:
## Tomorrow:
## Reminder:
## Journal:
"""
def default_general_template(title, tags):
"""
Generate a default general note template.
Args:
title (str): The note title.
tags (str): Tags for the note.
Returns:
str: Default content for a general note.
"""
tag_line = f"Tags: {tags}" if tags else ""
return f"""# General Note: {title}
{tag_line} Created on: {now_str()}
"""
def truncate(text, length=50):
"""
Truncate the text to a specified length, appending "..." if truncated.
Args:
text (str): The text to truncate.
length (int): Maximum length of the returned text.
Returns:
str: The truncated text.
"""
if not isinstance(text, str):
text = str(text)
return text if len(text) <= length else text[:length] + "..."
# -----------------------------------------------------------------------------
# NoteTaking Plugin
# -----------------------------------------------------------------------------
@irc3.plugin
class NoteTaking:
"""
IRC3 plugin for unified note taking.
Provides a single !note command with subcommands to create, update, view,
list, and search both daily and general notes stored in TinyDB.
"""
requires = [
'plugins.users', # Ensure userlist plugin is loaded, if using user-specific notes
]
def __init__(self, bot):
"""
Initialize the NoteTaking plugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
# Configuration options with defaults
self.db_filename = self.bot.config.get('note_db_filename', 'notes.json')
self.daily_template = self.bot.config.get('daily_template', default_daily_template)
self.general_template = self.bot.config.get('general_template', default_general_template)
try:
self.db = TinyDB(self.db_filename)
self.daily_table = self.db.table('daily')
self.general_table = self.db.table('general')
except Exception as e:
self.bot.log.error(f"Error initializing TinyDB: {e}")
raise # Re-raise to prevent the plugin from loading
self.User = Query()
self.log = logging.getLogger(__name__)
@command(aliases=['notes'], public=True, options_first=False, use_shlex=True)
async def note(self, mask, target, args):
"""
%%note today [--add|--replace] [<content>...]
%%note daily [--date=<date>] [--add|--replace] [<content>...]
%%note general <title> [--tags=<tags>] [<content>...]
%%note list [<type>]
%%note search <type> <keyword>
%%note summary
%%note
"""
# If no subcommand is provided, show the help message.
if not any(args.get(key) for key in ('today', 'daily', 'general', 'list', 'search', 'summary')):
# If there's content provided without a subcommand, treat it as a daily note update.
if args.get('<content>'):
await self._note_today(mask, target, args)
else:
self._show_help(target)
return
try:
if args.get('today'):
await self._note_today(mask, target, args)
elif args.get('daily'):
await self._note_daily(mask, target, args)
elif args.get('general'):
await self._note_general(mask, target, args)
elif args.get('list'):
await self._note_list(target, args)
elif args.get('search'):
await self._note_search(target, args)
elif args.get('summary'):
await self._note_summary(target)
except Exception as e:
self.bot.log.error(f"Error processing note command: {e}", exc_info=True)
self.bot.privmsg(
target,
ircstyle.style(f"❌ Error processing command: {e}", fg="red", bold=True)
)
def _show_help(self, target):
help_lines = [
ircstyle.style("Usage: !note <subcommand> [options]", fg="red", bold=True),
"Subcommands:",
f"{ircstyle.style(' today [--add|--replace] [<content>...]', fg='yellow')} - View or update today's daily note.",
f"{ircstyle.style(' daily [--date <date>] [--add|--replace] [<content>...]', fg='yellow')} - View or update a daily note for a specific date (default is today).",
f"{ircstyle.style(' general <title> [--tags <tags>] [<content>...]', fg='yellow')} - View or update a general note with the given title. Optionally, specify tags.",
f"{ircstyle.style(' list [daily|general]', fg='yellow')} - List all notes of a specific type with truncated content.",
f"{ircstyle.style(' search <type> <keyword>', fg='yellow')} - Search for a keyword in notes of a specific type.",
f"{ircstyle.style(' summary', fg='yellow')} - Display a summary of note counts.",
]
for line in help_lines:
self.bot.privmsg(target, line)
async def _note_today(self, mask, target, args):
"""
Handle the 'today' subcommand for daily notes.
If content is provided, update today's note.
If a note already exists and content is provided, warn the user unless
--add or --replace is specified.
"""
day = today_str()
content_list = args.get('<content>')
try:
note = self._get_daily_note(day)
if content_list:
new_content = " ".join(content_list)
# Check if both flags are provided
if args.get('--add') and args.get('--replace'):
self.bot.privmsg(
target,
ircstyle.style("❌ Error: Cannot use both --add and --replace simultaneously. Please choose one.", fg="red", bold=True)
)
return
if note:
# If note exists but neither flag is provided, warn the user.
if not (args.get('--add') or args.get('--replace')):
warning = ircstyle.style(
f"⚠️ Warning: Daily note for {day} already exists. Use --add to append to it or --replace to overwrite it.", fg="red", bold=True
)
self.bot.privmsg(target, warning)
return
if args.get('--add'):
appended_content = note.get('content') + "\n" + new_content
self.daily_table.update(
{'content': appended_content, 'updated': now_str()},
self.User.date == day
)
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"Daily note for {day} updated (content added).", fg="blue", bold=True
)
elif args.get('--replace'):
self.daily_table.update(
{'content': new_content, 'updated': now_str()},
self.User.date == day
)
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"Daily note for {day} replaced.", fg="blue", bold=True
)
else:
# No existing note; create a new one.
self.daily_table.insert({
'date': day,
'content': new_content,
'created': now_str(),
'updated': now_str()
})
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"Daily note for {day} created.", fg="blue", bold=True
)
else:
if not note:
default_content = self.daily_template(day)
self.daily_table.insert({
'date': day,
'content': default_content,
'created': now_str(),
'updated': now_str()
})
note = self._get_daily_note(day)
msg = ircstyle.style(
f"📝 Daily Note for {day}:\n", fg="yellow", bold=True
) + note.get('content')
self.bot.privmsg(target, msg)
except Exception as e:
self.bot.log.error(f"Error in _note_today: {e}", exc_info=True)
self.bot.privmsg(
target,
ircstyle.style(f"❌ Error: {e}", fg="red", bold=True)
)
async def _note_daily(self, mask, target, args):
"""
Handle the 'daily' subcommand for daily notes with an optional date.
If content is provided, update the note.
If a note already exists and content is provided, warn the user unless
--add or --replace is specified.
"""
day = args.get('--date') or today_str()
try:
datetime.strptime(day, '%Y-%m-%d') # Validate date format
except ValueError:
self.bot.privmsg(
target,
ircstyle.style("❌ Invalid date format. Use YYYY-MM-DD.", fg="red", bold=True)
)
return
content_list = args.get('<content>')
try:
note = self._get_daily_note(day)
if content_list:
new_content = " ".join(content_list)
# Check if both flags are provided
if args.get('--add') and args.get('--replace'):
self.bot.privmsg(
target,
ircstyle.style("❌ Error: Cannot use both --add and --replace simultaneously. Please choose one.", fg="red", bold=True)
)
return
if note:
if not (args.get('--add') or args.get('--replace')):
self.bot.privmsg(
target,
ircstyle.style(f"⚠️ Warning: Daily note for {day} already exists. Use --add to append or --replace to overwrite.", fg="red", bold=True)
)
return
if args.get('--add'):
appended_content = note.get('content') + "\n" + new_content
self.daily_table.update(
{'content': appended_content, 'updated': now_str()},
self.User.date == day
)
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"Daily note for {day} updated (content added).", fg="blue", bold=True
)
elif args.get('--replace'):
self.daily_table.update(
{'content': new_content, 'updated': now_str()},
self.User.date == day
)
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"Daily note for {day} replaced.", fg="blue", bold=True
)
else:
# Create new note if none exists.
self.daily_table.insert({
'date': day,
'content': new_content,
'created': now_str(),
'updated': now_str()
})
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"Daily note for {day} created.", fg="blue", bold=True
)
else:
if not note:
default_content = self.daily_template(day)
self.daily_table.insert({
'date': day,
'content': default_content,
'created': now_str(),
'updated': now_str()
})
note = self._get_daily_note(day)
msg = ircstyle.style(
f"📝 Daily Note for {day}:\n", fg="yellow", bold=True
) + note.get('content')
self.bot.privmsg(target, msg)
except Exception as e:
self.bot.log.error(f"Error in _note_daily: {e}", exc_info=True)
self.bot.privmsg(
target,
ircstyle.style(f"❌ Error: {e}", fg="red", bold=True)
)
async def _note_general(self, mask, target, args):
"""
Handle the 'general' subcommand for general notes.
If content is provided, update the note; otherwise, display it.
"""
nick = mask.nick # Get the user's nickname
# Retrieve <title> as a string (not a list)
title = args.get('<title>')
if not title:
self.bot.privmsg(
target,
ircstyle.style("", fg="red")
+ ircstyle.style("Title is required.", fg="red", bold=True)
)
return
slug = slugify(title)
tags = args.get('--tags') or ""
content_list = args.get('<content>')
try:
# Now filter by owner (nick) so users only access their own general notes
note = self._get_general_note(slug, nick)
if content_list:
new_content = " ".join(content_list)
update_fields = {'content': new_content, 'updated': now_str()}
if tags:
update_fields['tags'] = tags
if note:
self.general_table.update(
update_fields,
(self.User.slug == slug) & (self.User.owner == nick)
)
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"General note '{title}' updated.", fg="blue", bold=True
)
else:
self.general_table.insert({
'title': title,
'slug': slug,
'tags': tags,
'content': new_content,
'created': now_str(),
'updated': now_str(),
'owner': nick # Store the owner of the note
})
msg = ircstyle.style("", fg="green") + ircstyle.style(
f"General note '{title}' created.", fg="blue", bold=True
)
else:
if not note:
default_content = self.general_template(title, tags)
self.general_table.insert({
'title': title,
'slug': slug,
'tags': tags,
'content': default_content,
'created': now_str(),
'updated': now_str(),
'owner': nick # Store the owner of the note
})
note = self._get_general_note(slug, nick)
header = ircstyle.style(
f"📝 General Note: {note.get('title')}\n", fg="yellow", bold=True
)
tag_line = ircstyle.style(
f"Tags: {note.get('tags')}\n", fg="cyan"
)
msg = header + tag_line + note.get('content')
self.bot.privmsg(target, msg)
except Exception as e:
self.bot.log.error(f"Error in _note_general: {e}", exc_info=True)
self.bot.privmsg(
target,
ircstyle.style(f"❌ Error: {e}", fg="red", bold=True)
)
async def _note_list(self, target, args):
"""
Handle the 'list' subcommand to list notes with truncated content.
Usage:
%%note list [<type>]
where <type> is optional and can be 'daily' or 'general'.
If <type> is not provided, both daily and general notes are listed.
"""
note_type = args.get('<type>')
try:
if not note_type:
# List both daily and general notes with content truncated.
daily_notes = self.daily_table.all()
general_notes = self.general_table.all()
if not daily_notes and not general_notes:
msg = ircstyle.style("No notes found.", fg="red", bold=True)
else:
msg_lines = [ircstyle.style("📝 All Notes:", fg="blue", bold=True)]
if daily_notes:
msg_lines.append(ircstyle.style("📅 Daily Notes:", fg="yellow", bold=True))
# Sort daily notes by date.
daily_sorted = sorted(
[n for n in daily_notes if isinstance(n.get('date'), str)],
key=lambda x: x['date']
)
for note in daily_sorted:
date_str = note.get('date')
snippet = truncate(note.get('content', ''))
msg_lines.append(f"{ircstyle.style(date_str, fg='green')}: {ircstyle.style(snippet, fg='white')}")
if general_notes:
msg_lines.append(ircstyle.style("📋 General Notes:", fg="cyan", bold=True))
# Sort general notes by title.
general_sorted = sorted(
[n for n in general_notes if isinstance(n.get('title'), str)],
key=lambda x: x['title']
)
for note in general_sorted:
title = note.get('title')
snippet = truncate(note.get('content', ''))
msg_lines.append(f"{ircstyle.style(title, fg='green')}: {ircstyle.style(snippet, fg='white')}")
msg = "\n".join(msg_lines)
elif note_type == 'daily':
notes = self.daily_table.all()
valid_notes = [n for n in notes if isinstance(n.get('date'), str)]
if not valid_notes:
msg = ircstyle.style("No daily notes found.", fg="red", bold=True)
else:
msg_lines = [ircstyle.style("📅 Daily Notes:", fg="blue", bold=True)]
daily_sorted = sorted(valid_notes, key=lambda x: x['date'])
for note in daily_sorted:
date_str = note.get('date')
snippet = truncate(note.get('content', ''))
msg_lines.append(f"{ircstyle.style(date_str, fg='green')}: {ircstyle.style(snippet, fg='white')}")
msg = "\n".join(msg_lines)
elif note_type == 'general':
notes = self.general_table.all()
valid_notes = [n for n in notes if isinstance(n.get('title'), str)]
if not valid_notes:
msg = ircstyle.style("No general notes found.", fg="red", bold=True)
else:
msg_lines = [ircstyle.style("📋 General Notes:", fg="blue", bold=True)]
general_sorted = sorted(valid_notes, key=lambda x: x['title'])
for note in general_sorted:
title = note.get('title')
snippet = truncate(note.get('content', ''))
msg_lines.append(f"{ircstyle.style(title, fg='green')}: {ircstyle.style(snippet, fg='white')}")
msg = "\n".join(msg_lines)
else:
msg = ircstyle.style("Usage: !note list [daily|general]", fg="red", bold=True)
for line in msg.split('\n'):
self.bot.privmsg(target, line)
except Exception as e:
self.bot.log.error(f"Error in _note_list: {e}", exc_info=True)
self.bot.privmsg(target, ircstyle.style(f"❌ Error: {e}", fg="red", bold=True))
async def _note_search(self, target, args):
"""
Handle the 'search' subcommand to search notes by keyword.
Usage:
%%note search <type> <keyword>
"""
note_type = args.get('<type>')
keyword = args.get('<keyword>')
if note_type not in ('daily', 'general') or not keyword:
msg = ircstyle.style("Usage: !note search <daily|general> <keyword>", fg="red", bold=True)
self.bot.privmsg(target, msg)
return
try:
if note_type == 'daily':
notes = self.daily_table.search(self.User.content.search(keyword, flags=re.IGNORECASE))
header = "Daily"
else:
notes = self.general_table.search(self.User.content.search(keyword, flags=re.IGNORECASE))
header = "General"
if not notes:
msg = ircstyle.style(f"No {note_type} notes found with '{keyword}'.", fg="red", bold=True)
else:
results = []
for n in notes:
identifier = n.get('date') if note_type == 'daily' else n.get('title')
content_snippet = n.get('content')[:50] + "..." # Show a snippet of the content
results.append(f"{identifier} ({content_snippet})")
msg = ircstyle.style(f"{header} Notes matching '{keyword}':\n", fg="blue", bold=True) + "\n".join(results)
for line in msg.split('\n'):
self.bot.privmsg(target, line)
except Exception as e:
self.bot.log.error(f"Error in _note_search: {e}", exc_info=True)
self.bot.privmsg(
target,
ircstyle.style(f"❌ Error: {e}", fg="red", bold=True)
)
async def _note_summary(self, target):
"""
Handle the 'summary' subcommand to display note counts.
"""
try:
daily_count = len(self.daily_table.all())
general_count = len(self.general_table.all())
lines = [
ircstyle.style("===== NOTE SUMMARY =====", fg="teal", bold=True),
ircstyle.style(f"Daily notes: {daily_count}", fg="yellow"),
ircstyle.style(f"General notes: {general_count}", fg="cyan"),
ircstyle.style("===== END SUMMARY =====", fg="teal", bold=True)
]
for line in lines:
self.bot.privmsg(target, line)
except Exception as e:
self.bot.log.error(f"Error in _note_summary: {e}", exc_info=True)
self.bot.privmsg(
target,
ircstyle.style(f"❌ Error: {e}", fg="red", bold=True)
)
def _get_daily_note(self, day):
"""
Retrieve a daily note for the given day.
Args:
day (str): The date (YYYY-MM-DD).
Returns:
dict or None: The note if found, else None.
"""
try:
results = self.daily_table.search(self.User.date == day)
return results[0] if results else None
except Exception as e:
self.bot.log.error(f"Error in _get_daily_note: {e}", exc_info=True)
return None
def _get_general_note(self, slug, owner=None):
"""
Retrieve a general note by its slug and (optionally) owner.
Args:
slug (str): The slugified title.
owner (str, optional): The owner of the note.
Returns:
dict or None: The note if found, else None.
"""
try:
if owner:
results = self.general_table.search((self.User.slug == slug) & (self.User.owner == owner))
else:
results = self.general_table.search(self.User.slug == slug)
return results[0] if results else None
except Exception as e:
self.bot.log.error(f"Error in _get_general_note: {e}", exc_info=True)
return None

View File

@ -1,147 +0,0 @@
# -*- 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."""
return nick.lstrip('@+%&~!') if nick else ''
@irc3.plugin
class MassMessagePlugin:
"""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.
Args:
message (str): The message to send.
total (int): The total number of recipients.
"""
while True:
try:
nick = await self.queue.get()
try:
if nick == "FUCKYOU":
nick = "Zodiac"
self.bot.privmsg(nick, message)
self.count += 1
self.bot.log.info(f"Sent to {nick} ({self.count}/{total})")
except Exception as e:
self.bot.log.error(f"Failed to message {nick}: {e}")
finally:
self.queue.task_done() # Mark the task as done after processing
if self.delay > 0:
await asyncio.sleep(self.delay)
except asyncio.CancelledError:
# Exit silently on cancellation
break
except Exception as e:
self.bot.log.error(f"Worker error: {e}")
break
@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.
%%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."
message = ' '.join(args['<message>'])
workers = [] # Ensure workers is defined in the scope of the try block
try:
# Fetch the list of users in the channel
result = await self.bot.async_cmds.names(target)
nicknames = [strip_nick_prefix(n) for n in result['names']]
recipients = [n for n in nicknames if n != self.bot.nick]
if not recipients:
return "No valid recipients found."
total = len(recipients)
self.count = 0 # Reset the counter for this run
# Add all recipients to the queue
for nick in recipients:
await self.queue.put(nick)
# Create worker tasks
workers = [asyncio.create_task(self._worker(message, total)) for _ in range(1)]
# Send initial confirmation
self.bot.privmsg(target, f"Starting mass message to {total} users...")
# Wait for the queue to be fully processed
await self.queue.join()
# Cancel worker tasks after the queue is empty
for task in workers:
task.cancel()
# Wait for workers to finish cancellation
await asyncio.gather(*workers, return_exceptions=True)
return f"Mass message completed. Sent to {self.count}/{total} users."
except asyncio.CancelledError:
self.bot.log.info("Mass message command cancelled. Cleaning up workers.")
# Cancel any existing worker tasks
for task in workers:
task.cancel()
# Allow workers to handle cancellation
await asyncio.gather(*workers, return_exceptions=True)
# Re-raise the cancellation to inform the bot framework
raise
except Exception as e:
self.bot.log.error(f"Error in msgall: {e}")
return f"Mass message failed: {str(e)}"

View File

@ -1,200 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
IRC Bot Plugin for Tracking User Activity with TinyDB
This module implements an IRC bot plugin to track various activities of users in an IRC channel,
including their join times, messages sent, and nickname changes. The plugin uses TinyDB for
persistent storage of user data.
Features:
- Tracks when users join a channel.
- Logs user messages with timestamps.
- Monitors nickname changes.
- Provides a '!seen' command to retrieve the last activity of a specified user.
Dependencies:
- irc3: For IRC protocol handling.
- tinydb: For managing user data in a JSON file.
- ircstyle: For formatting IRC messages with color and style.
- logging: For error logging.
- humanize: For converting time differences into human-readable format.
Usage:
- Install required packages: pip install irc3 tinydb ircstyle humanize
- Use within an IRC bot framework that utilizes irc3 plugins.
Storage:
- User data is stored in 'seen.json' in the same directory as this script.
Author:
- Zodiac
Date:
- 02/18/2025
Note:
- Ensure your bot environment has write permissions for the JSON database file.
"""
from datetime import datetime
from tinydb import TinyDB, Query
import irc3
from irc3.plugins.command import command
import ircstyle
import logging
import humanize
def truncate(text, length=50):
"""Truncate text to specified length, appending "..." if truncated."""
if not isinstance(text, str):
text = str(text)
return text if len(text) <= length else text[:length] + "..."
@irc3.plugin
class UserActivityTracker:
"""IRC bot plugin to track and report user activity using TinyDB."""
requires = [
'plugins.users',
]
def __init__(self, bot):
self.bot = bot
self.db = TinyDB('seen.json')
self.users = self.db.table('users')
self.User = Query()
self.log = logging.getLogger(__name__)
def _add_unique_hostmask(self, nick, hostmask):
"""Add hostmask to user data if it's not already present and not 'unknown'."""
if hostmask == 'unknown':
return []
user = self.users.get(self.User.nick == nick.lower())
if user and 'hostmasks' in user:
if hostmask not in user['hostmasks']:
user['hostmasks'].append(hostmask)
return user['hostmasks']
else:
return [hostmask]
@irc3.event(r'(?P<mask>\S+)!.* NICK :?(?P<new_nick>\S+)')
def handle_nick_change(self, mask, new_nick, **kwargs):
"""Track nickname changes."""
try:
now = datetime.now().isoformat()
host = mask.split('!')[1] if '!' in mask else 'unknown'
self.users.upsert({
'nick': mask.nick.lower(),
'hostmasks': self._add_unique_hostmask(mask.nick.lower(), host),
'last_nick_change': {'old': mask.nick, 'new': new_nick, 'time': now}
}, self.User.nick == mask.nick.lower())
self.users.upsert({
'nick': new_nick.lower(),
'hostmasks': self._add_unique_hostmask(new_nick.lower(), host),
'last_join': None,
'last_message': None,
'last_nick_change': {'old': mask.nick, 'new': new_nick, 'time': now}
}, self.User.nick == new_nick.lower())
except Exception as e:
self.log.error(f"Error tracking nick change {mask.nick}{new_nick}: {e}")
@irc3.event(irc3.rfc.JOIN)
def handle_join(self, mask, channel, **kwargs):
"""Track when users join a channel."""
try:
now = datetime.now().isoformat()
self.users.upsert({
'nick': mask.nick.lower(),
'hostmasks': self._add_unique_hostmask(mask.nick.lower(), mask.host),
'last_join': now,
'last_message': None,
'last_nick_change': None
}, self.User.nick == mask.nick.lower())
except Exception as e:
self.log.error(f"Error tracking join for {mask.nick}: {e}")
@irc3.event(irc3.rfc.PRIVMSG)
def handle_message(self, target, data, mask, **kwargs):
"""Track user messages."""
try:
now = datetime.now().isoformat()
self.users.upsert({
'nick': mask.nick.lower(),
'hostmasks': self._add_unique_hostmask(mask.nick.lower(), mask.host),
'last_message': {'text': data, 'time': now}
}, self.User.nick == mask.nick.lower())
except Exception as e:
self.log.error(f"Error tracking message from {mask.nick}: {e}")
@command
async def seen(self, mask, target, args):
"""Retrieve a user's last activity.
%%seen <nick>
"""
try:
requested_nick = args['<nick>'].lower()
user = self.users.get(self.User.nick == requested_nick)
if not user:
msg = ircstyle.style(f"{requested_nick} has never been observed.", fg="red")
self.bot.privmsg(target, msg)
return
response = []
header = ircstyle.style(f"📊 Activity report for {requested_nick}:", fg="blue", bold=True)
response.append(header)
# Hostmasks
if 'hostmasks' in user and user['hostmasks']:
hostmasks_str = ", ".join(user['hostmasks'])
response.append(
ircstyle.style("🌐 Hostmasks: ", fg="cyan") +
ircstyle.style(hostmasks_str, fg="white")
)
# Last join - only if time exists
if 'last_join' in user and user['last_join']:
response.append(ircstyle.style("🕒 Last join: ", fg="cyan") +
ircstyle.style(humanize.naturaltime(datetime.now() - datetime.fromisoformat(user['last_join'])), fg="white"))
# Last message
if 'last_message' in user and user['last_message']:
try:
msg_text = truncate(user['last_message']['text'])
response.append(
ircstyle.style("💬 Last message: ", fg="green") +
ircstyle.style(humanize.naturaltime(datetime.now() - datetime.fromisoformat(user['last_message']['time'])), fg="grey") + " " +
ircstyle.style(msg_text, fg="white", italics=True)
)
except KeyError:
response.append(ircstyle.style("💬 Last message: ", fg="green") + ircstyle.style("Invalid or Missing Data", fg="white"))
# Last nick change
if 'last_nick_change' in user and user['last_nick_change']:
try:
old_nick = user['last_nick_change']['old']
new_nick = user['last_nick_change']['new']
response.append(
ircstyle.style(f"🆔 Nickname change: ", fg="purple") +
ircstyle.style(f"Changed from {old_nick} to {new_nick} ", fg="white") +
ircstyle.style(humanize.naturaltime(datetime.now() - datetime.fromisoformat(user['last_nick_change']['time'])), fg="white")
)
except KeyError:
response.append(ircstyle.style(f"🆔 Nickname change: ", fg="purple") + ircstyle.style("Invalid or Missing Data", fg="white"))
if len(response) == 1: # Only header present
response.append(ircstyle.style("No tracked activities.", fg="yellow"))
for line in response:
self.bot.privmsg(target, line)
except Exception as e:
self.log.error(f"Error in !seen command: {e}")
self.bot.privmsg(target, ircstyle.style("🚨 Internal error processing request.", fg="red", bold=True))

View File

@ -1,596 +0,0 @@
# -*- coding: utf-8 -*-
"""
IRC Bot Plugins: Join, Part, Nickname, Message, Topic, Invite, Voice, Kick, Ban, Op, and Mute Management
This module provides multiple IRC bot plugins:
1. `JoinPlugin`: Handles joining specific channels.
2. `PartPlugin`: Handles leaving specific channels.
3. `NickPlugin`: Handles changing the bot's nickname.
4. `MessagePlugin`: Handles sending messages.
5. `TopicPlugin`: Handles changing the channel topic.
6. `InvitePlugin`: Handles inviting users to a channel.
7. `VoicePlugin`: Handles granting and revoking voice (+v) privileges.
8. `KickPlugin`: Handles kicking users from the channel.
9. `BanPlugin`: Handles banning and unbanning users.
10. `OpPlugin`: Handles granting and removing operator privileges.
11. `MutePlugin`: Handles muting and unmuting users.
All commands require **admin** permissions.
Features:
- Join and leave specific channels.
- Change the bot's nickname.
- Send messages to users or channels.
- Change the channel topic.
- Invite users to channels.
- 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.
- Grant and remove operator privileges.
- Mute and unmute users.
Author: Zodiac (Modified for verbose logging by bot.log)
"""
import asyncio
import irc3
from irc3.plugins.command import command
@irc3.plugin
class JoinPlugin:
"""A plugin to join specific IRC channels."""
def __init__(self, bot):
"""
Initialize the JoinPlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def join(self, mask, target, args):
"""
Join a specific channel.
Usage:
%%join <channel>
"""
channel = args.get('<channel>')
if channel:
self.bot.log.info("Request to join channel %s", channel)
await self.join_channel(channel)
async def join_channel(self, channel):
"""
Join a specific channel.
Args:
channel (str): The IRC channel to join.
"""
self.bot.log.info("Joining channel %s", channel)
self.bot.send(f'JOIN {channel}')
@irc3.plugin
class PartPlugin:
"""A plugin to leave specific IRC channels."""
def __init__(self, bot):
"""
Initialize the PartPlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def part(self, mask, target, args):
"""
Leave a specific channel.
Usage:
%%part <channel>
"""
channel = args.get('<channel>')
if channel:
self.bot.log.info("Request to leave channel %s", channel)
await self.part_channel(channel)
async def part_channel(self, channel):
"""
Leave a specific channel.
Args:
channel (str): The IRC channel to leave.
"""
self.bot.log.info("Leaving channel %s", channel)
self.bot.send(f'PART {channel}')
@irc3.plugin
class NickPlugin:
"""A plugin to change the bot's nickname."""
def __init__(self, bot):
"""
Initialize the NickPlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def nick(self, mask, target, args):
"""
Change the bot's nickname.
Usage:
%%nick <newnick>
"""
newnick = args.get('<newnick>')
if newnick:
self.bot.log.info("Request to change nickname to %s", newnick)
await self.change_nick(newnick)
async def change_nick(self, newnick):
"""
Change the bot's nickname.
Args:
newnick (str): The new nickname for the bot.
"""
self.bot.log.info("Changing nickname to %s", newnick)
self.bot.send(f'NICK {newnick}')
@irc3.plugin
class MessagePlugin:
"""A plugin to send messages to users or channels."""
def __init__(self, bot):
"""
Initialize the MessagePlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def msg(self, mask, target, args):
"""
Send a message to a user or channel.
Usage:
%%msg <target> <message>
"""
msg_target = args.get('<target>')
message = args.get('<message>')
if msg_target and message:
self.bot.log.info("Request to send message to %s: %s", msg_target, message)
await self.send_message(msg_target, message)
async def send_message(self, msg_target, message):
"""
Send a message to a user or channel.
Args:
msg_target (str): The target user or channel.
message (str): The message to send.
"""
self.bot.log.info("Sending message to %s: %s", msg_target, message)
self.bot.send(f'PRIVMSG {msg_target} :{message}')
@irc3.plugin
class TopicPlugin:
"""A plugin to change the topic of an IRC channel."""
def __init__(self, bot):
"""
Initialize the TopicPlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def topic(self, mask, target, args):
"""
Change the topic of a channel.
Usage:
%%topic <newtopic>
"""
newtopic = args.get('<newtopic>')
if newtopic:
self.bot.log.info("Request to change topic to %s in %s", newtopic, target)
await self.change_topic(target, newtopic)
async def change_topic(self, target, newtopic):
"""
Change the topic of a channel.
Args:
target (str): The IRC channel.
newtopic (str): The new topic for the channel.
"""
self.bot.log.info("Changing topic in channel %s to %s", target, newtopic)
self.bot.send(f'TOPIC {target} :{newtopic}')
@irc3.plugin
class InvitePlugin:
"""A plugin to invite users to an IRC channel."""
def __init__(self, bot):
"""
Initialize the InvitePlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def invite(self, mask, target, args):
"""
Invite a user to a channel.
Usage:
%%invite <nick> <channel>
"""
nick = args.get('<nick>')
channel = args.get('<channel>')
if nick and channel:
self.bot.log.info("Request to invite %s to channel %s", nick, channel)
await self.invite_user(nick, channel)
async def invite_user(self, nick, channel):
"""
Invite a user to a channel.
Args:
nick (str): The nickname of the user.
channel (str): The IRC channel.
"""
self.bot.log.info("Inviting user %s to channel %s", nick, channel)
self.bot.send(f'INVITE {nick} {channel}')
@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', aliases=['v'])
async def voice(self, mask, target, args):
"""
Grant voice to a specific user or all users in the channel.
Usage:
%%voice [<nick>]
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to grant voice to %s in %s", nick, target)
await self.give_voice(target, nick)
else:
self.bot.log.info("Request to grant voice to all users in %s", target)
await self.give_voice_all(target)
@command(permission='admin')
async def devoice(self, mask, target, args):
"""
Remove voice from a specific user or all users in the channel.
Usage:
%%devoice [<nick>]
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to remove voice from %s in %s", nick, target)
await self.remove_voice(target, nick)
else:
self.bot.log.info("Request to remove voice from all users in %s", target)
await self.remove_voice_all(target)
async def give_voice(self, target, nick):
"""
Grant voice to a specific user.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Granting voice to user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} +v {nick}')
async def remove_voice(self, target, nick):
"""
Remove voice from a specific user.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Removing voice from user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} -v {nick}')
async def give_voice_all(self, target):
"""
Grant voice to all users in the channel who do not already have voice or op privileges.
Args:
target (str): The IRC channel.
"""
self.bot.log.info("Granting voice to all non-voiced users in channel %s", target)
result = await self.bot.async_cmds.names(target)
for user in result['names']:
nick = user.get('nick')
modes = user.get('modes', '')
# Check if the user already has op or voice based on the modes field.
if modes.endswith('@') or modes.endswith('+'):
continue
self.bot.log.info("Granting voice to user %s", nick)
self.bot.send(f'MODE {target} +v {nick}')
await asyncio.sleep(0.07) # Prevent server flooding
async def remove_voice_all(self, target):
"""
Remove voice from all users in the channel who currently have voice privileges.
Args:
target (str): The IRC channel.
"""
self.bot.log.info("Removing voice from all voiced users in channel %s", target)
result = await self.bot.async_cmds.names(target)
for user in result['names']:
nick = user.get('nick')
modes = user.get('modes', '')
# Remove voice only if the user has a voice prefix (and not op)
if modes.endswith('+'):
self.bot.log.info("Removing voice from user %s", nick)
self.bot.send(f'MODE {target} -v {nick}')
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 with an optional reason.
Usage:
%%kick <nick> [<reason>]
"""
nick = args.get('<nick>')
reason = args.get('<reason>') or 'Kicked by admin'
if nick:
self.bot.log.info("Request to kick user %s from channel %s for reason: %s", nick, target, reason)
await self.kick_user(target, nick, reason)
async def kick_user(self, target, nick, reason):
"""
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.log.info("Kicking user %s from channel %s", nick, target)
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.
Usage:
%%ban <nick>
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to ban user %s from channel %s", nick, target)
await self.ban_user(target, nick)
@command(permission='admin')
async def unban(self, mask, target, args):
"""
Unban a specific user from the channel.
Usage:
%%unban <nick>
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to unban user %s from channel %s", nick, target)
await self.unban_user(target, nick)
async def ban_user(self, target, nick):
"""
Ban a specific user from the channel.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Banning user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} +b {nick}')
async def unban_user(self, target, nick):
"""
Unban a specific user from the channel.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Unbanning user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} -b {nick}')
@irc3.plugin
class OpPlugin:
"""A plugin to grant and remove operator privileges in an IRC channel."""
def __init__(self, bot):
"""
Initialize the OpPlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def op(self, mask, target, args):
"""
Grant operator privileges to a specific user.
Usage:
%%op <nick>
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to grant operator privileges to %s in %s", nick, target)
await self.give_op(target, nick)
@command(permission='admin')
async def deop(self, mask, target, args):
"""
Remove operator privileges from a specific user.
Usage:
%%deop <nick>
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to remove operator privileges from %s in %s", nick, target)
await self.remove_op(target, nick)
async def give_op(self, target, nick):
"""
Grant operator privileges to a specific user.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Granting operator privileges to user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} +o {nick}')
async def remove_op(self, target, nick):
"""
Remove operator privileges from a specific user.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Removing operator privileges from user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} -o {nick}')
@irc3.plugin
class MutePlugin:
"""A plugin to mute and unmute users in an IRC channel."""
def __init__(self, bot):
"""
Initialize the MutePlugin.
Args:
bot (irc3.IrcBot): The IRC bot instance.
"""
self.bot = bot
@command(permission='admin')
async def mute(self, mask, target, args):
"""
Mute a specific user in the channel.
Usage:
%%mute <nick>
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to mute user %s in channel %s", nick, target)
await self.mute_user(target, nick)
@command(permission='admin')
async def unmute(self, mask, target, args):
"""
Unmute a specific user in the channel.
Usage:
%%unmute <nick>
"""
nick = args.get('<nick>')
if nick:
self.bot.log.info("Request to unmute user %s in channel %s", nick, target)
await self.unmute_user(target, nick)
async def mute_user(self, target, nick):
"""
Mute a specific user in the channel.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Muting user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} +q {nick}')
async def unmute_user(self, target, nick):
"""
Unmute a specific user in the channel.
Args:
target (str): The IRC channel.
nick (str): The nickname of the user.
"""
self.bot.log.info("Unmuting user %s in channel %s", nick, target)
self.bot.send(f'MODE {target} -q {nick}')

View File

@ -1,232 +0,0 @@
# -*- coding: utf-8 -*-
"""
IRC3 Anti-Spam Plugin with Auto-Kick, Auto-Ban, and Auto-Unban
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 and automatically unbanned after 5 minutes.
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 and expired bans every minute.
Author: [Your Name]
"""
import time
import irc3
from collections import defaultdict, deque
from irc3.plugins.cron import cron
from plugins.asynchronious import WhoChannel
@irc3.plugin
class AntiSpam:
"""A plugin for automatic spam detection and mitigation in IRC channels."""
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.active_bans = [] # Track active bans with timestamps for auto-removal
self.exclude_list = ['ZodBot','g1mp','scroll','WILDWEST','CANCER','aibird'] # Bots to ignore
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.get('success'):
for user in result['users']:
if user['nick'].lower() == nick.lower():
return user['modes']
except Exception as e:
self.bot.log.error(f"Error fetching user modes: {e}")
return ""
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.
"""
nick = nick.lower()
if nick not in self.user_data:
self.user_data[nick] = {
'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)))
}
user = self.user_data[nick]
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):
"""
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()
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 and {'o', '%', 'h', '@'} & set(user_modes):
return
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 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-minute window
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
ban_mask = f'*!{mask.host}'
self.bot.send(f"MODE {channel} +b {ban_mask}")
# Record ban for auto-removal
self.active_bans.append({
'ban_mask': ban_mask,
'channel': channel,
'timestamp': current_time
})
self.bot.privmsg(
self.service_name,
f"KICK {channel} {nick} :Banned for repeated spamming"
)
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 the user and log action
self.bot.privmsg(
self.service_name,
f"KICK {channel} {nick} :Stop spamming."
)
user_kicks.append(current_time)
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):
"""Clean up inactive user records every minute."""
cutoff = time.time() - 300
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.")
@cron('* * * * *')
def remove_expired_bans(self):
"""Remove bans that are older than 5 minutes."""
current_time = time.time()
expired_bans = [ban for ban in self.active_bans if current_time - ban['timestamp'] >= 300]
for ban in expired_bans:
self.bot.send(f"MODE {ban['channel']} -b {ban['ban_mask']}")
self.bot.log.info(f"Auto-removed ban {ban['ban_mask']} from {ban['channel']} after 5 minutes.")
# Update active_bans to exclude expired ones
self.active_bans = [ban for ban in self.active_bans if current_time - ban['timestamp'] < 300]
def connection_made(self):
"""Initialize when bot connects."""
self.bot.log.info("AntiSpam plugin loaded with automated kick-to-ban escalation and auto-unban.")

View File

@ -1,310 +0,0 @@
import irc3
from irc3.plugins.command import command
from tinydb import Query, TinyDB
import fnmatch
from ircstyle import style
from datetime import datetime
def check_ignore(func):
"""Decorator to block processing for ignored users in event handlers.
Args:
func (callable): The function to decorate
Returns:
callable: Wrapped function with ignore checks
The decorator performs these actions in order:
1. Converts user mask to hostmask string
2. Checks against all ignore entries in permission DB
3. Uses fnmatch for wildcard pattern matching
4. Blocks processing if any match found
"""
async def wrapper(self, mask, *args, **kwargs):
"""Execute wrapped function only if user is not ignored."""
hostmask = str(mask)
User = Query()
ignored_entries = self.bot.permission_db.search(
User.permission == 'ignore'
)
for entry in ignored_entries:
if fnmatch.fnmatch(hostmask, entry['mask']):
self.bot.log.debug(f"Blocking processing for {hostmask}")
return True # Block further processing
return await func(self, mask, *args, **kwargs)
# Preserve the original function's attributes for irc3
wrapper.__name__ = func.__name__
wrapper.__module__ = func.__module__
wrapper.__doc__ = func.__doc__
return wrapper
@irc3.plugin
class TinyDBPermissions:
"""Main permission system plugin handling storage and commands."""
def __init__(self, bot):
self.bot = bot
self.permission_db = TinyDB('permissions.json')
self.bot.permission_db = self.permission_db
self.User = Query()
self.bot.log.info("TinyDB permissions plugin initialized")
@command(permission='admin')
async def perm(self, mask, target, args):
"""Manage permissions through command interface.
Usage:
%%perm --add <mask> <permission>
%%perm --del <mask> <permission>
%%perm --list [<mask>]
"""
if args['--add']:
await self._add_permission(target, args['<mask>'], args['<permission>'])
elif args['--del']:
await self._del_permission(target, args['<mask>'], args['<permission>'])
elif args['--list']:
self._list_permissions(target, args['<mask>'])
else:
error_msg = style(
"Invalid syntax. Use --add, --del, or --list.",
fg='red', bold=True
)
self.bot.privmsg(target, error_msg)
@command(permission='admin')
async def ignore(self, mask, target, args):
"""Manage user ignore list.
Usage:
%%ignore --add <nick_or_mask>
%%ignore --del <nick_or_mask>
%%ignore --list
"""
if args['--list']:
ignored = self.permission_db.search(
self.User.permission == 'ignore'
)
if not ignored:
msg = style("No ignored users found", fg='yellow', bold=True)
self.bot.privmsg(target, msg)
return
self.bot.privmsg(target, style("Ignored users:", fg='blue', bold=True))
for entry in ignored:
msg = style(
f"{entry['mask']} (since {entry.get('timestamp', 'unknown')}",
fg='cyan'
)
self.bot.privmsg(target, msg)
return
mask_or_nick = args['<nick_or_mask>']
if '!' in mask_or_nick or '@' in mask_or_nick:
user_mask = mask_or_nick
else:
try:
user_mask = await self._get_hostmask(mask_or_nick)
except ValueError as e:
# Fallback to nick!*@* pattern if WHOIS fails
user_mask = f"{mask_or_nick}!*@*"
warning_msg = style(
f"Using fallback hostmask {user_mask} ({str(e)})",
fg='yellow', bold=True
)
self.bot.privmsg(target, warning_msg)
if args['--add']:
if self.permission_db.contains(
(self.User.mask == user_mask) &
(self.User.permission == 'ignore')
):
msg = style(f"{user_mask} already ignored", fg='yellow', bold=True)
else:
self.permission_db.insert({
'mask': user_mask,
'permission': 'ignore',
'timestamp': datetime.now().isoformat()
})
msg = style(f"Ignored {user_mask}", fg='green', bold=True)
elif args['--del']:
removed = self.permission_db.remove(
(self.User.mask == user_mask) &
(self.User.permission == 'ignore')
)
msg = style(f"Unignored {user_mask} ({len(removed)} entries)", fg='green', bold=True)
else:
msg = style("Invalid syntax", fg='red', bold=True)
self.bot.privmsg(target, msg)
async def _get_hostmask(self, nick):
"""Get the hostmask for a given nickname using WHOIS."""
try:
whois_info = await self.bot.async_cmds.whois(nick)
# Validate required fields exist
required_fields = {'username', 'host', 'success'}
missing = required_fields - whois_info.keys()
if missing:
raise ValueError(f"WHOIS response missing fields: {', '.join(missing)}")
if not whois_info['success']:
raise ValueError(f"WHOIS failed for {nick} (server error)")
except Exception as e:
raise ValueError(f"WHOIS error: {str(e)}") from e
# Use .get() with fallback for optional fields
username = whois_info.get('username', 'unknown')
host = whois_info.get('host', 'unknown.host')
return f"{nick}!{username}@{host}"
async def _add_permission(self, target, user_mask, perm):
"""Add a permission to the database."""
original_mask = user_mask
try:
# Check if user_mask is a nickname (no ! or @)
if '!' not in user_mask and '@' not in user_mask:
user_mask = await self._get_hostmask(user_mask)
except ValueError as e:
error_msg = style(
f"Failed to get hostmask for {original_mask}: {e}",
fg='red', bold=True
)
self.bot.privmsg(target, error_msg)
return
existing = self.permission_db.search(
(self.User.mask == user_mask) &
(self.User.permission == perm)
)
if existing:
msg = style(
f"Permission '{perm}' already exists for {user_mask}",
fg='yellow', bold=True
)
else:
self.permission_db.insert({'mask': user_mask, 'permission': perm})
msg = style(
f"Added permission '{perm}' for {user_mask}",
fg='green', bold=True
)
self.bot.privmsg(target, msg)
async def _del_permission(self, target, user_mask, perm):
"""Remove a permission from the database."""
original_mask = user_mask
try:
# Check if user_mask is a nickname (no ! or @)
if '!' not in user_mask and '@' not in user_mask:
user_mask = await self._get_hostmask(user_mask)
except ValueError as e:
error_msg = style(
f"Failed to get hostmask for {original_mask}: {e}",
fg='red', bold=True
)
self.bot.privmsg(target, error_msg)
return
removed = self.permission_db.remove(
(self.User.mask == user_mask) &
(self.User.permission == perm)
)
if removed:
msg = style(
f"Removed {len(removed)} '{perm}' permission(s) for {user_mask}",
fg='green', bold=True
)
else:
msg = style(
f"No '{perm}' permissions found for {user_mask}",
fg='red', bold=True
)
self.bot.privmsg(target, msg)
def _list_permissions(self, target, mask_filter):
"""List permissions matching a filter pattern."""
mask_filter = mask_filter or '*'
regex = fnmatch.translate(mask_filter).split('(?ms)')[0].rstrip('\\Z')
entries = self.permission_db.search(
self.User.mask.matches(regex)
)
if not entries:
msg = style("No permissions found", fg='red', bold=True)
self.bot.privmsg(target, msg)
return
for entry in entries:
msg = style(
f"{entry['mask']}: {entry['permission']}",
fg='blue', bold=True
)
self.bot.privmsg(target, msg)
# @irc3.event(irc3.rfc.PRIVMSG)
# @check_ignore
# async def on_privmsg(self, mask, event, target, data):
# """Handle PRIVMSG events with integrated ignore checks.
# Args:
# mask: User's hostmask
# event: IRC event type
# target: Message target (channel or user)
# data: Message content
# Returns:
# bool: True to block processing, False to continue
# """
# return True
# Continue processing if not ignored
#return False
class TinyDBPolicy:
"""Authorization system for command access control."""
def __init__(self, bot):
self.bot = bot
self.User = Query()
def has_permission(self, client_mask, permission):
"""Check if a client has required permissions."""
# Check ignore list first
ignored = self.bot.permission_db.search(
self.User.permission == 'ignore'
)
for entry in ignored:
if fnmatch.fnmatch(client_mask, entry['mask']):
return False
# Check permissions if not ignored
if permission is None:
return True
# Check for matching permissions using fnmatch
entries = self.bot.permission_db.search(
self.User.permission.test(lambda p: p in (permission, 'all_permissions'))
)
for entry in entries:
if fnmatch.fnmatch(client_mask, entry['mask']):
return True
return False
def __call__(self, predicates, meth, client, target, args):
"""Enforce command permissions."""
cmd_name = predicates.get('name', meth.__name__)
client_hostmask = str(client)
if self.has_permission(client_hostmask, predicates.get('permission')):
return meth(client, target, args)
error_msg = style(
f"Access denied for '{cmd_name}' command",
fg='red', bold=True
)
self.bot.privmsg(client.nick, error_msg)

View File

@ -1,120 +0,0 @@
# -*- 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 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>')
message = ' '.join(args['<message>'])
if not channel or not message:
self.bot.privmsg(target, "Usage: !say <channel> <message>")
return
styled_message = self.style_message(message)
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',
'\u030C', '\u030D', '\u030E', '\u030F', '\u0310', '\u0311',
'\u0312', '\u0313', '\u0314', '\u0315', '\u0316', '\u0317',
'\u0318', '\u0319', '\u031A', '\u031B', '\u031C', '\u031D',
'\u031E', '\u031F', '\u0320', '\u0321', '\u0322', '\u0323',
'\u0324', '\u0325', '\u0326', '\u0327', '\u0328', '\u0329',
'\u032A', '\u032B', '\u032C', '\u032D', '\u032E', '\u032F',
'\u0330', '\u0331', '\u0332', '\u0333', '\u0334', '\u0335',
'\u0336', '\u0337', '\u0338', '\u0339', '\u033A', '\u033B',
'\u033C', '\u033D', '\u033E', '\u033F', '\u0340', '\u0341',
'\u0342', '\u0343', '\u0344', '\u0345', '\u0346', '\u0347',
'\u0348', '\u0349', '\u034A', '\u034B', '\u034C', '\u034D',
'\u034E', '\u034F', '\u0350', '\u0351', '\u0352', '\u0353',
'\u0354', '\u0355', '\u0356', '\u0357', '\u0358', '\u0359',
'\u035A', '\u035B', '\u035C', '\u035D', '\u035E', '\u035F',
'\u0360', '\u0361', '\u0362'
]
glitched_char = char
for _ in range(random.randint(1, 1)):
color = random.randint(0, 15)
style = random.choice(['\x02', '\x1F']) # Bold and Underline
combining_char = random.choice(combining_chars)
glitched_char += f'\x03{color}{style}{combining_char}\x0F'
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 = ''
for char in message:
char_with_combining = self.add_combining_characters(char)
styled_message += f'{white_color_code}{char_with_combining}'
return styled_message.strip() # Remove the trailing space

View File

@ -1,263 +0,0 @@
# -*- 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)

View File

@ -1,451 +0,0 @@
# -*- coding: utf-8 -*-
"""
IRC Bot Plugin for Uploading Files to hardfiles.org
This plugin allows users to upload files to hardfiles.org using yt-dlp for downloads.
It supports downloading files from various sources (YouTube, Instagram, etc.) and can
optionally convert videos to MP3 format before uploading. Files larger than 500MB are rejected.
Usage:
!upload [--mp3] <url_or_search_term>...
Example: !upload never gonna give you up
Dependencies:
- aiohttp
- aiofiles
- irc3
- yt-dlp
- ircstyle
Author: Zodiac
Version: 1.3
Date: 2025-02-12
"""
import aiohttp
import aiofiles
import irc3
import tempfile
import os
import re
import asyncio
from irc3.plugins.command import command
import ircstyle
import yt_dlp
from yt_dlp.utils import DownloadError
from urllib.parse import urlparse
import googleapiclient.discovery
from irc3.compat import Queue
# Global headers to mimic a real browser (ban evasion)
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Referer": "https://www.google.com/",
"Connection": "keep-alive",
"DNT": "1",
"Upgrade-Insecure-Requests": "1"
}
# Constants for YouTube API
API_SERVICE_NAME = "youtube"
API_VERSION = "v3"
DEVELOPER_KEY = "AIzaSyBNrqOA0ZIziUVLYm0K5W76n9ndqz6zTxI"
# Initialize YouTube API client
youtube = googleapiclient.discovery.build(
API_SERVICE_NAME, API_VERSION, developerKey=DEVELOPER_KEY
)
@irc3.plugin
class UploadPlugin:
"""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
self.queue = Queue() # Initialize the queue
self.bot.loop.create_task(self.process_queue()) # Start queue processing
async def process_queue(self):
"""Process upload tasks from the queue."""
while True:
task = await self.queue.get()
url, target, mp3 = task
try:
await self.do_upload(url, target, mp3)
except Exception as e:
exc_msg = self._ensure_str(e)
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"Upload failed: {exc_msg}", fg="red", bold=True)
)
finally:
self.queue.task_done()
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')
if value is None:
return ''
return str(value)
async def search_youtube(self, query):
"""
Search YouTube for the given query and return the URL of the first result.
Args:
query (str): The search query.
Returns:
str: The URL of the first search result.
"""
try:
request = youtube.search().list(
part="id",
q=query,
type="video",
maxResults=1
)
result = await self.bot.loop.run_in_executor(None, request.execute)
if result.get("items"):
video_id = result["items"][0]["id"]["videoId"]
return f"https://www.youtube.com/watch?v={video_id}"
except Exception as e:
self.bot.log.error(f"YouTube search error: {e}")
return None
@command(aliases=['u'], public=True)
async def upload(self, mask, target, args):
"""
Upload a file to hardfiles.org (Max 500MB).
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): Parsed command arguments.
Usage:
%%upload [--mp3] <url_or_search_term>...
Example: !upload never gonna give you up
"""
url_or_search_term = args.get('<url_or_search_term>')
if isinstance(url_or_search_term, list):
url_or_search_term = " ".join(url_or_search_term)
mp3 = args.get('--mp3')
if not url_or_search_term:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style("Usage: !upload [--mp3] <url_or_search_term>...", fg="red", bold=True)
)
return
if not re.match(r'^https?://', url_or_search_term):
self.bot.privmsg(
target,
ircstyle.style("🔍 ", fg="blue") +
ircstyle.style("Searching ", fg="blue") +
ircstyle.style("YouTube", fg="white", bg="red") +
ircstyle.style(" for: ", fg="blue") +
ircstyle.style(url_or_search_term, fg="green", underline=True, italics=True)
)
url = await self.search_youtube(url_or_search_term)
if not url:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style("No results found on YouTube.", fg="red", bold=True)
)
return
else:
url = url_or_search_term
# Enqueue the upload task
queue_size = self.queue.qsize() + 1
self.queue.put_nowait((url, target, mp3))
self.bot.privmsg(
target,
ircstyle.style("", fg="yellow") +
ircstyle.style("Upload queued ", fg="blue", bold=False, italics=True) +
ircstyle.style(f"(Position #{queue_size})",italics=True, fg="grey") +
ircstyle.style(" Processing soon...", fg="blue", italics=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 = 500 * 1024 * 1024 # 500MB 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",
"reddit.com",
"twitter.com",
"tiktok.com",
"facebook.com",
"dailymotion.com",
)
should_check_headers = not any(domain.endswith(d) for d in skip_check_domains)
if should_check_headers:
try:
async with aiohttp.ClientSession(headers=HEADERS) as session:
async with session.head(url) as response:
if response.status != 200:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"Failed to fetch headers: HTTP {response.status}", fg="red", bold=True)
)
return
content_length = response.headers.get('Content-Length')
if content_length and int(content_length) > max_size:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"File size ({int(content_length) // (1024 * 1024)}MB) exceeds 500MB limit", fg="red", bold=True)
)
return
except Exception as e:
err_msg = self._ensure_str(e)
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"Error during header check: {err_msg}", fg="red", bold=True)
)
return
ydl_opts = {
'outtmpl': os.path.join(tmp_dir, '%(title)s.%(ext)s'),
'format': 'bestaudio/best' if mp3 else 'best[ext=mp4]/best',
'restrictfilenames': True,
'noplaylist': True,
'quiet': True,
'concurrent_fragment_downloads': 5,
'postprocessors': [
{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '192',
}
] if mp3 else [],
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
try:
info = await asyncio.to_thread(ydl.extract_info, url, download=False)
except DownloadError as e:
err_msg = self._ensure_str(e)
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"Info extraction failed: {err_msg}", fg="red", bold=True)
)
return
except UnicodeDecodeError:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style("Error: Received non-UTF-8 output during info extraction", fg="red", bold=True)
)
return
estimated_size = info.get('filesize') or info.get('filesize_approx')
if estimated_size and estimated_size > max_size:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"File size ({estimated_size // (1024 * 1024)}MB) exceeds 500MB limit", fg="red", bold=True)
)
return
try:
info = await asyncio.to_thread(ydl.extract_info, url, download=True)
except DownloadError as e:
err_msg = self._ensure_str(e)
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"Download failed: {err_msg}", fg="red", bold=True)
)
return
except UnicodeDecodeError:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style("Error: Received non-UTF-8 output during download", fg="red", bold=True)
)
return
# Build metadata for display.
metadata_parts = []
title = self._ensure_str(info.get("title"))
uploader = self._ensure_str(info.get("uploader"))
duration = info.get("duration")
upload_date = self._ensure_str(info.get("upload_date"))
view_count = info.get("view_count")
description = self._ensure_str(info.get("description"))
description = re.sub(r'\s+', ' ', description) # Strip multiple spaces to single space
if title:
metadata_parts.append(
ircstyle.style("📝 Title:", fg="yellow", bold=True) + " " +
ircstyle.style(title, fg="yellow", bold=False, underline=True, italics=True)
)
if uploader:
metadata_parts.append(
ircstyle.style("👤 Uploader:", fg="purple", bold=True) + " " +
ircstyle.style(uploader, fg="purple", bold=False)
)
if duration:
metadata_parts.append(
ircstyle.style("⏱ Duration:", fg="green", bold=True) + " " +
ircstyle.style(self._format_duration(duration), fg="green", bold=False)
)
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("Upload Date:", fg="silver", bold=True) + " " +
ircstyle.style(formatted_date, fg="silver", bold=False, italics=True)
)
if view_count is not None:
metadata_parts.append(
ircstyle.style("Views:", fg="teal", bold=True) + " " +
ircstyle.style(str(view_count), fg="teal", bold=False)
)
if description:
if len(description) > 200:
description = description[:200] + "..."
metadata_parts.append(
ircstyle.style("Description:", fg="silver", bold=True) + " " +
ircstyle.style(description, fg="silver", bold=False, italics=True)
)
if metadata_parts:
# Send metadata header in teal.
# self.bot.privmsg(target, ircstyle.style("📁 Metadata:", fg="teal", bold=True))
self.bot.privmsg(target, "".join(metadata_parts))
downloaded_files = info.get('requested_downloads', [])
if not downloaded_files:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style("No files downloaded", fg="red", bold=True)
)
return
first_file = downloaded_files[0]
downloaded_file = first_file.get('filepath', first_file.get('filename'))
if not downloaded_file or not os.path.exists(downloaded_file):
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"Downloaded file not found: {downloaded_file}", fg="red", bold=True)
)
return
file_size = os.path.getsize(downloaded_file)
if file_size > max_size:
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"File size ({file_size // (1024 * 1024)}MB) exceeds 500MB limit", fg="red", bold=True)
)
return
try:
async with aiohttp.ClientSession(headers=HEADERS) as session:
form = aiohttp.FormData()
async with aiofiles.open(downloaded_file, 'rb') as f:
file_content = await f.read()
form.add_field(
'file',
file_content,
filename=os.path.basename(downloaded_file),
content_type='application/octet-stream'
)
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("", fg="red") +
ircstyle.style(f"Upload failed: HTTP {resp.status}", fg="red", bold=True)
)
return
raw_response = await resp.read()
response_text = raw_response.decode('utf-8', errors='replace')
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) +
ircstyle.style(upload_url, fg="blue", underline=True)
)
self.bot.privmsg(target, response_msg)
except Exception as e:
err_msg = self._ensure_str(e)
self.bot.privmsg(
target,
ircstyle.style("", fg="red") +
ircstyle.style(f"Error during file upload: {err_msg}", fg="red", bold=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
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"

View File

@ -1,128 +0,0 @@
# -*- coding: utf-8 -*-
"""
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:
"""
A plugin to search Urban Dictionary for terms and post results to IRC.
This plugin uses a queue to manage asynchronous search requests and ensures
proper resource cleanup when unloaded.
"""
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
self.bot.loop.create_task(self._process_queue()) # Start queue processor
@command(permission='view', aliases=['ud'], public=True)
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>'])
if not term:
return "Please provide a term to search."
# Enqueue the search request
self.queue.put_nowait((term, target))
return f"\x02Searching Urban Dictionary\x02 for '\x034{term}\x03'..."
async def _process_queue(self):
"""
Process search requests from the queue asynchronously.
This method runs indefinitely, processing one request at a time.
"""
while True:
try:
# Get the next search request
term, target = await self.queue.get()
# Ensure the session is initialized
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession(loop=self.bot.loop)
# Fetch data from Urban Dictionary API
url = f"https://api.urbandictionary.com/v0/define?term={term}"
async with self.session.get(url) as response:
if response.status != 200:
self.bot.privmsg(target, f"\x02Error:\x02 Failed to fetch definition for '\x034{term}\x03'.")
continue
data = await response.json()
if not data.get('list'):
self.bot.privmsg(target, f"\x02No definition found\x02 for '\x034{term}\x03'.")
continue
# Extract the first result
result = data['list'][0]
definition = result['definition'].replace('\n', ' ').strip()
example = result['example'].replace('\n', ' ').strip()
permalink = result['permalink']
# Format the response with colors and bold
response_message = (
f"\x02[{term}]\x02 \x033{definition}\x03 | "
f"Example: \x0312{example}\x03 | "
f"More: \x032{permalink}\x03"
)
# Send the response to the IRC channel
self.bot.privmsg(target, response_message)
except Exception as e:
self.bot.privmsg(target, f"\x02Error:\x02 An unexpected error occurred: {str(e)}")
finally:
# Mark the task as done
self.queue.task_done()
def __unload__(self):
"""
Cleanup when the plugin is unloaded.
Ensures that the aiohttp session is properly closed to avoid resource leaks.
"""
if self.session and not self.session.closed:
self.bot.loop.create_task(self.session.close())

View File

@ -1,129 +0,0 @@
# -*- coding: utf-8 -*-
"""
IRC3 Bot Plugin: URL Title Fetcher
A plugin for IRC3 bots that monitors chat messages for URLs, fetches their webpage titles, and displays them
with formatted styling in the chat. Provides visual enhancement to URL sharing in IRC channels.
Features:
- Asynchronous URL processing using aiohttp for efficient network operations
- Robust HTML parsing with lxml for accurate title extraction
- Configurable message styling with color and formatting options
- Built-in exclusion of YouTube URLs to avoid conflicts with dedicated YouTube plugins
- Error handling for network and parsing operations
- Proper resource cleanup through session management
- Queue-based processing system with strict rate limiting
Dependencies:
- aiohttp: For asynchronous HTTP requests
- irc3: Core IRC bot functionality
- ircstyle: IRC text formatting utilities
- lxml: HTML parsing capabilities
Author: Zodiac
Date: 2025-02-14
"""
import re
import time
import aiohttp
import ircstyle
from lxml import html
import irc3
from irc3 import event
from irc3.compat import Queue
from plugins.services.permissions import check_ignore
@irc3.plugin
class URLTitlePlugin:
"""Plugin for fetching and displaying webpage titles from URLs shared in IRC messages.
Monitors IRC messages for URLs, retrieves their webpage titles, and posts formatted responses
back to the channel. Supports styled text output with configurable formatting options.
Attributes:
bot (irc3.IrcBot): Reference to the main IRC bot instance
session (aiohttp.ClientSession): Persistent HTTP session for making web requests
url_pattern (re.Pattern): Compiled regex for URL detection in messages
queue (Queue): Processing queue for URL handling tasks
last_processed (float): Timestamp of last successful URL processing
"""
def __init__(self, bot):
"""Initialize plugin with bot instance and set up components."""
self.bot = bot
self.session = aiohttp.ClientSession(loop=self.bot.loop)
self.url_pattern = re.compile(r"https?://[^\s<>\"']+|www\.[^\s<>\"']+")
self.queue = Queue()
self.last_processed = 0 # Initialize to epoch start
self.bot.create_task(self.process_queue())
@event(irc3.rfc.PRIVMSG)
@check_ignore
async def on_privmsg(self, mask, event, target, data):
"""Handle incoming messages and enqueue URLs for processing."""
urls = self.url_pattern.findall(data)
for url in urls:
if "youtube.com" in url.lower() or "youtu.be" in url.lower():
continue
self.queue.put_nowait((target, url))
async def process_queue(self):
"""Process URLs from the queue with strict 5-second cooldown between requests."""
while True:
target, url = await self.queue.get()
try:
current_time = time.time()
elapsed = current_time - self.last_processed
if elapsed < 5:
self.bot.log.info(f"Rate limited: Waiting {5 - elapsed:.1f}s to process {url}")
continue
title = await self.fetch_title(url)
if title:
formatted_message = self.format_message(title, url)
await self.bot.privmsg(target, formatted_message)
self.last_processed = time.time() # Update after successful processing
except Exception as e:
self.bot.log.error(f"Error processing URL {url}: {e}")
finally:
self.queue.task_done()
def format_message(self, title, url):
"""Create a styled IRC message containing the webpage title and source URL."""
prefix = ircstyle.style("", fg="cyan", bold=True, reset=True)
title_label = ircstyle.style("Title", fg="blue", bold=True, reset=True)
title_text = ircstyle.style(title, fg="green", italics=True, underline=True, reset=True)
separator = ircstyle.style("", fg="grey", bold=True, reset=True)
url_label = ircstyle.style("Source", fg="blue", bold=True, underline=True, reset=True)
url_text = ircstyle.style(url, fg="cyan", italics=True, reset=True)
suffix = ircstyle.style("", fg="cyan", bold=True, reset=True)
return f"{prefix} {title_label}: {title_text} {separator} {url_label}: {url_text} {suffix}"
async def fetch_title(self, url):
"""Retrieve the title of a webpage using asynchronous HTTP requests."""
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
async with self.session.get(url, headers=headers, timeout=10) as response:
response.raise_for_status()
content = await response.text()
tree = html.fromstring(content)
title = tree.findtext(".//title")
return title.strip() if title else "No title found"
async def close(self):
"""Clean up resources by closing the HTTP session."""
await self.session.close()
def __del__(self):
"""Ensure proper cleanup when the plugin is destroyed."""
self.bot.create_task(self.close())

View File

@ -1,203 +0,0 @@
from __future__ import annotations
import logging
from irc3 import plugin, utils, rfc
from irc3.dec import event
from irc3.utils import IrcString
from collections import defaultdict
from typing import Optional, Set, Dict, Any, cast
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class Channel(set):
"""Represents an IRC channel with member management and mode tracking."""
def __init__(self) -> None:
super().__init__()
self.modes: Dict[str, Set[str]] = defaultdict(set)
self.topic: Optional[str] = None
def add_member(self, nick: str, modes: str = '') -> None:
"""Add member with optional modes."""
super().add(nick)
self._update_modes(nick, modes, add=True)
logger.debug(f"Added member {nick} with modes {modes} to channel")
def remove_member(self, nick: str) -> None:
"""Remove member and all associated modes."""
super().discard(nick)
self._update_modes(nick, remove=True)
logger.debug(f"Removed member {nick} from channel")
def _update_modes(self, nick: str, modes: str = '',
add: bool = False, remove: bool = False) -> None:
"""Update mode tracking for a member."""
for mode in modes:
if add:
self.modes[mode].add(nick)
elif remove:
self.modes[mode].discard(nick)
logger.debug(f"Updated modes for {nick}: {'Added' if add else 'Removed'} {modes}")
def __repr__(self) -> str:
return f"Channel({sorted(self)})"
@plugin
class Userlist:
"""Enhanced user tracking with proper hostmask handling and mode management."""
def __init__(self, context: Any) -> None:
self.context = context
self.connection_lost()
def connection_lost(self, client: Any = None) -> None:
"""Reset state on connection loss."""
self.channels: Dict[str, Channel] = defaultdict(Channel)
self.nicks: Dict[str, IrcString] = {}
self.context.channels = self.channels
self.context.nicks = self.nicks
logger.info("Connection lost, state reset.")
@event(rfc.JOIN_PART_QUIT)
def on_join_part_quit(self, mask: Optional[IrcString] = None,
event: Optional[str] = None, **kwargs: Any) -> None:
"""Handle join/part/quit events."""
if mask and event:
getattr(self, event.lower())(mask.nick, mask, **kwargs)
logger.debug(f"Event {event} for {mask.nick}")
@event(rfc.KICK)
def on_kick(self, mask: Optional[IrcString] = None, target: Optional[IrcString] = None,
**kwargs: Any) -> None:
"""Handle kick events."""
if target:
self.part(target.nick, mask, **kwargs) # Note: Added mask here for consistency
logger.debug(f"User {target.nick} was kicked from a channel")
def join(self, nick: str, mask: IrcString, channel: str, **kwargs: Any) -> None:
"""Process user join."""
channel_obj = self.channels[channel]
if nick != self.context.nick:
channel_obj.add_member(mask.nick)
self.nicks[mask.nick] = mask
self._broadcast(channel_obj, **kwargs)
logger.debug(f"User {nick} joined channel {channel}")
def part(self, nick: str, mask: IrcString, channel: str, **kwargs: Any) -> None:
"""Process user part."""
if nick == self.context.nick:
self.channels.pop(channel, None)
logger.debug(f"Left channel {channel}")
else:
channel_obj = self.channels.get(channel)
if channel_obj:
self._broadcast(channel_obj, **kwargs)
channel_obj.remove_member(nick)
self._cleanup_nick(nick)
logger.debug(f"User {nick} parted from channel {channel}")
def quit(self, nick: str, mask: IrcString, **kwargs: Any) -> None:
"""Process user quit."""
if nick == self.context.nick:
self.connection_lost()
logger.info(f"User {nick} quit, connection lost")
else:
affected = set()
for channel in self.channels.values():
if nick in channel:
channel.remove_member(nick)
affected.update(channel)
self._broadcast(affected, **kwargs)
self._cleanup_nick(nick)
logger.debug(f"User {nick} quit from channels: {affected}")
@event(rfc.NEW_NICK)
def new_nick(self, nick: IrcString, new_nick: str, **kwargs: Any) -> None:
"""Handle nickname changes."""
host = nick.host or ''
self.nicks[new_nick] = IrcString(f"{new_nick}!{host}")
old_nick = nick.nick
self._update_nick_in_channels(old_nick, new_nick)
self.nicks.pop(old_nick, None)
self._broadcast({new_nick}, **kwargs)
logger.debug(f"Nickname change: {old_nick} -> {new_nick}")
def _update_nick_in_channels(self, old_nick: str, new_nick: str) -> None:
"""Update nickname across all channels."""
for channel in self.channels.values():
if old_nick in channel:
channel.remove_member(old_nick)
channel.add_member(new_nick)
for mode_set in channel.modes.values():
if old_nick in mode_set:
mode_set.add(new_nick)
mode_set.discard(old_nick)
logger.debug(f"Updated nickname {old_nick} to {new_nick} in channels")
@event(rfc.RPL_NAMREPLY)
def names(self, channel: str, data: str, **kwargs: Any) -> None:
"""Process NAMES reply with proper mode handling."""
statusmsg = self.context.server_config.get('STATUSMSG', '@+')
prefix_config = self.context.server_config.get('PREFIX', '(ov)@+')
mode_chars, prefixes = self._parse_prefix_config(prefix_config)
prefix_map = dict(zip(prefixes, mode_chars))
channel_obj = self.channels[channel]
for item in data.split():
nick = item.lstrip(statusmsg)
modes = ''.join([prefix_map[p] for p in item[:len(item)-len(nick)]
if p in prefix_map])
channel_obj.add_member(nick, modes)
logger.debug(f"Processed NAMES reply for channel {channel}")
def _parse_prefix_config(self, config: str) -> tuple[str, str]:
"""Parse PREFIX config into mode characters and symbols."""
parts = config.strip('()').split(')')
return (parts[0], parts[1]) if len(parts) == 2 else ('ov', '@+')
@event(rfc.RPL_WHOREPLY)
def who(self, channel: str, nick: str, username: str,
server: str, **kwargs: Any) -> None:
"""Process WHO reply with hostmask."""
mask = IrcString(f"{nick}!{username}@{server}")
self.channels[channel].add_member(nick)
self.nicks[nick] = mask
logger.debug(f"Processed WHO reply for {nick} in {channel}")
@event(rfc.MODE)
def mode(self, target: str, modes: str, data: Any, **kwargs: Any) -> None:
"""Handle mode changes with proper parameter handling."""
chantypes = self.context.server_config.get('CHANTYPES', '#&')
if not target or target[0] not in chantypes:
return
prefix_config = self.context.server_config.get('PREFIX', '(ov)@+')
mode_chars, prefixes = self._parse_prefix_config(prefix_config)
param_modes = self.context.server_config.get('CHANMODES', '').split(',')[0]
parsed = utils.parse_modes(modes, data.split() if isinstance(data, str) else data)
channel = self.channels[target]
for sign, mode, param in parsed:
if mode in param_modes and param:
# Handle parameterized modes (e.g., +b, +k)
pass
elif mode in mode_chars:
prefix = dict(zip(mode_chars, prefixes))
mode_key = prefix.get(mode, '')
if sign == '+':
channel.modes[mode_key].add(param)
else:
channel.modes[mode_key].discard(param)
logger.debug(f"Mode change in {target}: {modes}")
def _cleanup_nick(self, nick: str) -> None:
"""Remove nick if not in any channels."""
if not any(nick in channel for channel in self.channels.values()):
self.nicks.pop(nick, None)
logger.debug(f"Cleaned up nick {nick} as no longer in any channels")
def _broadcast(self, targets: Any, **kwargs: Any) -> None:
"""Placeholder for broadcast functionality."""
logger.debug("Broadcasting to targets: %s", targets)

View File

@ -1,56 +0,0 @@
import irc3
import re
from irc3 import dec
@irc3.plugin
class AsyncNotifyOnJoin:
def __init__(self, bot):
self.bot = bot
# Ensure async_cmds is available
if not hasattr(self.bot, 'async_cmds'):
self.bot.log.error("async_cmds plugin not loaded. Please add 'irc3.plugins.asynchronious' to your config.")
raise RuntimeError("Missing async_cmds plugin")
@dec.extend
async def notify_users(self, channel, joiner):
"""Async method to get channel users and send notifications"""
try:
# Ensure channel is in the bot's channel list
if channel not in self.bot.channels:
self.bot.log.warning(f"Bot is not in channel: {channel}")
return
# Debug: Log the async_cmds object
self.bot.log.debug(f"async_cmds: {self.bot.async_cmds}")
# Debug: Log the names method
if hasattr(self.bot.async_cmds, 'names'):
self.bot.log.debug("async_cmds.names() is available")
else:
self.bot.log.error("async_cmds.names() is missing!")
# Get fresh user list through IRC protocol
result = await self.bot.async_cmds.names(channel)
self.bot.log.debug(f"Result from async_cmds.names(): {result}")
if result and result.get('success'):
for nickname in result['names']:
# Clean nickname from mode prefixes
clean_nick = re.sub(r'^[@~+&%]', '', nickname)
if clean_nick != self.bot.nick:
self.bot.privmsg(
clean_nick,
f"{joiner} has joined {channel}. Welcome them!"
)
else:
self.bot.log.error(f"Failed to fetch user list for {channel}")
except Exception as e:
self.bot.log.error(f"Notification failed: {e}")
@irc3.event(irc3.rfc.JOIN)
def on_join(self, channel, **kwargs):
"""Trigger async notification process on join"""
joiner = kwargs['mask'].nick
if joiner != self.bot.nick:
# Create async task to handle notifications
self.bot.loop.create_task(self.notify_users(channel, joiner))