Merge pull request 'feat: Refactored all cogs implementation' (#4) from feature/refactorization into main
Some checks failed
CD / Build and push Docker image (push) Has been cancelled
CD / Deploy to VPS (push) Has been cancelled
CI / Run tests (push) Has been cancelled

Reviewed-on: #4
This commit was merged in pull request #4.
This commit is contained in:
2026-09-12 21:18:15 +00:00
10 changed files with 451 additions and 123 deletions

42
cogs/__init__.py Normal file
View 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)

View File

@@ -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:

View File

@@ -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
View 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

View File

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

View File

@@ -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
View File

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

View File

@@ -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:

View File

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

View File

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