kami
/
OokamiPupV2
1
0
Fork 0
Code Issues 4 Pull Requests Packages Projects 1 Releases Wiki Activity

Improved platform logging and Custom VC 

- Platform logs associated to a message are now associated with platform-specific IDs
- Fixes for `!customvc`, specifically in relation to commands execution.
  - Optimized logic
  - Fixed limits for certain values like VC users limit
  - Better code integrity: now self-references as a Cog
- Lots of minor other tweaks, adjustments, and improvements
kami_dev
Kami 2025-03-06 17:53:54 +01:00
parent 86ac83f34c
commit cce0f21ab0
14 changed files with 988 additions and 803 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

215
bot_discord.py
View File

@ -5,6 +5,7 @@ from discord.ext import commands, tasks
import importlib
import cmd_discord
import json
import os
import globals
@ -23,7 +24,6 @@ class DiscordBot(commands.Bot):
self.log = globals.log # Use the logging function from bots.py
self.db_conn = None # We'll set this later
self.help_data = None # We'll set this later
self.load_commands()
globals.log("Discord bot initiated")
@ -45,46 +45,39 @@ class DiscordBot(commands.Bot):
"""
self.db_conn = db_conn
def load_commands(self):
"""
Load all commands from cmd_discord.py
"""
try:
importlib.reload(cmd_discord) # Reload the commands file
cmd_discord.setup(self) # Ensure commands are registered
globals.log("Discord commands loaded successfully.")
async def setup_hook(self):
# This is an async-ready function you can override in discord.py 2.0.
for filename in os.listdir("cmd_discord"):
if filename.endswith(".py") and filename != "__init__.py":
cog_name = f"cmd_discord.{filename[:-3]}"
await self.load_extension(cog_name) # now we can await it
# Load help info
help_json_path = "dictionary/help_discord.json"
modules.utility.initialize_help_data(
bot=self,
help_json_path=help_json_path,
is_discord=True
)
except Exception as e:
_result = f"Error loading Discord commands: {e}"
globals.log(_result, "ERROR", True)
# Log which cogs got loaded
short_name = filename[:-3]
globals.log(f"Loaded Discord command cog '{short_name}'", "DEBUG")
@commands.command(name="cmd_reload")
globals.log("All Discord command cogs loaded successfully.", "INFO")
# Now that cogs are all loaded, run any help file initialization:
help_json_path = "dictionary/help_discord.json"
modules.utility.initialize_help_data(
bot=self,
help_json_path=help_json_path,
is_discord=True
)
@commands.command(name="reload")
@commands.is_owner()
async def cmd_reload(self, ctx: commands.Context):
async def reload(ctx, cog_name: str):
"""Reloads a specific cog without restarting the bot."""
try:
importlib.reload(cmd_discord) # Reload the commands file
cmd_discord.setup(self) # Ensure commands are registered
# Load help info
help_json_path = "dictionary/help_discord.json"
modules.utility.initialize_help_data(
bot=self,
help_json_path=help_json_path,
is_discord=True
)
_result = "Commands reloaded successfully"
globals.log("Discord commands reloaded successfully.")
await ctx.bot.unload_extension(f"cmd_discord.{cog_name}")
await ctx.bot.load_extension(f"cmd_discord.{cog_name}")
await ctx.reply(f"✅ Reloaded `{cog_name}` successfully!")
globals.log(f"Successfully reloaded the command cog `{cog_name}`", "INFO")
except Exception as e:
_result = f"Error reloading Discord commands: {e}"
globals.log(_result, "ERROR", True)
await ctx.reply(_result)
await ctx.reply(f"❌ Error reloading `{cog_name}`: {e}")
globals.log(f"Failed to reload the command cog `{cog_name}`", "ERROR")
async def on_message(self, message):
if message.guild:
@ -95,17 +88,16 @@ class DiscordBot(commands.Bot):
channel_name = "Direct Message"
globals.log(f"Message detected by '{message.author.name}' in '{guild_name}' - #'{channel_name}'", "DEBUG")
globals.log(f"Message content: '{message.content}'", "DEBUG")
globals.log(f"Attempting UUI lookup on '{message.author.name}' ...", "DEBUG")
try:
# If it's a bot message, ignore or pass user_is_bot=True
is_bot = message.author.bot
user_id = str(message.author.id)
user_name = message.author.name # no discriminator
user_name = message.author.name
display_name = message.author.display_name
platform_str = f"discord-{guild_name}"
channel_str = channel_name
# Track user activity first
# Track user activity
modules.utility.track_user_activity(
db_conn=self.db_conn,
platform="discord",
@ -115,10 +107,6 @@ class DiscordBot(commands.Bot):
user_is_bot=is_bot
)
# Let log_message() handle UUID lookup internally
platform_str = f"discord-{guild_name}"
channel_str = channel_name
attachments = ", ".join(a.url for a in message.attachments) if message.attachments else ""
log_message(
@ -128,19 +116,152 @@ class DiscordBot(commands.Bot):
message_content=message.content or "",
platform=platform_str,
channel=channel_str,
attachments=attachments
attachments=attachments,
platform_message_id=str(message.id) # Include Discord message ID
)
except Exception as e:
globals.log(f"... UUI lookup failed: {e}", "WARNING")
pass
try:
# Pass message contents to commands processing
await self.process_commands(message)
globals.log(f"Command processing complete", "DEBUG")
except Exception as e:
globals.log(f"Command processing failed: {e}", "ERROR")
# async def on_reaction_add(self, reaction, user):
# if user.bot:
# return # Ignore bot reactions
# guild_name = reaction.message.guild.name if reaction.message.guild else "DM"
# channel_name = reaction.message.channel.name if hasattr(reaction.message.channel, "name") else "Direct Message"
# globals.log(f"Reaction added by '{user.name}' on message '{reaction.message.id}' in '{guild_name}' - #{channel_name}", "DEBUG")
# modules.utility.track_user_activity(
# db_conn=self.db_conn,
# platform="discord",
# user_id=str(user.id),
# username=user.name,
# display_name=user.display_name,
# user_is_bot=user.bot
# )
# platform_str = f"discord-{guild_name}"
# channel_str = channel_name
# log_message(
# db_conn=self.db_conn,
# identifier=str(user.id),
# identifier_type="discord_user_id",
# message_content=f"Reaction Added: {reaction.emoji} on Message ID: {reaction.message.id}",
# platform=platform_str,
# channel=channel_str
# )
# async def on_reaction_remove(self, reaction, user):
# if user.bot:
# return # Ignore bot reactions
# guild_name = reaction.message.guild.name if reaction.message.guild else "DM"
# channel_name = reaction.message.channel.name if hasattr(reaction.message.channel, "name") else "Direct Message"
# globals.log(f"Reaction removed by '{user.name}' on message '{reaction.message.id}' in '{guild_name}' - #{channel_name}", "DEBUG")
# modules.utility.track_user_activity(
# db_conn=self.db_conn,
# platform="discord",
# user_id=str(user.id),
# username=user.name,
# display_name=user.display_name,
# user_is_bot=user.bot
# )
# platform_str = f"discord-{guild_name}"
# channel_str = channel_name
# log_message(
# db_conn=self.db_conn,
# identifier=str(user.id),
# identifier_type="discord_user_id",
# message_content=f"Reaction Removed: {reaction.emoji} on Message ID: {reaction.message.id}",
# platform=platform_str,
# channel=channel_str
# )
# async def on_message_edit(self, before, after):
# if before.author.bot:
# return # Ignore bot edits
# guild_name = before.guild.name if before.guild else "DM"
# channel_name = before.channel.name if hasattr(before.channel, "name") else "Direct Message"
# globals.log(f"Message edited by '{before.author.name}' in '{guild_name}' - #{channel_name}", "DEBUG")
# globals.log(f"Before: {before.content}\nAfter: {after.content}", "DEBUG")
# modules.utility.track_user_activity(
# db_conn=self.db_conn,
# platform="discord",
# user_id=str(before.author.id),
# username=before.author.name,
# display_name=before.author.display_name,
# user_is_bot=before.author.bot
# )
# platform_str = f"discord-{guild_name}"
# channel_str = channel_name
# log_message(
# db_conn=self.db_conn,
# identifier=str(before.author.id),
# identifier_type="discord_user_id",
# message_content=f"Message Edited:\nBefore: {before.content}\nAfter: {after.content}",
# platform=platform_str,
# channel=channel_str
# )
# async def on_thread_create(self, thread):
# globals.log(f"Thread '{thread.name}' created in #{thread.parent.name}", "DEBUG")
# modules.utility.track_user_activity(
# db_conn=self.db_conn,
# platform="discord",
# user_id=str(thread.owner_id),
# username=thread.owner.name if thread.owner else "Unknown",
# display_name=thread.owner.display_name if thread.owner else "Unknown",
# user_is_bot=False
# )
# log_message(
# db_conn=self.db_conn,
# identifier=str(thread.owner_id),
# identifier_type="discord_user_id",
# message_content=f"Thread Created: {thread.name} in #{thread.parent.name}",
# platform=f"discord-{thread.guild.name}",
# channel=thread.parent.name
# )
# async def on_thread_update(self, before, after):
# globals.log(f"Thread updated: '{before.name}' -> '{after.name}'", "DEBUG")
# log_message(
# db_conn=self.db_conn,
# identifier=str(before.owner_id),
# identifier_type="discord_user_id",
# message_content=f"Thread Updated: '{before.name}' -> '{after.name}'",
# platform=f"discord-{before.guild.name}",
# channel=before.parent.name
# )
# async def on_thread_delete(self, thread):
# globals.log(f"Thread '{thread.name}' deleted", "DEBUG")
# log_message(
# db_conn=self.db_conn,
# identifier=str(thread.owner_id),
# identifier_type="discord_user_id",
# message_content=f"Thread Deleted: {thread.name}",
# platform=f"discord-{thread.guild.name}",
# channel=thread.parent.name
# )
def load_bot_settings(self):
"""Loads bot activity settings from JSON file."""

27
bot_twitch.py
View File

@ -46,27 +46,17 @@ class TwitchBot(commands.Bot):
async def event_message(self, message):
"""
Called every time a Twitch message is received (chat message in a channel).
We'll use this to track the user in our 'users' table.
"""
# If it's the bot's own message, ignore
if message.echo:
return
# Log the command if it's a command
if message.content.startswith("!"):
_cmd = message.content[1:] # Remove the leading "!"
_cmd_args = _cmd.split(" ")[1:]
_cmd = _cmd.split(" ", 1)[0]
globals.log(f"Command '{_cmd}' (Twitch) initiated by {message.author.name} in #{message.channel.name}", "DEBUG")
if len(_cmd_args) > 1: globals.log(f"!{_cmd} arguments: {_cmd_args}", "DEBUG")
try:
# Typically message.author is not None for normal chat messages
author = message.author
if not author: # just in case
if not author:
return
is_bot = False # TODO Implement automatic bot account check
is_bot = False
user_id = str(author.id)
user_name = author.name
display_name = author.display_name or user_name
@ -84,23 +74,20 @@ class TwitchBot(commands.Bot):
globals.log("... UUI lookup complete.", "DEBUG")
user_data = lookup_user(db_conn=self.db_conn, identifier=str(message.author.id), identifier_type="twitch_user_id")
user_uuid = user_data["UUID"] if user_data else "UNKNOWN"
from modules.db import log_message
log_message(
db_conn=self.db_conn,
identifier=str(message.author.id),
identifier=user_id,
identifier_type="twitch_user_id",
message_content=message.content or "",
platform="twitch",
channel=message.channel.name,
attachments=""
attachments="",
platform_message_id=str(message.id) # Include Twitch message ID
)
except Exception as e:
globals.log(f"... UUI lookup failed: {e}", "ERROR")
# Pass message contents to commands processing
await self.handle_commands(message)

28
cmd_discord/__init__.py
View File

@ -2,18 +2,18 @@
import os
import importlib
def setup(bot):
"""
Dynamically load all commands from the cmd_discord folder.
"""
# Get a list of all command files (excluding __init__.py)
command_files = [
f.replace('.py', '') for f in os.listdir(os.path.dirname(__file__))
if f.endswith('.py') and f != '__init__.py'
]
# def setup(bot):
# """
# Dynamically load all commands from the cmd_discord folder.
# """
# # Get a list of all command files (excluding __init__.py)
# command_files = [
# f.replace('.py', '') for f in os.listdir(os.path.dirname(__file__))
# if f.endswith('.py') and f != '__init__.py'
# ]
# Import and set up each command module
for command in command_files:
module = importlib.import_module(f".{command}", package='cmd_discord')
if hasattr(module, 'setup'):
module.setup(bot)
# # Import and set up each command module
# for command in command_files:
# module = importlib.import_module(f".{command}", package='cmd_discord')
# if hasattr(module, 'setup'):
# module.setup(bot)

927
cmd_discord/customvc.py
View File

@ -3,513 +3,500 @@
import discord
from discord.ext import commands
import datetime
import globals
from modules.permissions import has_custom_vc_permission
# IDs
LOBBY_VC_ID = 1345509388651069583 # The lobby voice channel
CUSTOM_VC_CATEGORY_ID = 1345509307503874149 # Where new custom VCs go
class CustomVCCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.settings_data = globals.load_settings_file("discord_guilds_config.json")
# Track data in memory
CUSTOM_VC_INFO = {} # voice_chan_id -> {owner_id, locked, allowed_ids, denied_ids, ...}
USER_LAST_CREATED = {} # user_id -> datetime (for rate-limits)
GLOBAL_CREATIONS = [] # list of datetime for global rate-limits
CHANNEL_COUNTER = 0
guild_id = "896713616089309184" # Adjust based on your settings structure
self.LOBBY_VC_ID = self.settings_data[guild_id]["customvc_settings"]["lobby_vc_id"]
self.CUSTOM_VC_CATEGORY_ID = self.settings_data[guild_id]["customvc_settings"]["customvc_category_id"]
self.USER_COOLDOWN_MINUTES = self.settings_data[guild_id]["customvc_settings"]["vc_creation_user_cooldown"]
self.GLOBAL_CHANNELS_PER_MINUTE = self.settings_data[guild_id]["customvc_settings"]["vc_creation_global_per_min"]
self.DELETE_DELAY_SECONDS = self.settings_data[guild_id]["customvc_settings"]["empty_vc_autodelete_delay"]
self.MAX_CUSTOM_CHANNELS = self.settings_data[guild_id]["customvc_settings"]["customvc_max_limit"]
# Rate-limits
USER_COOLDOWN_MINUTES = 5
GLOBAL_CHANNELS_PER_MINUTE = 2
self.CUSTOM_VC_INFO = {}
self.USER_LAST_CREATED = {}
self.GLOBAL_CREATIONS = []
self.CHANNEL_COUNTER = 0
self.PENDING_DELETIONS = {}
# Auto-delete scheduling
PENDING_DELETIONS = {}
DELETE_DELAY_SECONDS = 10
@commands.Cog.listener()
async def on_ready(self):
"""Handles checking existing voice channels on bot startup."""
await self.scan_existing_custom_vcs()
# Arbitrary max, so new channels get “VC x” or “Overflow x”
MAX_CUSTOM_CHANNELS = 9
@commands.Cog.listener()
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState):
"""Handles user movement between voice channels."""
await self.on_voice_update(member, before, after)
def setup(bot: commands.Bot):
"""
Called by your `load_commands()` in bot_discord.py
"""
@bot.listen("on_ready")
async def after_bot_ready():
@commands.group(name="customvc", invoke_without_command=True)
async def customvc(self, ctx):
"""
When the bot starts, we scan leftover custom voice channels in the category:
- If it's the lobby => skip
- If it's empty => delete
- If it has users => pick first occupant as owner
Then mention them in that channel's built-in text chat (if supported).
Base !customvc command -> show help if no subcommand used.
"""
await scan_existing_custom_vcs(bot)
if ctx.invoked_subcommand is None:
msg = (
"**Custom VC Help**\n"
"Join the lobby to create a new voice channel automatically.\n"
"Subcommands:\n"
"- **name** <new_name> - rename\n"
" - `!customvc name my_custom_vc`\n"
"- **claim** - claim ownership\n"
" - `!customvc claim`\n"
"- **lock** - lock the channel\n"
" - `!customvc lock`\n"
"- **allow** <user> - allow user\n"
" - `!customvc allow some_user`\n"
"- **deny** <user> - deny user\n"
" - `!customvc deny some_user`\n"
"- **unlock** - unlock channel\n"
" - `!customvc unlock`\n"
"- **users** <count> - set user limit\n"
" - `!customvc users 2`\n"
"- **bitrate** <kbps> - set channel bitrate\n"
" - `!customvc bitrate some_bitrate`\n"
"- **status** <status_str> - set a custom status **(TODO)**\n"
"- **op** <user> - co-owner\n"
" - `!customvc op some_user`\n"
"- **settings** - show config\n"
" - `!customvc settings`\n"
)
await ctx.reply(msg)
else:
await self.bot.invoke(ctx) # This will ensure subcommands get processed.
@bot.listen("on_voice_state_update")
async def handle_voice_update(member: discord.Member, before: discord.VoiceState, after: discord.VoiceState):
# The logic for creation, auto-delete, etc.
await on_voice_state_update(bot, member, before, after)
@bot.group(name="customvc", invoke_without_command=True)
async def customvc_base(ctx: commands.Context):
"""
Base !customvc command -> show help if no subcommand used
"""
msg = (
"**Custom VC Help**\n"
"Join the lobby to create a new voice channel automatically.\n"
"Subcommands:\n"
"- **name** <new_name> - rename\n"
" - `!customvc name my_custom_vc`\n"
"- **claim** - claim ownership\n"
" - `!customvc claim`\n"
"- **lock** - lock the channel\n"
" - `!customvc lock`\n"
"- **allow** <user> - allow user\n"
" - `!customvc allow some_user`\n"
"- **deny** <user> - deny user\n"
" - `!customvc deny some_user`\n"
"- **unlock** - unlock channel\n"
" - `!customvc unlock`\n"
"- **users** <count> - set user limit\n"
" - `!customvc users 2`\n"
"- **bitrate** <kbps> - set channel bitrate\n"
" - `!customvc bitrate some_bitrate`\n"
"- **status** <status_str> - set a custom status **(TODO)**\n"
"- **op** <user> - co-owner\n"
" - `!customvc op some_user`\n"
"- **settings** - show config\n"
" - `!customvc settings`\n"
)
await ctx.send(msg)
# Subcommands:
@customvc_base.command(name="name")
async def name_subcommand(ctx: commands.Context, *, new_name: str):
await rename_channel(ctx, new_name)
@customvc.command(name="name")
async def rename_channel(self, ctx, *, new_name: str):
"""Renames a custom VC."""
await self.rename_channel_logic(ctx, new_name)
@customvc_base.command(name="claim")
async def claim_subcommand(ctx: commands.Context):
await claim_channel(ctx)
@customvc.command(name="claim")
async def claim_channel(self, ctx):
"""Claims ownership of a VC if the owner is absent."""
await self.claim_channel_logic(ctx)
@customvc_base.command(name="lock")
async def lock_subcommand(ctx: commands.Context):
await lock_channel(ctx)
@customvc.command(name="lock")
async def lock_channel(self, ctx):
"""Locks a custom VC."""
await self.lock_channel_logic(ctx)
@customvc_base.command(name="allow")
async def allow_subcommand(ctx: commands.Context, *, user: str):
await allow_user(ctx, user)
@customvc.command(name="allow")
async def allow_user(self, ctx, *, user: str):
"""Allows a user to join a VC."""
await self.allow_user_logic(ctx, user)
@customvc_base.command(name="deny")
async def deny_subcommand(ctx: commands.Context, *, user: str):
await deny_user(ctx, user)
@customvc.command(name="deny")
async def deny_user(self, ctx, *, user: str):
"""Denies a user from joining a VC."""
await self.deny_user_logic(ctx, user)
@customvc_base.command(name="unlock")
async def unlock_subcommand(ctx: commands.Context):
await unlock_channel(ctx)
@customvc.command(name="unlock")
async def unlock_channel(self, ctx):
"""Unlocks a custom VC."""
await self.unlock_channel_logic(ctx)
@customvc_base.command(name="users")
async def users_subcommand(ctx: commands.Context, limit: int):
await set_user_limit(ctx, limit)
@customvc.command(name="settings")
async def show_settings(self, ctx):
"""Shows the settings of the current VC."""
await self.show_settings_logic(ctx)
@customvc_base.command(name="bitrate")
async def bitrate_subcommand(ctx: commands.Context, kbps: int):
await set_bitrate(ctx, kbps)
@customvc_base.command(name="status")
async def status_subcommand(ctx: commands.Context, *, status_str: str):
await set_status(ctx, status_str)
@customvc_base.command(name="op")
async def op_subcommand(ctx: commands.Context, *, user: str):
await op_user(ctx, user)
@customvc_base.command(name="settings")
async def settings_subcommand(ctx: commands.Context):
await show_settings(ctx)
#
# Main voice update logic
#
async def on_voice_state_update(bot: commands.Bot, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState):
global CHANNEL_COUNTER
if before.channel != after.channel:
# 1) If user just joined the lobby
if after.channel and after.channel.id == LOBBY_VC_ID:
if not await can_create_vc(member):
# Rate-limit => forcibly disconnect
await member.move_to(None)
# Also mention them in the ephemeral chat of LOBBY_VC_ID if we want (if possible)
lobby_chan = after.channel
if lobby_chan and hasattr(lobby_chan, "send"):
# We'll try to mention them
await lobby_chan.send(f"{member.mention}, you’ve exceeded custom VC creation limits. Try again later!")
return
# 2) Create a new custom VC
CHANNEL_COUNTER += 1
vc_name = f"VC {CHANNEL_COUNTER}"
if CHANNEL_COUNTER > MAX_CUSTOM_CHANNELS:
vc_name = f"Overflow {CHANNEL_COUNTER}"
category = member.guild.get_channel(CUSTOM_VC_CATEGORY_ID)
if not category or not isinstance(category, discord.CategoryChannel):
print("Could not find a valid custom VC category.")
return
new_vc = await category.create_voice_channel(name=vc_name)
await member.move_to(new_vc)
# Record memory
CUSTOM_VC_INFO[new_vc.id] = {
"owner_id": member.id,
"locked": False,
"allowed_ids": set(),
"denied_ids": set(),
"user_limit": None,
"bitrate": None,
}
now = datetime.datetime.utcnow()
USER_LAST_CREATED[member.id] = now
GLOBAL_CREATIONS.append(now)
# prune old
cutoff = now - datetime.timedelta(seconds=60)
GLOBAL_CREATIONS[:] = [t for t in GLOBAL_CREATIONS if t > cutoff]
# Announce in the new VC's ephemeral chat
if hasattr(new_vc, "send"):
await new_vc.send(
f"{member.mention}, your custom voice channel is ready! "
"Type `!customvc` here for help with subcommands."
)
# 3) If user left a custom VC -> maybe it’s now empty
if before.channel and before.channel.id in CUSTOM_VC_INFO:
old_vc = before.channel
if len(old_vc.members) == 0:
schedule_deletion(old_vc)
# If user joined a custom VC that was pending deletion, cancel it
if after.channel and after.channel.id in CUSTOM_VC_INFO:
if after.channel.id in PENDING_DELETIONS:
PENDING_DELETIONS[after.channel.id].cancel()
del PENDING_DELETIONS[after.channel.id]
@customvc.command(name="users")
async def set_users_limit(self, ctx):
"""Assign a VC users limit"""
await self.set_user_limit_logic(ctx)
def schedule_deletion(vc: discord.VoiceChannel):
"""
Schedules this custom VC for deletion in 10s if it stays empty.
"""
import asyncio
#
# Main voice update logic
#
async def delete_task():
await asyncio.sleep(DELETE_DELAY_SECONDS)
if len(vc.members) == 0:
CUSTOM_VC_INFO.pop(vc.id, None)
try:
await vc.delete()
except:
pass
PENDING_DELETIONS.pop(vc.id, None)
async def on_voice_update(self, member, before, after):
if before.channel != after.channel:
# 1) If user just joined the lobby
if after.channel and after.channel.id == self.LOBBY_VC_ID:
if not await self.can_create_vc(member):
# Rate-limit => forcibly disconnect
await member.move_to(None)
# Also mention them in the ephemeral chat of LOBBY_VC_ID if we want (if possible)
lobby_chan = after.channel
if lobby_chan and hasattr(lobby_chan, "send"):
# We'll try to mention them
await lobby_chan.send(f"{member.mention}, you’ve exceeded custom VC creation limits. Try again later!")
return
loop = vc.guild._state.loop
task = loop.create_task(delete_task())
PENDING_DELETIONS[vc.id] = task
# 2) Create a new custom VC
self.CHANNEL_COUNTER += 1
vc_name = f"VC {self.CHANNEL_COUNTER}"
if self.CHANNEL_COUNTER > self.MAX_CUSTOM_CHANNELS:
vc_name = f"Overflow {self.CHANNEL_COUNTER}"
async def can_create_vc(member: discord.Member) -> bool:
"""
Return False if user or global rate-limits are hit:
- user can only create 1 channel every 5 minutes
- globally only 2 channels per minute
"""
now = datetime.datetime.utcnow()
category = member.guild.get_channel(self.CUSTOM_VC_CATEGORY_ID)
if not category or not isinstance(category, discord.CategoryChannel):
globals.log("Could not find a valid custom VC category.", "INFO")
return
last_time = USER_LAST_CREATED.get(member.id)
if last_time:
diff = (now - last_time).total_seconds()
if diff < (USER_COOLDOWN_MINUTES * 60):
return False
new_vc = await category.create_voice_channel(name=vc_name)
await member.move_to(new_vc)
# global limit
cutoff = now - datetime.timedelta(seconds=60)
recent = [t for t in GLOBAL_CREATIONS if t > cutoff]
if len(recent) >= GLOBAL_CHANNELS_PER_MINUTE:
return False
return True
async def scan_existing_custom_vcs(bot: commands.Bot):
"""
On startup: check the custom VC category for leftover channels:
- If channel.id == LOBBY_VC_ID -> skip (don't delete the main lobby).
- If empty, delete immediately.
- If non-empty, first occupant is new owner. Mention them in ephemeral chat if possible.
"""
for g in bot.guilds:
cat = g.get_channel(CUSTOM_VC_CATEGORY_ID)
if not cat or not isinstance(cat, discord.CategoryChannel):
continue
for ch in cat.voice_channels:
# skip the LOBBY
if ch.id == LOBBY_VC_ID:
continue
if len(ch.members) == 0:
# safe to delete
try:
await ch.delete()
except:
pass
print(f"Deleted empty leftover channel: {ch.name}")
else:
# pick first occupant
first = ch.members[0]
global CHANNEL_COUNTER
CHANNEL_COUNTER += 1
CUSTOM_VC_INFO[ch.id] = {
"owner_id": first.id,
# Record memory
self.CUSTOM_VC_INFO[new_vc.id] = {
"owner_id": member.id,
"locked": False,
"allowed_ids": set(),
"denied_ids": set(),
"user_limit": None,
"bitrate": None,
}
print(f"Assigned {first.display_name} as owner of leftover VC: {ch.name}")
if hasattr(ch, "send"):
now = datetime.datetime.utcnow()
self.USER_LAST_CREATED[member.id] = now
self.GLOBAL_CREATIONS.append(now)
# prune old
cutoff = now - datetime.timedelta(seconds=60)
self.GLOBAL_CREATIONS[:] = [t for t in self.GLOBAL_CREATIONS if t > cutoff]
# Announce in the new VC's ephemeral chat
if hasattr(new_vc, "send"):
await new_vc.send(
f"{member.mention}, your custom voice channel is ready! "
"Type `!customvc` here for help with subcommands."
)
# 3) If user left a custom VC -> maybe it’s now empty
if before.channel and before.channel.id in self.CUSTOM_VC_INFO:
old_vc = before.channel
if len(old_vc.members) == 0:
self.schedule_deletion(old_vc)
# If user joined a custom VC that was pending deletion, cancel it
if after.channel and after.channel.id in self.CUSTOM_VC_INFO:
if after.channel.id in self.PENDING_DELETIONS:
self.PENDING_DELETIONS[after.channel.id].cancel()
del self.PENDING_DELETIONS[after.channel.id]
def schedule_deletion(self, vc: discord.VoiceChannel):
"""
Schedules this custom VC for deletion in 10s if it stays empty.
"""
import asyncio
async def delete_task():
await asyncio.sleep(self.DELETE_DELAY_SECONDS)
if len(vc.members) == 0:
self.CUSTOM_VC_INFO.pop(vc.id, None)
try:
await vc.delete()
except:
pass
self.PENDING_DELETIONS.pop(vc.id, None)
loop = vc.guild._state.loop
task = loop.create_task(delete_task())
self.PENDING_DELETIONS[vc.id] = task
async def can_create_vc(self, member: discord.Member) -> bool:
"""
Return False if user or global rate-limits are hit:
- user can only create 1 channel every 5 minutes
- globally only 2 channels per minute
"""
now = datetime.datetime.utcnow()
last_time = self.USER_LAST_CREATED.get(member.id)
if last_time:
diff = (now - last_time).total_seconds()
if diff < (self.USER_COOLDOWN_MINUTES * 60):
return False
# global limit
cutoff = now - datetime.timedelta(seconds=60)
recent = [t for t in self.GLOBAL_CREATIONS if t > cutoff]
if len(recent) >= self.GLOBAL_CHANNELS_PER_MINUTE:
return False
return True
async def scan_existing_custom_vcs(self):
"""
On startup: check the custom VC category for leftover channels:
- If channel.id == LOBBY_VC_ID -> skip (don't delete the main lobby).
- If empty, delete immediately.
- If non-empty, first occupant is new owner. Mention them in ephemeral chat if possible.
"""
for g in self.bot.guilds:
cat = g.get_channel(self.CUSTOM_VC_CATEGORY_ID)
if not cat or not isinstance(cat, discord.CategoryChannel):
continue
for ch in cat.voice_channels:
# skip the LOBBY
if ch.id == self.LOBBY_VC_ID:
continue
if len(ch.members) == 0:
# safe to delete
try:
await ch.send(
f"{first.mention}, you're now the owner of leftover channel **{ch.name}** after a restart. "
"Type `!customvc` here to see available commands."
)
await ch.delete()
except:
pass
globals.log(f"Deleted empty leftover channel: {ch.name}", "INFO")
else:
# pick first occupant
first = ch.members[0]
self.CHANNEL_COUNTER += 1
self.CUSTOM_VC_INFO[ch.id] = {
"owner_id": first.id,
"locked": False,
"allowed_ids": set(),
"denied_ids": set(),
"user_limit": None,
"bitrate": None,
}
globals.log(f"Assigned {first.display_name} as owner of leftover VC: {ch.name}", "INFO")
if hasattr(ch, "send"):
try:
await ch.send(
f"{first.mention}, you're now the owner of leftover channel **{ch.name}** after a restart. "
"Type `!customvc` here to see available commands."
)
except:
pass
#
# The subcommand logic below, referencing CUSTOM_VC_INFO
#
#
# The subcommand logic below, referencing CUSTOM_VC_INFO
#
def get_custom_vc_for(ctx: commands.Context) -> discord.VoiceChannel:
"""
Return the custom voice channel the command user is currently in (or None).
We'll look at ctx.author.voice.
"""
if not isinstance(ctx.author, discord.Member):
def get_custom_vc_for(self, ctx: commands.Context) -> discord.VoiceChannel:
"""
Return the custom voice channel the command user is currently in (or None).
We'll look at ctx.author.voice.
"""
if not isinstance(ctx.author, discord.Member):
return None
vs = ctx.author.voice
if not vs or not vs.channel:
return None
vc = vs.channel
if vc.id not in self.CUSTOM_VC_INFO:
return None
return vc
def is_initiator_or_mod(self, ctx: commands.Context, vc_id: int) -> bool:
info = self.CUSTOM_VC_INFO.get(vc_id)
if not info:
return False
return has_custom_vc_permission(ctx.author, info["owner_id"])
async def rename_channel_logic(self, ctx: commands.Context, new_name: str):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("You are not in a custom voice channel.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to rename.")
await vc.edit(name=new_name)
await ctx.send(f"Renamed channel to **{new_name}**.")
async def claim_channel_logic(self, ctx: commands.Context):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
info = self.CUSTOM_VC_INFO.get(vc.id)
if not info:
return await ctx.send("No memory for this channel?")
old = info["owner_id"]
# if old owner is still inside
if any(m.id == old for m in vc.members):
return await ctx.send("The original owner is still here; you cannot claim.")
info["owner_id"] = ctx.author.id
await ctx.send("You are now the owner of this channel!")
async def lock_channel_logic(self, ctx: commands.Context):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("You're not in a custom VC.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to lock.")
info = self.CUSTOM_VC_INFO[vc.id]
info["locked"] = True
overwrites = vc.overwrites or {}
overwrites[ctx.guild.default_role] = discord.PermissionOverwrite(connect=False)
await vc.edit(overwrites=overwrites)
await ctx.send("Channel locked.")
async def allow_user_logic(self, ctx: commands.Context, user: str):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to allow a user.")
mem = await self.resolve_member(ctx, user)
if not mem:
return await ctx.send(f"Could not find user: {user}.")
info = self.CUSTOM_VC_INFO[vc.id]
info["allowed_ids"].add(mem.id)
overwrites = vc.overwrites or {}
overwrites[mem] = discord.PermissionOverwrite(connect=True)
await vc.edit(overwrites=overwrites)
await ctx.send(f"Allowed **{mem.display_name}** to join.")
async def deny_user_logic(self, ctx: commands.Context, user: str):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to deny user.")
mem = await self.resolve_member(ctx, user)
if not mem:
return await ctx.send(f"Could not find user: {user}.")
info = self.CUSTOM_VC_INFO[vc.id]
# check if they're mod or owner
if has_custom_vc_permission(mem, info["owner_id"]):
return await ctx.send("Cannot deny a moderator or the owner.")
info["denied_ids"].add(mem.id)
overwrites = vc.overwrites or {}
overwrites[mem] = discord.PermissionOverwrite(connect=False)
await vc.edit(overwrites=overwrites)
await ctx.send(f"Denied {mem.display_name} from connecting.")
async def unlock_channel_logic(self, ctx: commands.Context):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to unlock.")
info = self.CUSTOM_VC_INFO[vc.id]
info["locked"] = False
overwrites = vc.overwrites or {}
overwrites[ctx.guild.default_role] = discord.PermissionOverwrite(connect=True)
for d_id in info["denied_ids"]:
m = ctx.guild.get_member(d_id)
if m:
overwrites[m] = discord.PermissionOverwrite(connect=False)
await vc.edit(overwrites=overwrites)
await ctx.send("Unlocked the channel. Denied users still cannot join.")
async def set_user_limit_logic(self, ctx: commands.Context, limit: int):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if limit < 2:
return await ctx.send("Minimum limit is 2.")
if limit > 99:
return await ctx.send("Maximum limit is 99.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to set user limit.")
info = self.CUSTOM_VC_INFO[vc.id]
info["user_limit"] = limit
await vc.edit(user_limit=limit)
await ctx.send(f"User limit set to {limit}.")
async def set_bitrate_logic(self, ctx: commands.Context, kbps: int):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to set bitrate.")
info = self.CUSTOM_VC_INFO[vc.id]
info["bitrate"] = kbps
if kbps < 8 or kbps > 128:
return await ctx.send(f"Invalid bitrate of {kbps} kbps! Must be a number between `8` and `128`! Default is 64 kbps")
await vc.edit(bitrate=kbps * 1000)
await ctx.send(f"Bitrate set to {kbps} kbps.")
async def set_status_logic(self, ctx: commands.Context, status_str: str):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to set channel status.")
info = self.CUSTOM_VC_INFO[vc.id]
info["status"] = status_str
await ctx.send(f"Channel status set to: **{status_str}** (placeholder).")
async def op_user_logic(self, ctx: commands.Context, user: str):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not self.is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to op someone.")
mem = await self.resolve_member(ctx, user)
if not mem:
return await ctx.send(f"Could not find user: {user}.")
info = self.CUSTOM_VC_INFO[vc.id]
if "ops" not in info:
info["ops"] = set()
info["ops"].add(mem.id)
await ctx.send(f"Granted co-ownership to {mem.display_name}.")
async def show_settings_logic(self, ctx: commands.Context):
vc = self.get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
info = self.CUSTOM_VC_INFO.get(vc.id)
if not info:
return await ctx.send("No data found for this channel?")
locked_str = "Yes" if info["locked"] else "No"
user_lim = info["user_limit"] if info["user_limit"] else "Default"
bitrate_str = info["bitrate"] if info["bitrate"] else "Default (64 kbps)"
status_str = info.get("status", "None")
owner_id = info["owner_id"]
owner = ctx.guild.get_member(owner_id)
owner_str = owner.display_name if owner else f"<ID {owner_id}>"
allowed_str = ", ".join(map(str, info["allowed_ids"])) if info["allowed_ids"] else "None specified"
denied_str = ", ".join(map(str, info["denied_ids"])) if info["denied_ids"] else "None specified"
msg = (
f"**Channel Settings**\n"
f"Owner: {owner_str}\n"
f"Locked: {locked_str}\n"
f"User Limit: {user_lim}\n"
f"Bitrate: {bitrate_str}\n"
f"Status: {status_str}\n"
f"Allowed: {allowed_str}\n"
f"Denied: {denied_str}\n"
)
await ctx.send(msg)
async def resolve_member(ctx: commands.Context, user_str: str) -> discord.Member:
"""
Attempt to parse user mention/ID/partial name.
"""
user_str = user_str.strip()
if user_str.startswith("<@") and user_str.endswith(">"):
uid = user_str.strip("<@!>")
if uid.isdigit():
return ctx.guild.get_member(int(uid))
elif user_str.isdigit():
return ctx.guild.get_member(int(user_str))
lower = user_str.lower()
for m in ctx.guild.members:
if m.name.lower() == lower or m.display_name.lower() == lower:
return m
return None
vs = ctx.author.voice
if not vs or not vs.channel:
return None
vc = vs.channel
if vc.id not in CUSTOM_VC_INFO:
return None
return vc
def is_initiator_or_mod(ctx: commands.Context, vc_id: int) -> bool:
info = CUSTOM_VC_INFO.get(vc_id)
if not info:
return False
return has_custom_vc_permission(ctx.author, info["owner_id"])
async def rename_channel(ctx: commands.Context, new_name: str):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("You are not in a custom voice channel.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to rename.")
await vc.edit(name=new_name)
await ctx.send(f"Renamed channel to **{new_name}**.")
async def claim_channel(ctx: commands.Context):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
info = CUSTOM_VC_INFO.get(vc.id)
if not info:
return await ctx.send("No memory for this channel?")
old = info["owner_id"]
# if old owner is still inside
if any(m.id == old for m in vc.members):
return await ctx.send("The original owner is still here; you cannot claim.")
info["owner_id"] = ctx.author.id
await ctx.send("You are now the owner of this channel!")
async def lock_channel(ctx: commands.Context):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("You're not in a custom VC.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to lock.")
info = CUSTOM_VC_INFO[vc.id]
info["locked"] = True
overwrites = vc.overwrites or {}
overwrites[ctx.guild.default_role] = discord.PermissionOverwrite(connect=False)
await vc.edit(overwrites=overwrites)
await ctx.send("Channel locked.")
async def allow_user(ctx: commands.Context, user: str):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to allow a user.")
mem = await resolve_member(ctx, user)
if not mem:
return await ctx.send(f"Could not find user: {user}.")
info = CUSTOM_VC_INFO[vc.id]
info["allowed_ids"].add(mem.id)
overwrites = vc.overwrites or {}
overwrites[mem] = discord.PermissionOverwrite(connect=True)
await vc.edit(overwrites=overwrites)
await ctx.send(f"Allowed **{mem.display_name}** to join.")
async def deny_user(ctx: commands.Context, user: str):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to deny user.")
mem = await resolve_member(ctx, user)
if not mem:
return await ctx.send(f"Could not find user: {user}.")
info = CUSTOM_VC_INFO[vc.id]
# check if they're mod or owner
if has_custom_vc_permission(mem, info["owner_id"]):
return await ctx.send("Cannot deny a moderator or the owner.")
info["denied_ids"].add(mem.id)
overwrites = vc.overwrites or {}
overwrites[mem] = discord.PermissionOverwrite(connect=False)
await vc.edit(overwrites=overwrites)
await ctx.send(f"Denied {mem.display_name} from connecting.")
async def unlock_channel(ctx: commands.Context):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to unlock.")
info = CUSTOM_VC_INFO[vc.id]
info["locked"] = False
overwrites = vc.overwrites or {}
overwrites[ctx.guild.default_role] = discord.PermissionOverwrite(connect=True)
for d_id in info["denied_ids"]:
m = ctx.guild.get_member(d_id)
if m:
overwrites[m] = discord.PermissionOverwrite(connect=False)
await vc.edit(overwrites=overwrites)
await ctx.send("Unlocked the channel. Denied users still cannot join.")
async def set_user_limit(ctx: commands.Context, limit: int):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if limit < 2:
return await ctx.send("Minimum limit is 2.")
if limit > 99:
return await ctx.send("Maximum limit is 99.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to set user limit.")
info = CUSTOM_VC_INFO[vc.id]
info["user_limit"] = limit
await vc.edit(user_limit=limit)
await ctx.send(f"User limit set to {limit}.")
async def set_bitrate(ctx: commands.Context, kbps: int):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to set bitrate.")
info = CUSTOM_VC_INFO[vc.id]
info["bitrate"] = kbps
if kbps < 8 or kbps > 128:
return await ctx.send(f"Invalid bitrate of {kbps} kbps! Must be a number between `8` and `128`! Default is 64 kbps")
await vc.edit(bitrate=kbps * 1000)
await ctx.send(f"Bitrate set to {kbps} kbps.")
async def set_status(ctx: commands.Context, status_str: str):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to set channel status.")
info = CUSTOM_VC_INFO[vc.id]
info["status"] = status_str
await ctx.send(f"Channel status set to: **{status_str}** (placeholder).")
async def op_user(ctx: commands.Context, user: str):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
if not is_initiator_or_mod(ctx, vc.id):
return await ctx.send("No permission to op someone.")
mem = await resolve_member(ctx, user)
if not mem:
return await ctx.send(f"Could not find user: {user}.")
info = CUSTOM_VC_INFO[vc.id]
if "ops" not in info:
info["ops"] = set()
info["ops"].add(mem.id)
await ctx.send(f"Granted co-ownership to {mem.display_name}.")
async def show_settings(ctx: commands.Context):
vc = get_custom_vc_for(ctx)
if not vc:
return await ctx.send("Not in a custom VC.")
info = CUSTOM_VC_INFO.get(vc.id)
if not info:
return await ctx.send("No data found for this channel?")
locked_str = "Yes" if info["locked"] else "No"
user_lim = info["user_limit"] if info["user_limit"] else "Default"
bitrate_str = info["bitrate"] if info["bitrate"] else "Default (64 kbps)"
status_str = info.get("status", "None")
owner_id = info["owner_id"]
owner = ctx.guild.get_member(owner_id)
owner_str = owner.display_name if owner else f"<ID {owner_id}>"
allowed_str = ", ".join(map(str, info["allowed_ids"])) if info["allowed_ids"] else "None specified"
denied_str = ", ".join(map(str, info["denied_ids"])) if info["denied_ids"] else "None specified"
msg = (
f"**Channel Settings**\n"
f"Owner: {owner_str}\n"
f"Locked: {locked_str}\n"
f"User Limit: {user_lim}\n"
f"Bitrate: {bitrate_str}\n"
f"Status: {status_str}\n"
f"Allowed: {allowed_str}\n"
f"Denied: {denied_str}\n"
)
await ctx.send(msg)
async def resolve_member(ctx: commands.Context, user_str: str) -> discord.Member:
"""
Attempt to parse user mention/ID/partial name.
"""
user_str = user_str.strip()
if user_str.startswith("<@") and user_str.endswith(">"):
uid = user_str.strip("<@!>")
if uid.isdigit():
return ctx.guild.get_member(int(uid))
elif user_str.isdigit():
return ctx.guild.get_member(int(user_str))
lower = user_str.lower()
for m in ctx.guild.members:
if m.name.lower() == lower or m.display_name.lower() == lower:
return m
return None
async def setup(bot):
await bot.add_cog(CustomVCCog(bot))

29
cmd_discord/funfact.py
View File

@ -1,14 +1,23 @@
# cmd_discord/howl.py
# cmd_discord/funfact.py
import discord
from discord.ext import commands
import cmd_common.common_commands as cc
def setup(bot):
"""
Registers the '!howl' command for Discord.
"""
@bot.command(name='funfact', aliases=['fun-fact'])
async def funfact_command(ctx, *keywords):
# keywords is a tuple of strings from the command arguments.
class FunfactCog(commands.Cog):
"""Cog for the '!funfact' command."""
def __init__(self, bot):
self.bot = bot
@commands.command(name='funfact', aliases=['fun-fact'])
async def funfact_command(self, ctx, *keywords):
"""
Fetches a fun fact based on provided keywords and replies to the user.
"""
fact = cc.get_fun_fact(list(keywords))
# Reply to the invoking user.
await ctx.reply(fact)
await ctx.reply(fact)
# Required for loading the cog dynamically
async def setup(bot):
await bot.add_cog(FunfactCog(bot))

49
cmd_discord/help.py
View File

@ -1,37 +1,30 @@
# cmd_discord/howl.py
from discord.ext import commands
# cmd_discord/help.py
import discord
from discord.ext import commands
from discord import app_commands
from typing import Optional
import cmd_common.common_commands as cc
from modules.utility import handle_help_command
import globals
# Retrieve primary guild info if needed (for logging or other purposes)
primary_guild = globals.constants.primary_discord_guild() # e.g., {"object": discord.Object(id=1234567890), "id": 1234567890}
class HelpCog(commands.Cog):
"""Handles the !help and /help commands."""
def setup(bot):
"""
Registers the '!help' command for Discord.
"""
@bot.command(name="help")
async def cmd_help_text(ctx, *, command: str = ""):
"""
Get help information about commands.
def __init__(self, bot):
self.bot = bot
self.primary_guild = globals.constants.primary_discord_guild()
Usage:
- !help
-> Provides a list of all commands with brief descriptions.
- !help <command>
-> Provides detailed help information for the specified command.
"""
result = await cc.handle_help_command(ctx, command, bot, is_discord=True)
@commands.command(name="help")
async def cmd_help_text(self, ctx, *, command: str = ""):
"""Handles text-based help command."""
result = await handle_help_command(ctx, command, self.bot, is_discord=True)
await ctx.reply(result)
# -------------------------------------------------------------------------
# SLASH COMMAND: help
# -------------------------------------------------------------------------
@bot.tree.command(name="help", description="Get information about commands", guild=primary_guild["object"])
@app_commands.describe(command="The command to get help info about. Defaults to 'help'")
async def cmd_help_slash(interaction: discord.Interaction, command: Optional[str] = ""):
result = await cc.handle_help_command(interaction, command, bot, is_discord=True)
await interaction.response.send_message(result)
@app_commands.command(name="help", description="Get information about commands")
async def cmd_help_slash(self, interaction: discord.Interaction, command: Optional[str] = ""):
"""Handles slash command for help."""
result = await handle_help_command(interaction, command, self.bot, is_discord=True)
await interaction.response.send_message(result)
async def setup(bot):
await bot.add_cog(HelpCog(bot))

26
cmd_discord/howl.py
View File

@ -1,18 +1,20 @@
# cmd_discord/howl.py
import discord
from discord.ext import commands
import cmd_common.common_commands as cc
def setup(bot):
"""
Registers the '!howl' command for Discord.
"""
@bot.command(name="howl")
async def cmd_howl_text(ctx):
"""
Handle the '!howl' command.
Usage:
- !howl -> Attempts a howl.
- !howl stat <user> -> Looks up howling stats for a user.
"""
class HowlCog(commands.Cog):
"""Cog for the '!howl' command."""
def __init__(self, bot):
self.bot = bot
@commands.command(name="howl")
async def cmd_howl_text(self, ctx):
"""Handles the '!howl' command."""
result = cc.handle_howl_command(ctx)
await ctx.reply(result)
async def setup(bot):
await bot.add_cog(HowlCog(bot))

29
cmd_discord/ping.py
View File

@ -1,21 +1,22 @@
# cmd_discord/howl.py
# cmd_discord/ping.py
import discord
from discord.ext import commands
import cmd_common.common_commands as cc
def setup(bot):
"""
Registers the '!ping' command for Discord.
"""
@bot.command(name="ping")
async def cmd_ping_text(ctx):
"""
Check the bot's uptime and latency.
class PingCog(commands.Cog):
"""Handles the '!ping' command."""
Usage:
- !ping
-> Returns the bot's uptime along with its latency in milliseconds.
"""
def __init__(self, bot):
self.bot = bot
@commands.command(name="ping")
async def cmd_ping_text(self, ctx):
"""Checks bot's uptime and latency."""
result = cc.ping()
latency = round(float(bot.latency) * 1000)
latency = round(self.bot.latency * 1000)
result += f" (*latency: {latency}ms*)"
await ctx.reply(result)
async def setup(bot):
await bot.add_cog(PingCog(bot))

43
cmd_discord/quote.py
View File

@ -1,41 +1,26 @@
# cmd_discord/howl.py
# cmd_discord/quote.py
import discord
from discord.ext import commands
import globals
import cmd_common.common_commands as cc
def setup(bot):
"""
Registers the '!quote' command for Discord.
"""
@bot.command(name="quote")
async def cmd_quote_text(ctx, *, arg_str: str = ""):
"""
Handle the '!quote' command with multiple subcommands.
class QuoteCog(commands.Cog):
"""Handles the '!quote' command."""
Usage:
- !quote
-> Retrieves a random (non-removed) quote.
- !quote <number>
-> Retrieves a specific quote by its ID.
- !quote add <quote text>
-> Adds a new quote and replies with its quote number.
- !quote remove <number>
-> Removes the specified quote.
- !quote restore <number>
-> Restores a previously removed quote.
- !quote info <number>
-> Displays stored information about the quote.
- !quote search [keywords]
-> Searches for the best matching quote.
- !quote latest
-> Retrieves the latest (most recent) non-removed quote.
"""
def __init__(self, bot):
self.bot = bot
@commands.command(name="quote")
async def cmd_quote_text(self, ctx, *, arg_str: str = ""):
"""Handles various quote-related subcommands."""
if not globals.init_db_conn:
await ctx.reply("Database is unavailable, sorry.")
return
args = arg_str.split() if arg_str else []
globals.log(f"'quote' command initiated with arguments: {args}", "DEBUG")
result = await cc.handle_quote_command(
db_conn=globals.init_db_conn,
is_discord=True,
@ -43,8 +28,12 @@ def setup(bot):
args=args,
game_name=None
)
globals.log(f"'quote' result: {result}", "DEBUG")
if hasattr(result, "to_dict"):
await ctx.reply(embed=result)
else:
await ctx.reply(result)
async def setup(bot):
await bot.add_cog(QuoteCog(bot))

3
dictionary/funfacts.json
View File

@ -532,5 +532,6 @@
"Cows are related to whales, as they both evolved from land-dwelling, even-toed ungulates, hence why a baby whale is called a 'calf'.",
"McNamee was the first to reach 999 of an item, sticks, and was awarded the sticker 'McNamee's Stick' for his efforts.",
"Jenni is 159cm (5.2 feet) tall, while Kami is 186cm (6.1 feet), making Kami almost 20% taller, or the length of a sheet of A4 paper.",
"If Jenni were to bury Kami's dead body in the back of the garden, she wouldn't be able to get out of the hole without a ladder."
"If Jenni were to bury Kami's dead body in the back of the garden, she wouldn't be able to get out of the hole without a ladder.",
"Aibophobia is the suggested name for a fear of words that can be reversed without changing the word. Aibophobia itself can be reversed."
]

252
dictionary/help_discord.json
View File

@ -1,106 +1,164 @@
{
"commands": {
"help": {
"description": "Show information about available commands.",
"subcommands": {},
"examples": [
"help",
"help quote"
]
},
"quote": {
"description": "Manage quotes (add, remove, restore, fetch, info).",
"subcommands": {
"no subcommand": {
"desc": "Fetch a random quote."
},
"[quote_number]": {
"args": "[quote_number]",
"desc": "Fetch a specific quote by ID."
},
"search": {
"args": "[keywords]",
"desc": "Search for a quote with keywords."
},
"last/latest/newest": {
"desc": "Returns the newest quote."
},
"add": {
"args": "[quote_text]",
"desc": "Adds a new quote."
},
"remove": {
"args": "[quote_number]",
"desc": "Removes the specified quote by ID."
},
"restore": {
"args": "[quote_number]",
"desc": "Restores the specified quote by ID."
},
"info": {
"args": "[quote_number]",
"desc": "Retrieves info about the specified quote."
}
"commands": {
"help": {
"description": "Show information about available commands.",
"subcommands": {},
"examples": [
"help",
"help quote"
]
},
"quote": {
"description": "Manage quotes (add, remove, restore, fetch, info).",
"subcommands": {
"no subcommand": {
"desc": "Fetch a random quote."
},
"examples": [
"quote : Fetch a random quote",
"quote 3 : Fetch quote #3",
"quote search ookamikuntv : Search for quote containing 'ookamikuntv'",
"quote add This is my new quote : Add a new quote",
"quote remove 3 : Remove quote #3",
"quote restore 3 : Restores quote #3",
"quote info 3 : Gets info about quote #3"
]
},
"ping": {
"description": "Check my uptime.",
"subcommands": {
"stat": {}
"[quote_number]": {
"args": "[quote_number]",
"desc": "Fetch a specific quote by ID."
},
"examples": [
"ping"
]
},
"howl": {
"description": "Attempt a howl, measured 0-100%.\n(*Adventure Command*)",
"subcommands": {
"no subcommand": {
"desc": "Attempt a howl"
},
"stat/stats": {
"args": "[username]",
"desc": "Get statistics about another user. Can be empty for self, or 'all' for everyone."
}
"search": {
"args": "[keywords]",
"desc": "Search for a quote with keywords."
},
"examples": [
"howl : Perform a normal howl attempt.",
"howl stat : Check your own howl statistics.",
"howl stats : same as above, just an alias.",
"howl stat [username] : Check someone else's statistics",
"howl stat all : Check the community statistics"
]
"last/latest/newest": {
"desc": "Returns the newest quote."
},
"add": {
"args": "[quote_text]",
"desc": "Adds a new quote."
},
"remove": {
"args": "[quote_number]",
"desc": "Removes the specified quote by ID."
},
"restore": {
"args": "[quote_number]",
"desc": "Restores the specified quote by ID."
},
"info": {
"args": "[quote_number]",
"desc": "Retrieves info about the specified quote."
}
},
"hi": {
"description": "Hello there.",
"subcommands": {},
"examples": [
"hi"
]
"examples": [
"quote : Fetch a random quote",
"quote 3 : Fetch quote #3",
"quote search ookamikuntv : Search for quote containing 'ookamikuntv'",
"quote add This is my new quote : Add a new quote",
"quote remove 3 : Remove quote #3",
"quote restore 3 : Restores quote #3",
"quote info 3 : Gets info about quote #3"
]
},
"ping": {
"description": "Check my uptime.",
"subcommands": {
"stat": {}
},
"greet": {
"description": "Make me greet you to Discord!",
"subcommands": {},
"examples": [
"greet"
]
"examples": [
"ping"
]
},
"customvc": {
"description": "Manage custom voice channels.",
"subcommands": {
"name": {
"args": "<new_name>",
"desc": "Rename your custom voice channel."
},
"claim": {
"desc": "Claim ownership of the channel if the owner has left."
},
"lock": {
"desc": "Lock your voice channel to prevent others from joining."
},
"allow": {
"args": "<user>",
"desc": "Allow a specific user to join your locked channel."
},
"deny": {
"args": "<user>",
"desc": "Deny a specific user from joining your channel."
},
"unlock": {
"desc": "Unlock your voice channel."
},
"users": {
"args": "<count>",
"desc": "Set a user limit for your voice channel."
},
"bitrate": {
"args": "<kbps>",
"desc": "Set the bitrate of your voice channel (8-128 kbps)."
},
"status": {
"args": "<status_str>",
"desc": "Set a custom status for your voice channel. (TODO)"
},
"op": {
"args": "<user>",
"desc": "Grant co-ownership of your channel to another user."
},
"settings": {
"desc": "Show current settings of your custom voice channel."
}
},
"reload": {
"description": "Reload Discord commands dynamically. TODO.",
"subcommands": {},
"examples": [
"reload"
]
}
"examples": [
"customvc : Show help for custom VC commands",
"customvc name My New VC : Rename the VC",
"customvc claim : Claim ownership if the owner left",
"customvc lock : Lock the channel",
"customvc allow @User : Allow a user to join",
"customvc deny @User : Deny a user from joining",
"customvc unlock : Unlock the channel",
"customvc users 5 : Set user limit to 5",
"customvc bitrate 64 : Set bitrate to 64 kbps",
"customvc status AFK Zone : Set custom status",
"customvc op @User : Grant co-ownership",
"customvc settings : View VC settings"
]
},
"howl": {
"description": "Attempt a howl, measured 0-100%.\n(*Adventure Command*)",
"subcommands": {
"no subcommand": {
"desc": "Attempt a howl"
},
"stat/stats": {
"args": "[username]",
"desc": "Get statistics about another user. Can be empty for self, or 'all' for everyone."
}
},
"examples": [
"howl : Perform a normal howl attempt.",
"howl stat : Check your own howl statistics.",
"howl stats : same as above, just an alias.",
"howl stat [username] : Check someone else's statistics",
"howl stat all : Check the community statistics"
]
},
"hi": {
"description": "Hello there.",
"subcommands": {},
"examples": [
"hi"
]
},
"greet": {
"description": "Make me greet you to Discord!",
"subcommands": {},
"examples": [
"greet"
]
},
"reload": {
"description": "Reload Discord commands dynamically. TODO.",
"subcommands": {},
"examples": [
"reload"
]
}
}
}

30
globals.py
View File

@ -4,6 +4,7 @@ import sys
import traceback
import discord
import inspect
import os
# Store the start time globally.
_bot_start_time = time.time()
@ -52,6 +53,35 @@ def load_config_file():
# Load configuration file
config_data = load_config_file()
def load_settings_file(file: str):
"""
Load a settings file from the settings directory.
Args:
file (str): The name of the settings file, with or without the .json extension.
Returns:
dict: The configuration data loaded from the specified settings file.
"""
SETTINGS_PATH = "settings"
# Ensure the file has a .json extension
if not file.endswith(".json"):
file += ".json"
file_path = os.path.join(SETTINGS_PATH, file)
if not os.path.exists(file_path):
log(f"Unable to read the settings file {file}!", "FATAL")
try:
with open(file_path, "r", encoding="utf-8") as f:
config_data = json.load(f)
return config_data
except json.JSONDecodeError as e:
log(f"Error parsing {file}: {e}", "FATAL")
###############################
# Simple Logging System
###############################

121
modules/db.py
View File

@ -3,6 +3,7 @@ import os
import re
import time, datetime
import sqlite3
import uuid
import globals
@ -514,76 +515,55 @@ def lookup_user(db_conn, identifier: str, identifier_type: str, target_identifie
return user_data
def ensure_chatlog_table(db_conn):
"""
Checks if 'chat_log' table exists. If not, creates it.
The table layout:
MESSAGE_ID (PK, auto increment)
UUID (references users.UUID, if you want a foreign key, see note below)
MESSAGE_CONTENT (text)
PLATFORM (string, e.g. 'twitch' or discord server name)
CHANNEL (the twitch channel or discord channel name)
DATETIME (defaults to current timestamp)
ATTACHMENTS (text; store hyperlink(s) or empty)
For maximum compatibility, we won't enforce the foreign key at the DB level,
but you can add it if you want.
Ensures the 'chat_log' table exists, updating the schema to use a UUID primary key
and an additional column for platform-specific message IDs.
"""
is_sqlite = "sqlite3" in str(type(db_conn)).lower()
# 1) Check if table exists
if is_sqlite:
check_sql = """
SELECT name
FROM sqlite_master
WHERE type='table'
AND name='chat_log'
"""
else:
check_sql = """
SELECT table_name
FROM information_schema.tables
WHERE table_name = 'chat_log'
AND table_schema = DATABASE()
"""
# Check if table exists
check_sql = """
SELECT name FROM sqlite_master WHERE type='table' AND name='chat_log'
""" if is_sqlite else """
SELECT table_name FROM information_schema.tables
WHERE table_name = 'chat_log' AND table_schema = DATABASE()
"""
rows = run_db_operation(db_conn, "read", check_sql)
if rows and rows[0] and rows[0][0]:
globals.log("Table 'chat_log' already exists, skipping creation.", "DEBUG")
return
# 2) Table doesn't exist => create it
globals.log("Table 'chat_log' does not exist; creating now...")
# Table does not exist, create it
globals.log("Table 'chat_log' does not exist; creating now...", "INFO")
if is_sqlite:
create_sql = """
CREATE TABLE chat_log (
MESSAGE_ID INTEGER PRIMARY KEY AUTOINCREMENT,
UUID TEXT,
MESSAGE_CONTENT TEXT,
PLATFORM TEXT,
CHANNEL TEXT,
DATETIME TEXT DEFAULT CURRENT_TIMESTAMP,
ATTACHMENTS TEXT,
FOREIGN KEY (UUID) REFERENCES users(UUID)
)
"""
else:
create_sql = """
CREATE TABLE chat_log (
MESSAGE_ID INT PRIMARY KEY AUTO_INCREMENT,
UUID VARCHAR(36),
MESSAGE_CONTENT TEXT,
PLATFORM VARCHAR(100),
CHANNEL VARCHAR(100),
DATETIME DATETIME DEFAULT CURRENT_TIMESTAMP,
ATTACHMENTS TEXT,
FOREIGN KEY (UUID) REFERENCES users(UUID) ON DELETE SET NULL
)
"""
create_sql = """
CREATE TABLE chat_log (
UUID TEXT PRIMARY KEY,
PLATFORM_MESSAGE_ID TEXT DEFAULT NULL,
USER_UUID TEXT,
MESSAGE_CONTENT TEXT,
PLATFORM TEXT,
CHANNEL TEXT,
DATETIME TEXT DEFAULT CURRENT_TIMESTAMP,
ATTACHMENTS TEXT,
FOREIGN KEY (USER_UUID) REFERENCES users(UUID)
)
""" if is_sqlite else """
CREATE TABLE chat_log (
UUID VARCHAR(36) PRIMARY KEY,
PLATFORM_MESSAGE_ID VARCHAR(100) DEFAULT NULL,
USER_UUID VARCHAR(36),
MESSAGE_CONTENT TEXT,
PLATFORM VARCHAR(100),
CHANNEL VARCHAR(100),
DATETIME DATETIME DEFAULT CURRENT_TIMESTAMP,
ATTACHMENTS TEXT,
FOREIGN KEY (USER_UUID) REFERENCES users(UUID) ON DELETE SET NULL
)
"""
result = run_db_operation(db_conn, "write", create_sql)
if result is None:
@ -594,17 +574,30 @@ def ensure_chatlog_table(db_conn):
globals.log("Successfully created table 'chat_log'.", "INFO")
def log_message(db_conn, identifier, identifier_type, message_content, platform, channel, attachments=None):
def log_message(db_conn, identifier, identifier_type, message_content, platform, channel, attachments=None, platform_message_id=None):
"""
Logs a message in 'chat_log' with UUID fetched using the new Platform_Mapping structure.
Logs a message in 'chat_log' with UUID fetched using the Platform_Mapping structure.
- Uses a UUID as the primary key for uniqueness across platforms.
- Stores platform-specific message IDs when provided.
- Logs a warning if a message ID is expected but not provided.
"""
# Get UUID using the updated lookup_user
# Get UUID using lookup_user
user_data = lookup_user(db_conn, identifier, identifier_type)
if not user_data:
globals.log(f"User not found for {identifier_type}='{identifier}'", "WARNING")
return
user_uuid = user_data["UUID"]
message_uuid = str(uuid.uuid4()) # Generate a new UUID for the entry
# Determine if a message ID is required for this platform
requires_message_id = platform.startswith("discord") or platform == "twitch"
if requires_message_id and not platform_message_id:
globals.log(f"Warning: Platform '{platform}' usually requires a message ID, but none was provided.", "WARNING")
if attachments is None or not "https://" in attachments:
attachments = ""
@ -612,17 +605,19 @@ def log_message(db_conn, identifier, identifier_type, message_content, platform,
insert_sql = """
INSERT INTO chat_log (
UUID,
PLATFORM_MESSAGE_ID,
USER_UUID,
MESSAGE_CONTENT,
PLATFORM,
CHANNEL,
ATTACHMENTS
) VALUES (?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?)
"""
params = (user_uuid, message_content, platform, channel, attachments)
params = (message_uuid, platform_message_id, user_uuid, message_content, platform, channel, attachments)
rowcount = run_db_operation(db_conn, "write", insert_sql, params)
if rowcount and rowcount > 0:
globals.log(f"Logged message for UUID={user_uuid}.", "DEBUG")
globals.log(f"Logged message for UUID={user_uuid} with Message UUID={message_uuid}.", "DEBUG")
else:
globals.log("Failed to log message in 'chat_log'.", "ERROR")

12
settings/discord_guilds_config.json
View File

@ -0,0 +1,12 @@
{
"896713616089309184": {
"customvc_settings": {
"lobby_vc_id": 1345509388651069583,
"customvc_category_id": 1337034437136879628,
"vc_creation_user_cooldown": 5,
"vc_creation_global_per_min": 2,
"empty_vc_autodelete_delay": 10,
"customvc_max_limit": 9
}
}
}