feat: Refactored all cogs implementation
Some checks failed
CI / Run tests (pull_request) Has been cancelled
Some checks failed
CI / Run tests (pull_request) Has been cancelled
This commit is contained in:
42
cogs/__init__.py
Normal file
42
cogs/__init__.py
Normal file
@@ -0,0 +1,42 @@
|
|||||||
|
import discord, json, csv
|
||||||
|
from discord.ext import commands
|
||||||
|
from typing import Union, Any, Optional, Callable
|
||||||
|
from setup.config import Config
|
||||||
|
|
||||||
|
class BaseCog(commands.Cog):
|
||||||
|
def __init__(self,bot:discord.Bot):
|
||||||
|
self.bot = bot
|
||||||
|
|
||||||
|
class ConfigurableCog(BaseCog):
|
||||||
|
def __init__(self, bot: discord.Bot):
|
||||||
|
super().__init__(bot)
|
||||||
|
|
||||||
|
|
||||||
|
class FeatureCog(BaseCog):
|
||||||
|
"""Base Cog that provides per-guild feature flag checks.
|
||||||
|
|
||||||
|
Subclasses should set `feature_name` or pass it to the constructor.
|
||||||
|
This class reads the feature flags from `Config.GUILD_CONFIGS[guild_id]["features"]`.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, bot: discord.Bot, feature_name: Optional[str] = None):
|
||||||
|
super().__init__(bot)
|
||||||
|
self.feature_name = feature_name
|
||||||
|
|
||||||
|
def is_enabled(self, guild_id: int) -> bool:
|
||||||
|
if not self.feature_name:
|
||||||
|
return False
|
||||||
|
guild_cfg = Config.GUILD_CONFIGS.get(guild_id)
|
||||||
|
if not guild_cfg:
|
||||||
|
return False
|
||||||
|
features = guild_cfg.get("features", {})
|
||||||
|
return bool(features.get(self.feature_name, False))
|
||||||
|
|
||||||
|
def require_enabled_check(self) -> Callable[[commands.Context], bool]:
|
||||||
|
def predicate(ctx: commands.Context) -> bool:
|
||||||
|
if not ctx.guild:
|
||||||
|
return False
|
||||||
|
return self.is_enabled(ctx.guild.id)
|
||||||
|
|
||||||
|
return commands.check(predicate)
|
||||||
|
|
||||||
13
cogs/cs.py
13
cogs/cs.py
@@ -7,6 +7,7 @@ from discord.ext import commands, tasks, bridge
|
|||||||
from setup.logger import LOGGER
|
from setup.logger import LOGGER
|
||||||
from setup.config import Config
|
from setup.config import Config
|
||||||
from models.cs_models import *
|
from models.cs_models import *
|
||||||
|
from cogs import FeatureCog
|
||||||
|
|
||||||
BASE_URL = "https://www.hltv.org"
|
BASE_URL = "https://www.hltv.org"
|
||||||
|
|
||||||
@@ -24,11 +25,19 @@ OLD_MATCH_DURATION = 3600 * 24 * 2
|
|||||||
EMOJI_CREATE_EVENT = "✅" # Create guild event for the match
|
EMOJI_CREATE_EVENT = "✅" # Create guild event for the match
|
||||||
EMOJI_REMOVE_EVENT = "❌" # Remove guild event for the match
|
EMOJI_REMOVE_EVENT = "❌" # Remove guild event for the match
|
||||||
|
|
||||||
class CSCog(commands.Cog):
|
class CSCog(FeatureCog):
|
||||||
def __init__(self, bot: discord.Bot):
|
def __init__(self, bot: discord.Bot):
|
||||||
self.bot = bot
|
super().__init__(bot, feature_name="cs")
|
||||||
self.matches: list[Match] = []
|
self.matches: list[Match] = []
|
||||||
self.message_match_map = {} # Maps message IDs to Match objects
|
self.message_match_map = {} # Maps message IDs to Match objects
|
||||||
|
|
||||||
|
# Start scraping loop only if at least one guild enabled CS feature
|
||||||
|
try:
|
||||||
|
enabled_any = any(cfg.get("features", {}).get("cs") for cfg in Config.GUILD_CONFIGS.values())
|
||||||
|
except Exception:
|
||||||
|
enabled_any = False
|
||||||
|
|
||||||
|
if enabled_any:
|
||||||
self.get_matches.start()
|
self.get_matches.start()
|
||||||
|
|
||||||
def is_up_to_date(self) -> bool:
|
def is_up_to_date(self) -> bool:
|
||||||
|
|||||||
@@ -1,32 +1,76 @@
|
|||||||
import discord, json, os, random, re
|
import discord, random, re
|
||||||
import collections
|
|
||||||
from discord.ext import commands
|
from discord.ext import commands
|
||||||
from setup.logger import LOGGER
|
from setup.logger import LOGGER
|
||||||
from setup.config import Config
|
from setup.config import Config
|
||||||
|
from cogs import FeatureCog
|
||||||
|
|
||||||
class InsultsCog(commands.Cog):
|
class InsultsCog(FeatureCog):
|
||||||
def __init__(self, bot: discord.Bot):
|
def __init__(self, bot: discord.Bot):
|
||||||
self.bot = bot
|
super().__init__(bot, feature_name="insults")
|
||||||
|
LOGGER.debug("InsultsCog initialized")
|
||||||
|
|
||||||
with open(os.path.join(Config.CONFIG_PATH, "insults.txt")) as f:
|
def reload_men_list(self):
|
||||||
self.insults = [l.strip() for l in f.readlines()]
|
self.men: list[discord.Member] = []
|
||||||
|
for guild in self.bot.guilds:
|
||||||
|
for role in guild.roles:
|
||||||
|
if role.name and re.match(r'^([hH]im|[hH]e).*$', role.name):
|
||||||
|
self.men.extend(role.members)
|
||||||
|
# dedupe by id
|
||||||
|
self.men = list({m.id: m for m in self.men}.values())
|
||||||
|
|
||||||
@commands.Cog.listener()
|
@commands.Cog.listener()
|
||||||
async def on_message(self, message: discord.Message):
|
async def on_message(self, message: discord.Message):
|
||||||
if len(message.content.split()) > 3:
|
if not message.guild or message.author.bot:
|
||||||
|
return
|
||||||
|
|
||||||
|
# only run if feature enabled for this guild
|
||||||
|
if not self.is_enabled(message.guild.id):
|
||||||
|
return
|
||||||
|
|
||||||
|
if len(message.content.split()) <= 3:
|
||||||
|
return
|
||||||
|
|
||||||
author = message.author
|
author = message.author
|
||||||
|
if not hasattr(self, 'men') or not self.men:
|
||||||
|
self.reload_men_list()
|
||||||
|
|
||||||
if any(author.id == u.id for u in self.men):
|
if any(author.id == u.id for u in self.men):
|
||||||
r = int(random.random()*100)
|
guild_cfg = Config.GUILD_CONFIGS.get(message.guild.id, {})
|
||||||
print(r)
|
insults_cfg = guild_cfg.get("configs", {}).get("insults", {})
|
||||||
|
safeguard_patterns = insults_cfg.get("safeguard_patterns", [])
|
||||||
|
safeguard_match = any(re.match(pattern, message.content) for pattern in safeguard_patterns)
|
||||||
|
if safeguard_match:
|
||||||
|
return
|
||||||
|
|
||||||
|
r = random.random() * 100
|
||||||
|
exceptions = insults_cfg.get("exceptions", [])
|
||||||
|
filtered = [ex for ex in exceptions if ex.get("user_id") == author.id]
|
||||||
|
if filtered:
|
||||||
|
try:
|
||||||
|
r *= float(filtered[0].get("factor", 1))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
if r < 5:
|
if r < 5:
|
||||||
insult = random.choice(self.insults)
|
insults = insults_cfg.get("insults", [])
|
||||||
|
if insults:
|
||||||
|
insult = random.choice(insults)
|
||||||
await message.reply(insult)
|
await message.reply(insult)
|
||||||
|
|
||||||
@commands.Cog.listener()
|
@commands.Cog.listener()
|
||||||
async def on_ready(self):
|
async def on_ready(self):
|
||||||
self.men:list[discord.Member] = []
|
self.reload_men_list()
|
||||||
for guild in self.bot.guilds:
|
|
||||||
for role in guild.roles:
|
@commands.Cog.listener()
|
||||||
if re.match(r'^([hH]im|[hH]e).*$',role.name):
|
async def on_member_update(self, before: discord.Member, after: discord.Member):
|
||||||
self.men.extend(role.members)
|
before_roles = set(before.roles)
|
||||||
self.men = list(set(self.men))
|
after_roles = set(after.roles)
|
||||||
|
removed = before_roles.difference(after_roles)
|
||||||
|
added = after_roles.difference(before_roles)
|
||||||
|
role = None
|
||||||
|
if removed:
|
||||||
|
role = list(removed).pop()
|
||||||
|
elif added:
|
||||||
|
role = list(added).pop()
|
||||||
|
|
||||||
|
if role and re.match(r'^([hH]im|[hH]e).*$', role.name):
|
||||||
|
self.reload_men_list()
|
||||||
55
cogs/intern.py
Normal file
55
cogs/intern.py
Normal file
@@ -0,0 +1,55 @@
|
|||||||
|
import random
|
||||||
|
from discord.ext import commands
|
||||||
|
|
||||||
|
|
||||||
|
class InternCog(commands.Cog):
|
||||||
|
def __init__(self, bot=None):
|
||||||
|
self.bot = bot
|
||||||
|
# `data` expected to be a list of dicts with 'guild_id' and 'mentor_id'
|
||||||
|
self.data = []
|
||||||
|
self.connected_guild_ids = set()
|
||||||
|
self.in_voice_channel = False
|
||||||
|
|
||||||
|
async def on_voice_state_update(self, member, before, after):
|
||||||
|
# determine guild context
|
||||||
|
guild = None
|
||||||
|
if after and getattr(after, 'channel', None):
|
||||||
|
guild = after.channel.guild
|
||||||
|
elif before and getattr(before, 'channel', None):
|
||||||
|
guild = before.channel.guild
|
||||||
|
if not guild:
|
||||||
|
return
|
||||||
|
|
||||||
|
# find matching mentor entry for this guild
|
||||||
|
entry = None
|
||||||
|
for e in getattr(self, 'data', []):
|
||||||
|
if e.get('guild_id') == getattr(guild, 'id', None) and e.get('mentor_id') == getattr(member, 'id', None):
|
||||||
|
entry = e
|
||||||
|
break
|
||||||
|
if not entry:
|
||||||
|
return
|
||||||
|
|
||||||
|
# find existing voice client for this guild (if any)
|
||||||
|
voice_client = None
|
||||||
|
for vc in getattr(self.bot, 'voice_clients', []):
|
||||||
|
if getattr(vc, 'guild', None) and getattr(vc.guild, 'id', None) == getattr(guild, 'id', None):
|
||||||
|
voice_client = vc
|
||||||
|
break
|
||||||
|
|
||||||
|
# mentor joined a channel (initial join)
|
||||||
|
if (not before or not getattr(before, 'channel', None)) and after and getattr(after, 'channel', None):
|
||||||
|
# if bot is not connected, connect with 25% chance
|
||||||
|
if not voice_client:
|
||||||
|
if random.random() < 0.25:
|
||||||
|
await after.channel.connect()
|
||||||
|
return
|
||||||
|
|
||||||
|
# mentor moved channels while bot is connected -> move bot
|
||||||
|
if voice_client and after and getattr(after, 'channel', None):
|
||||||
|
await voice_client.move_to(after.channel)
|
||||||
|
return
|
||||||
|
|
||||||
|
# mentor left channel -> disconnect bot if connected
|
||||||
|
if voice_client and (not after or not getattr(after, 'channel', None)):
|
||||||
|
await voice_client.disconnect(force=True)
|
||||||
|
return
|
||||||
157
cogs/roles.py
157
cogs/roles.py
@@ -1,47 +1,64 @@
|
|||||||
import discord, os, json
|
import discord
|
||||||
|
from cogs import FeatureCog
|
||||||
from discord import option
|
from discord import option
|
||||||
from discord.ext import commands, bridge
|
from discord.ext import commands, bridge
|
||||||
from setup.config import Config
|
from setup.config import Config
|
||||||
from setup.logger import LOGGER
|
from setup.logger import LOGGER
|
||||||
|
|
||||||
CONFIG_FILE = os.path.join(Config.CONFIG_PATH,"roles.json")
|
|
||||||
|
|
||||||
class RolesCog(commands.Cog):
|
class RolesCog(FeatureCog):
|
||||||
def __init__(self, bot: discord.Bot):
|
def __init__(self, bot: discord.Bot):
|
||||||
self.bot = bot
|
super().__init__(bot, feature_name="roles")
|
||||||
self.load_config()
|
|
||||||
|
|
||||||
def load_config(self):
|
|
||||||
with open(CONFIG_FILE) as file:
|
|
||||||
self.config = json.load(file)
|
|
||||||
|
|
||||||
def save_config(self):
|
|
||||||
with open(CONFIG_FILE,"w") as file:
|
|
||||||
json.dump(self.config,file)
|
|
||||||
|
|
||||||
@bridge.bridge_command()
|
@bridge.bridge_command()
|
||||||
async def roles(self, ctx: discord.ApplicationContext, category: bridge.BridgeOption(name="category", description="The category to manipulate", input_type=str, required=True)):
|
async def roles(self, ctx: discord.ApplicationContext, category: bridge.BridgeOption(name="category", description="The category to manipulate", input_type=str, required=True)):
|
||||||
filtered = [config_category for config_category in self.config if config_category["category_id"] == category]
|
if not ctx.guild:
|
||||||
if len(filtered):
|
await ctx.respond("This command must be used in a guild.", ephemeral=True)
|
||||||
|
return
|
||||||
|
|
||||||
|
guild_id = ctx.guild.id
|
||||||
|
if not self.is_enabled(guild_id):
|
||||||
|
await ctx.respond("Roles feature is disabled for this guild.", ephemeral=True)
|
||||||
|
return
|
||||||
|
|
||||||
|
guild_cfg = Config.GUILD_CONFIGS.get(guild_id)
|
||||||
|
if not guild_cfg:
|
||||||
|
await ctx.respond("No configuration for this guild.", ephemeral=True)
|
||||||
|
return
|
||||||
|
|
||||||
|
categories = guild_cfg.get("configs", {}).get("roles", {}).get("categories", [])
|
||||||
|
filtered = [c for c in categories if c.get("category_id") == category]
|
||||||
|
if not filtered:
|
||||||
|
await ctx.respond("Category not found.", ephemeral=True)
|
||||||
|
return
|
||||||
|
|
||||||
await ctx.defer()
|
await ctx.defer()
|
||||||
config_category = filtered.pop()
|
config_category = filtered.pop()
|
||||||
embed = discord.Embed(
|
embed = discord.Embed(
|
||||||
title=config_category["category_name"],
|
title=config_category.get("category_name", ""),
|
||||||
description=config_category["message_header"],
|
description=config_category.get("message_header", ""),
|
||||||
color=discord.Color.blue()
|
color=discord.Color.blue()
|
||||||
)
|
)
|
||||||
roles = [
|
|
||||||
self.bot.get_guild(897157935006900254).get_role(role_config["role_id"])
|
roles_list = [
|
||||||
for role_config in config_category["roles"]
|
ctx.guild.get_role(role_config.get("role_id"))
|
||||||
|
for role_config in config_category.get("roles", [])
|
||||||
]
|
]
|
||||||
for role_instance,role_config in zip(roles,config_category["roles"]):
|
for role_instance, role_config in zip(roles_list, config_category.get("roles", [])):
|
||||||
embed.add_field(name=role_config["reaction"],value=role_instance.mention)
|
mention = role_instance.mention if role_instance else "(role missing)"
|
||||||
|
embed.add_field(name=role_config.get("reaction", ""), value=mention)
|
||||||
|
|
||||||
message = await ctx.send(embed=embed)
|
message = await ctx.send(embed=embed)
|
||||||
if message:
|
if message:
|
||||||
config_category["message_id"] = message.id
|
config_category["message_id"] = message.id
|
||||||
for config_role in config_category["roles"]:
|
# persist the updated guild config
|
||||||
await message.add_reaction(config_role["reaction"])
|
Config.save_guild_config(guild_id)
|
||||||
self.save_config()
|
|
||||||
|
for config_role in config_category.get("roles", []):
|
||||||
|
reaction = config_role.get("reaction")
|
||||||
|
if reaction:
|
||||||
|
await message.add_reaction(reaction)
|
||||||
|
|
||||||
await ctx.followup.send("Role message generated", ephemeral=True)
|
await ctx.followup.send("Role message generated", ephemeral=True)
|
||||||
|
|
||||||
@commands.Cog.listener()
|
@commands.Cog.listener()
|
||||||
@@ -49,24 +66,41 @@ class RolesCog(commands.Cog):
|
|||||||
if payload.user_id == self.bot.user.id:
|
if payload.user_id == self.bot.user.id:
|
||||||
return
|
return
|
||||||
|
|
||||||
message_id = payload.message_id
|
guild_id = payload.guild_id
|
||||||
filtered = [config_category for config_category in self.config if config_category["message_id"] == message_id]
|
guild_cfg = Config.GUILD_CONFIGS.get(guild_id)
|
||||||
if len(filtered):
|
if not guild_cfg:
|
||||||
config_category = filtered.pop()
|
return
|
||||||
role_ids = [config_role["role_id"] for config_role in config_category["roles"] if config_role["reaction"] == payload.emoji.name]
|
if not self.is_enabled(guild_id):
|
||||||
if len(role_ids):
|
return
|
||||||
role_id = role_ids.pop()
|
|
||||||
|
|
||||||
guild = self.bot.get_guild(payload.guild_id)
|
categories = guild_cfg.get("configs", {}).get("roles", {}).get("categories", [])
|
||||||
if guild:
|
filtered = [c for c in categories if c.get("message_id") == payload.message_id]
|
||||||
|
if not filtered:
|
||||||
|
return
|
||||||
|
|
||||||
|
config_category = filtered.pop()
|
||||||
|
role_ids = [config_role.get("role_id") for config_role in config_category.get("roles", []) if config_role.get("reaction") == payload.emoji.name]
|
||||||
|
if not role_ids:
|
||||||
|
return
|
||||||
|
|
||||||
|
role_id = role_ids.pop()
|
||||||
|
guild = self.bot.get_guild(guild_id)
|
||||||
|
if not guild:
|
||||||
|
return
|
||||||
role = guild.get_role(role_id)
|
role = guild.get_role(role_id)
|
||||||
if role:
|
if not role:
|
||||||
|
return
|
||||||
|
|
||||||
user = guild.get_member(payload.user_id)
|
user = guild.get_member(payload.user_id)
|
||||||
if user:
|
if not user:
|
||||||
if "parent_role_id" in config_category.keys():
|
return
|
||||||
parent_role = guild.get_role(config_category["parent_role_id"])
|
|
||||||
|
if "parent_role_id" in config_category:
|
||||||
|
parent_role = guild.get_role(config_category.get("parent_role_id"))
|
||||||
|
if parent_role:
|
||||||
await user.add_roles(parent_role, role)
|
await user.add_roles(parent_role, role)
|
||||||
else:
|
return
|
||||||
|
|
||||||
await user.add_roles(role)
|
await user.add_roles(role)
|
||||||
|
|
||||||
@commands.Cog.listener()
|
@commands.Cog.listener()
|
||||||
@@ -74,22 +108,39 @@ class RolesCog(commands.Cog):
|
|||||||
if payload.user_id == self.bot.user.id:
|
if payload.user_id == self.bot.user.id:
|
||||||
return
|
return
|
||||||
|
|
||||||
message_id = payload.message_id
|
guild_id = payload.guild_id
|
||||||
filtered = [config_category for config_category in self.config if config_category["message_id"] == message_id]
|
guild_cfg = Config.GUILD_CONFIGS.get(guild_id)
|
||||||
if len(filtered):
|
if not guild_cfg:
|
||||||
config_category = filtered.pop()
|
return
|
||||||
role_ids = [config_role["role_id"] for config_role in config_category["roles"] if config_role["reaction"] == payload.emoji.name]
|
if not self.is_enabled(guild_id):
|
||||||
if len(role_ids):
|
return
|
||||||
role_id = role_ids.pop()
|
|
||||||
|
|
||||||
guild = self.bot.get_guild(payload.guild_id)
|
categories = guild_cfg.get("configs", {}).get("roles", {}).get("categories", [])
|
||||||
if guild:
|
filtered = [c for c in categories if c.get("message_id") == payload.message_id]
|
||||||
|
if not filtered:
|
||||||
|
return
|
||||||
|
|
||||||
|
config_category = filtered.pop()
|
||||||
|
role_ids = [config_role.get("role_id") for config_role in config_category.get("roles", []) if config_role.get("reaction") == payload.emoji.name]
|
||||||
|
if not role_ids:
|
||||||
|
return
|
||||||
|
|
||||||
|
role_id = role_ids.pop()
|
||||||
|
guild = self.bot.get_guild(guild_id)
|
||||||
|
if not guild:
|
||||||
|
return
|
||||||
role = guild.get_role(role_id)
|
role = guild.get_role(role_id)
|
||||||
if role:
|
if not role:
|
||||||
|
return
|
||||||
|
|
||||||
user = guild.get_member(payload.user_id)
|
user = guild.get_member(payload.user_id)
|
||||||
if user:
|
if not user:
|
||||||
if "parent_role_id" in config_category.keys():
|
return
|
||||||
parent_role = guild.get_role(config_category["parent_role_id"])
|
|
||||||
|
if "parent_role_id" in config_category:
|
||||||
|
parent_role = guild.get_role(config_category.get("parent_role_id"))
|
||||||
|
if parent_role:
|
||||||
await user.remove_roles(parent_role, role)
|
await user.remove_roles(parent_role, role)
|
||||||
else:
|
return
|
||||||
|
|
||||||
await user.remove_roles(role)
|
await user.remove_roles(role)
|
||||||
@@ -5,11 +5,12 @@ import random
|
|||||||
import discord
|
import discord
|
||||||
from discord import option
|
from discord import option
|
||||||
from discord.ext import commands
|
from discord.ext import commands
|
||||||
|
from cogs import FeatureCog
|
||||||
|
|
||||||
|
|
||||||
class RouletteCog(commands.Cog):
|
class RouletteCog(FeatureCog):
|
||||||
def __init__(self, bot: discord.Bot):
|
def __init__(self, bot: discord.Bot):
|
||||||
self.bot = bot
|
super().__init__(bot, feature_name="roulette")
|
||||||
self.cooldown = 60 * 3
|
self.cooldown = 60 * 3
|
||||||
self.active_games = {}
|
self.active_games = {}
|
||||||
|
|
||||||
@@ -155,6 +156,10 @@ class RouletteCog(commands.Cog):
|
|||||||
ctx: discord.ApplicationContext,
|
ctx: discord.ApplicationContext,
|
||||||
target: discord.Member,
|
target: discord.Member,
|
||||||
):
|
):
|
||||||
|
# ensure feature enabled for this guild
|
||||||
|
if not ctx.guild or not self.is_enabled(ctx.guild.id):
|
||||||
|
await ctx.respond("Roulette feature is disabled for this guild.", ephemeral=True)
|
||||||
|
return
|
||||||
author = ctx.author
|
author = ctx.author
|
||||||
self_test = author.id == target.id
|
self_test = author.id == target.id
|
||||||
|
|
||||||
|
|||||||
55
main.py
55
main.py
@@ -1,29 +1,54 @@
|
|||||||
import os, discord, json, sys, random
|
import os
|
||||||
|
import sys
|
||||||
|
import discord
|
||||||
from argparse import ArgumentParser
|
from argparse import ArgumentParser
|
||||||
from datetime import datetime
|
|
||||||
from asyncio import create_task
|
# Parse minimal CLI args early so we can configure logging before imports that
|
||||||
from discord import option
|
# initialize loggers. Keep only `--log-level`; other args were unused.
|
||||||
from discord.ext import tasks, commands
|
PARSER = ArgumentParser(description='BotMafieux for Discord.')
|
||||||
|
PARSER.add_argument('--log-level', type=str, help='Logging level (DEBUG, INFO, WARNING, ERROR)')
|
||||||
|
EXEC_ARGS = PARSER.parse_args()
|
||||||
|
|
||||||
|
if EXEC_ARGS.log_level:
|
||||||
|
os.environ['LOG_LEVEL'] = str(EXEC_ARGS.log_level)
|
||||||
|
|
||||||
|
|
||||||
from setup.bot import bot, reload_feature, register_cogs
|
from setup.bot import bot, reload_feature, register_cogs
|
||||||
from setup.logger import LOGGER
|
from setup.logger import LOGGER
|
||||||
from setup.config import Config
|
from setup.config import Config
|
||||||
|
|
||||||
|
# Register cogs from configuration (no CLI activation)
|
||||||
PARSER = ArgumentParser(description='BotMafieux for Discord.')
|
register_cogs()
|
||||||
PARSER.add_argument('--guild_id',type=int,help='The guild ID.')
|
|
||||||
PARSER.add_argument('--channel_id',type=int,help='The channel ID.')
|
|
||||||
PARSER.add_argument('-r','--reload',action='store_true',help='Reload the bot.')
|
|
||||||
PARSER.add_argument('--cogs',nargs='+',default=None,help='Select which cogs to load (for example: --cogs cs insult).')
|
|
||||||
EXEC_ARGS = PARSER.parse_args()
|
|
||||||
selected_cogs = EXEC_ARGS.cogs if EXEC_ARGS.cogs is not None else Config.get_cogs()
|
|
||||||
register_cogs(selected_cogs)
|
|
||||||
|
|
||||||
@bot.event
|
@bot.event
|
||||||
async def on_ready():
|
async def on_ready():
|
||||||
if bot.user:
|
if bot.user:
|
||||||
LOGGER.info(f'{bot.user.name} has connected to Discord!')
|
LOGGER.info(f'{bot.user.name} has connected to Discord!')
|
||||||
LOGGER.debug(f"Guilds: {','.join([guild.name for guild in bot.guilds])}")
|
LOGGER.info(f"Guilds: {','.join([guild.name for guild in bot.guilds])}")
|
||||||
|
# Log configuration for each guild we are connected to
|
||||||
|
try:
|
||||||
|
for guild in bot.guilds:
|
||||||
|
cfg = Config.GUILD_CONFIGS.get(guild.id)
|
||||||
|
if cfg is not None:
|
||||||
|
LOGGER.info(f"Guild config for {guild.name} ({guild.id}): {cfg}")
|
||||||
|
else:
|
||||||
|
LOGGER.info(f"No guild config for {guild.name} ({guild.id})")
|
||||||
|
except Exception as e:
|
||||||
|
LOGGER.warning(f"Failed to log guild configs: {e}")
|
||||||
|
|
||||||
|
@bot.event
|
||||||
|
async def on_guild_join(guild: discord.Guild):
|
||||||
|
# Refresh configs and log the joined guild configuration
|
||||||
|
try:
|
||||||
|
Config.load_guilds_configs()
|
||||||
|
except Exception:
|
||||||
|
LOGGER.debug('Could not reload guild configs during on_guild_join.')
|
||||||
|
|
||||||
|
cfg = Config.GUILD_CONFIGS.get(guild.id)
|
||||||
|
if cfg is not None:
|
||||||
|
LOGGER.info(f"Joined guild {guild.name} ({guild.id}) with config: {cfg}")
|
||||||
|
else:
|
||||||
|
LOGGER.info(f"Joined guild {guild.name} ({guild.id}) with no configuration present.")
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
token = Config.TOKEN
|
token = Config.TOKEN
|
||||||
|
|||||||
32
setup/bot.py
32
setup/bot.py
@@ -5,7 +5,9 @@ from discord.ext import bridge
|
|||||||
from cogs.cs import CSCog
|
from cogs.cs import CSCog
|
||||||
from cogs.insult import InsultsCog
|
from cogs.insult import InsultsCog
|
||||||
from cogs.roulette import RouletteCog
|
from cogs.roulette import RouletteCog
|
||||||
|
from cogs.roles import RolesCog
|
||||||
from setup.logger import LOGGER
|
from setup.logger import LOGGER
|
||||||
|
from setup.config import Config
|
||||||
|
|
||||||
bot_intents = discord.Intents.default()
|
bot_intents = discord.Intents.default()
|
||||||
bot_intents.message_content = True
|
bot_intents.message_content = True
|
||||||
@@ -19,9 +21,26 @@ bot = bridge.Bot(command_prefix='$', intents=bot_intents)
|
|||||||
COG_REGISTRY = {
|
COG_REGISTRY = {
|
||||||
'cs': CSCog,
|
'cs': CSCog,
|
||||||
'insult': InsultsCog,
|
'insult': InsultsCog,
|
||||||
'roulette': RouletteCog
|
'roulette': RouletteCog,
|
||||||
|
'roles': RolesCog,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# Register a global admin slash command directly on the bot to avoid circular imports.
|
||||||
|
from discord import option as discord_option
|
||||||
|
|
||||||
|
|
||||||
|
@bot.slash_command(name="reload", description="Reload a feature (cog) by name, or 'all')")
|
||||||
|
@discord_option(name="feature", description="Feature to reload (e.g. cs, insult, roles, roulette, all)", input_type=str, required=True)
|
||||||
|
async def reload(ctx: discord.ApplicationContext, feature: str):
|
||||||
|
# Only allow administrators
|
||||||
|
if not ctx.author or not getattr(ctx.author, 'guild_permissions', None) or not ctx.author.guild_permissions.administrator:
|
||||||
|
await ctx.respond("You must be a server administrator to use this command.", ephemeral=True)
|
||||||
|
return
|
||||||
|
|
||||||
|
await ctx.defer()
|
||||||
|
reload_feature(feature)
|
||||||
|
await ctx.followup.send(f"Reload requested for feature: {feature}", ephemeral=True)
|
||||||
|
|
||||||
|
|
||||||
def _normalize_cog_names(cogs=None):
|
def _normalize_cog_names(cogs=None):
|
||||||
if cogs is None:
|
if cogs is None:
|
||||||
@@ -46,6 +65,12 @@ def _normalize_cog_names(cogs=None):
|
|||||||
|
|
||||||
|
|
||||||
def register_cogs(cogs=None):
|
def register_cogs(cogs=None):
|
||||||
|
# Ensure guild configs are loaded before instantiating cogs
|
||||||
|
try:
|
||||||
|
Config.load_guilds_configs()
|
||||||
|
except Exception:
|
||||||
|
LOGGER.debug('Could not reload guild configs during register_cogs.')
|
||||||
|
|
||||||
selected_cogs = _normalize_cog_names(cogs)
|
selected_cogs = _normalize_cog_names(cogs)
|
||||||
|
|
||||||
existing_cog_names = set(COG_REGISTRY.values())
|
existing_cog_names = set(COG_REGISTRY.values())
|
||||||
@@ -75,6 +100,11 @@ def reload_feature(feature: str):
|
|||||||
return
|
return
|
||||||
|
|
||||||
LOGGER.debug(f'Reloading {feature}...')
|
LOGGER.debug(f'Reloading {feature}...')
|
||||||
|
# Refresh guild configs before reloading a single feature
|
||||||
|
try:
|
||||||
|
Config.load_guilds_configs()
|
||||||
|
except Exception:
|
||||||
|
LOGGER.debug('Could not reload guild configs during reload_feature.')
|
||||||
target_cog_name = COG_REGISTRY[normalized_feature].__name__
|
target_cog_name = COG_REGISTRY[normalized_feature].__name__
|
||||||
for existing_name in list(bot.cogs.keys()):
|
for existing_name in list(bot.cogs.keys()):
|
||||||
if existing_name == target_cog_name:
|
if existing_name == target_cog_name:
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import os
|
import os, pathlib, json
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
|
from setup.logger import LOGGER
|
||||||
|
|
||||||
class Config:
|
class Config:
|
||||||
load_dotenv()
|
load_dotenv()
|
||||||
@@ -9,6 +9,10 @@ class Config:
|
|||||||
CONFIG_PATH = str(os.getenv("CONFIG_PATH"))
|
CONFIG_PATH = str(os.getenv("CONFIG_PATH"))
|
||||||
COGS = os.getenv("COGS", "")
|
COGS = os.getenv("COGS", "")
|
||||||
|
|
||||||
|
GUILDS_IDS:list[int] = []
|
||||||
|
|
||||||
|
GUILD_CONFIGS:dict[int,dict] = {}
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get_cogs(cls):
|
def get_cogs(cls):
|
||||||
raw_value = cls.COGS.strip()
|
raw_value = cls.COGS.strip()
|
||||||
@@ -16,3 +20,53 @@ class Config:
|
|||||||
return None
|
return None
|
||||||
return [item.strip().lower() for item in raw_value.split(",") if item.strip()]
|
return [item.strip().lower() for item in raw_value.split(",") if item.strip()]
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def load_guilds_configs(cls):
|
||||||
|
with os.scandir(cls.CONFIG_PATH) as d:
|
||||||
|
guilds_config_files = [f.path for f in d if f.is_file() and f.name.endswith(".json")]
|
||||||
|
for guild_config_file in guilds_config_files:
|
||||||
|
with open(guild_config_file) as f:
|
||||||
|
guild_config = json.load(f)
|
||||||
|
guild_id = guild_config["guild_id"]
|
||||||
|
cls.GUILDS_IDS.append(guild_id)
|
||||||
|
cls.GUILD_CONFIGS[guild_id] = {k:v for k,v in guild_config.items() if k != "guild_id"}
|
||||||
|
cls.GUILD_CONFIGS[guild_id]["config_path"] = guild_config_file
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def load_guild_config(cls,guild_id:int):
|
||||||
|
config_path = cls.GUILD_CONFIGS[guild_id]["config_path"]
|
||||||
|
with open(config_path) as f:
|
||||||
|
guild_config = json.load(f)
|
||||||
|
# Preserve config_path when re-loading
|
||||||
|
cls.GUILD_CONFIGS[guild_id] = {k: v for k, v in guild_config.items() if k != "guild_id"}
|
||||||
|
cls.GUILD_CONFIGS[guild_id]["config_path"] = config_path
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def save_guilds_configs(cls):
|
||||||
|
for guild_id, guild_config in cls.GUILD_CONFIGS.items():
|
||||||
|
config_path = guild_config.get("config_path")
|
||||||
|
if not config_path:
|
||||||
|
continue
|
||||||
|
with open(config_path, 'w') as file:
|
||||||
|
json_config = {k: v for k, v in guild_config.items() if k != "config_path"}
|
||||||
|
json_config["guild_id"] = guild_id
|
||||||
|
json.dump(json_config, file, ensure_ascii=False, indent=2)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def save_guild_config(cls,guild_id:int):
|
||||||
|
config = cls.GUILD_CONFIGS.get(guild_id)
|
||||||
|
if not config:
|
||||||
|
return
|
||||||
|
config_path = config.get("config_path")
|
||||||
|
if not config_path:
|
||||||
|
return
|
||||||
|
with open(config_path, 'w') as file:
|
||||||
|
json_config = {k: v for k, v in config.items() if k != "config_path"}
|
||||||
|
json_config["guild_id"] = guild_id
|
||||||
|
json.dump(json_config, file, ensure_ascii=False, indent=2)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def get_guild_config(cls,guild_id):
|
||||||
|
return cls.GUILD_CONFIGS[guild_id]
|
||||||
|
|
||||||
|
Config.load_guilds_configs()
|
||||||
@@ -1,8 +1,21 @@
|
|||||||
import logging
|
import logging
|
||||||
|
import os
|
||||||
|
|
||||||
logging.basicConfig(level=logging.INFO, format='[%(levelname)s] %(asctime)s: %(name)s: %(message)s')
|
# Determine log level from environment variable `LOG_LEVEL` (e.g. DEBUG, INFO)
|
||||||
|
raw_level = os.getenv("LOG_LEVEL", "INFO").upper()
|
||||||
|
level = logging.INFO
|
||||||
|
if raw_level.isdigit():
|
||||||
|
try:
|
||||||
|
level = int(raw_level)
|
||||||
|
except Exception:
|
||||||
|
level = logging.INFO
|
||||||
|
else:
|
||||||
|
level = getattr(logging, raw_level, logging.INFO)
|
||||||
|
|
||||||
|
logging.basicConfig(level=level, format='[%(levelname)s] %(asctime)s: %(name)s: %(message)s')
|
||||||
LOGGER = logging.getLogger('botmafieux')
|
LOGGER = logging.getLogger('botmafieux')
|
||||||
LOGGER.setLevel(logging.DEBUG)
|
LOGGER.setLevel(level)
|
||||||
|
# File handler (append)
|
||||||
handler = logging.FileHandler(filename='botmafieux.log', encoding='utf-8', mode='a')
|
handler = logging.FileHandler(filename='botmafieux.log', encoding='utf-8', mode='a')
|
||||||
handler.setFormatter(logging.Formatter('%(asctime)s:%(levelname)s:%(name)s: %(message)s'))
|
handler.setFormatter(logging.Formatter('%(asctime)s:%(levelname)s:%(name)s: %(message)s'))
|
||||||
LOGGER.addHandler(handler)
|
LOGGER.addHandler(handler)
|
||||||
Reference in New Issue
Block a user