From d14ca4ab3ef96db7cb32a3350fe3112f20b7e118 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Math=C3=A9o=20Nebulo9=20DESSAUVAGES?= Date: Tue, 14 Jul 2026 16:12:16 +0200 Subject: [PATCH] feat: Add Russian Roulette feature betwwen two players --- .gitignore | 4 +- Dockerfile | 7 +- cogs/cs.py | 34 ++++++-- cogs/insult.py | 3 +- cogs/roulette.py | 212 +++++++++++++++++++++++++++++++++++++++++++++++ main.py | 121 +++------------------------ setup/bot.py | 70 ++++++++++++++-- setup/config.py | 18 ++++ 8 files changed, 341 insertions(+), 128 deletions(-) create mode 100644 cogs/roulette.py create mode 100644 setup/config.py diff --git a/.gitignore b/.gitignore index 886492e..b6e1cad 100644 --- a/.gitignore +++ b/.gitignore @@ -162,4 +162,6 @@ cython_debug/ data/ .vscode/ -*.disabled \ No newline at end of file +*.disabled + +config/ \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index 3259dc2..8b5b41d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,12 +3,17 @@ FROM python:3.13.14-alpine WORKDIR /app COPY requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt +RUN pip install --no-cache-dir -r requirements.txt && \ + apk add --no-cache firefox geckodriver COPY . . ENV TOKEN="" ENV URL="" +ENV CONFIG_PATH="/config" +ENV COGS="" +ENV GECKODRIVER_PATH="/usr/bin/geckodriver" +RUN mkdir ${CONFIG_PATH} ENTRYPOINT ["python", "main.py"] \ No newline at end of file diff --git a/cogs/cs.py b/cogs/cs.py index 7cfb1df..b936753 100644 --- a/cogs/cs.py +++ b/cogs/cs.py @@ -1,15 +1,18 @@ import discord, datetime, os, re, json from selenium import webdriver +from selenium.webdriver.firefox.options import Options +from selenium.webdriver.firefox.service import Service from bs4 import BeautifulSoup from discord.ext import commands, tasks, bridge from setup.logger import LOGGER +from setup.config import Config from models.cs_models import * BASE_URL = "https://www.hltv.org" -PAGE_FILE = os.path.join("data","hltv.html") +PAGE_FILE = os.path.join(Config.CONFIG_PATH,"hltv.html") -MATCHES_FILE = os.path.join("data","matches.json") +MATCHES_FILE = os.path.join(Config.CONFIG_PATH,"matches.json") # Cache duration in seconds (6 hours) DATA_CACHE_DURATION = 3600 * 6 @@ -43,14 +46,27 @@ class CSCog(commands.Cog): return False def get_page(self) -> None: - """Get the page from https://www.hltv.org/matches/ using Firefox webdriver and save to PAGE_FILE.""" - driver = webdriver.Firefox() - driver.get("https://www.hltv.org/matches/") - html = driver.page_source - driver.close() + """Get the page from https://www.hltv.org/matches/ using a headless Firefox webdriver and save to PAGE_FILE.""" + options = Options() + options.add_argument("--headless=new") + options.add_argument("--no-sandbox") + options.add_argument("--disable-dev-shm-usage") + options.add_argument("--window-size=1920,1080") - with open(PAGE_FILE, 'w') as f: - f.write(html) + geckodriver_path = os.getenv("GECKODRIVER_PATH") + service = Service(executable_path=geckodriver_path) if geckodriver_path else None + + driver = None + try: + driver = webdriver.Firefox(service=service, options=options) + driver.get("https://www.hltv.org/matches/") + html = driver.page_source + + with open(PAGE_FILE, 'w', encoding='utf-8') as f: + f.write(html) + finally: + if driver is not None: + driver.quit() def save(self) -> None: """Save self.matches to MATCHES_FILE.""" diff --git a/cogs/insult.py b/cogs/insult.py index 57a21c5..c73ec48 100644 --- a/cogs/insult.py +++ b/cogs/insult.py @@ -2,12 +2,13 @@ import discord, json, os, random, re import collections from discord.ext import commands from setup.logger import LOGGER +from setup.config import Config class InsultsCog(commands.Cog): def __init__(self, bot:discord.Bot): self.bot = bot - with open(os.path.join("data", "insults.txt")) as f: + with open(os.path.join(Config.CONFIG_PATH, "insults.txt")) as f: self.insults = [l.strip() for l in f.readlines()] @commands.Cog.listener() diff --git a/cogs/roulette.py b/cogs/roulette.py new file mode 100644 index 0000000..6c4c2ca --- /dev/null +++ b/cogs/roulette.py @@ -0,0 +1,212 @@ +import asyncio +import datetime +import random + +import discord +from discord import option +from discord.ext import commands + + +class RouletteCog(commands.Cog): + def __init__(self, bot: discord.Bot): + self.bot = bot + self.cooldown = 60 * 3 + self.active_games = {} + + def _build_view(self, game: dict) -> discord.ui.View: + view = discord.ui.View(timeout=None) + + if game["status"] == "pending": + async def accept_callback(interaction: discord.Interaction): + if interaction.user.id != game["target"].id: + await interaction.response.send_message( + "Only the challenged member can accept this duel.", + ephemeral=True, + ) + return + + game["status"] = "ready" + game["current_player"] = game["author"] + + if game.get("message"): + await game["message"].edit( + content=( + f"{game['author'].mention} challenged {game['target'].mention} to a Russian roulette duel.\n" + f"✅ {game['target'].mention} accepted. {game['current_player'].mention}, click Fire!" + ), + view=self._build_view(game), + ) + + await interaction.response.defer() + + async def decline_callback(interaction: discord.Interaction): + if interaction.user.id != game["target"].id: + await interaction.response.send_message( + "Only the challenged member can decline this duel.", + ephemeral=True, + ) + return + + game["status"] = "declined" + self.active_games.pop((game["guild_id"], game["author"].id, game["target"].id), None) + + if game.get("message"): + await game["message"].edit( + content=f"{game['target'].mention} declined the challenge.", + view=None, + ) + + await interaction.response.defer() + + accept_button = discord.ui.Button(label="Accept", style=discord.ButtonStyle.green) + accept_button.callback = accept_callback + view.add_item(accept_button) + + decline_button = discord.ui.Button(label="Decline", style=discord.ButtonStyle.red) + decline_button.callback = decline_callback + view.add_item(decline_button) + + elif game["status"] in {"ready", "playing"}: + async def fire_callback(interaction: discord.Interaction): + if interaction.user.id != game["current_player"].id: + await interaction.response.send_message( + f"Wait for {game['current_player'].mention} to fire.", + ephemeral=True, + ) + return + + if interaction.user.id in game["fired"]: + await interaction.response.send_message( + "You already fired once. Only one shot per player is allowed.", + ephemeral=True, + ) + return + + game["fired"].add(interaction.user.id) + is_shot = random.randint(1, 6) == 1 + + if is_shot: + game["status"] = "finished" + self.active_games.pop((game["guild_id"], game["author"].id, game["target"].id), None) + + await self._mute_member(interaction.user, 60) + + if game.get("message"): + await game["message"].edit( + content=( + f"💥 {interaction.user.mention} pulled the trigger and got shot.\n" + f"They were muted for 60 seconds." + ), + view=None, + ) + else: + if len(game["fired"]) >= 2: + game["status"] = "finished" + self.active_games.pop((game["guild_id"], game["author"].id, game["target"].id), None) + + if game.get("message"): + await game["message"].edit( + content=( + f"🔫 {interaction.user.mention} survived and both players have fired once.\n" + f"No one was shot. The duel ends in a draw." + ), + view=None, + ) + else: + game["status"] = "playing" + game["current_player"] = game["target"] if interaction.user.id == game["author"].id else game["author"] + + if game.get("message"): + await game["message"].edit( + content=( + f"🔫 {interaction.user.mention} survived.\n" + f"Next turn: {game['current_player'].mention}" + ), + view=self._build_view(game), + ) + + await interaction.response.defer() + + fire_button = discord.ui.Button(label="Fire", style=discord.ButtonStyle.danger) + fire_button.callback = fire_callback + view.add_item(fire_button) + + return view + + async def _mute_member(self, member: discord.Member, seconds: int = 60): + until = discord.utils.utcnow() + datetime.timedelta(seconds=seconds) + try: + await member.timeout(until=until, reason="Russian roulette loss") + except AttributeError: + await member.edit(timed_out_until=until, reason="Russian roulette loss") + + @commands.slash_command( + name="roulette", + description="Challenge other server members. The loser is getting muted for 1 minute", + ) + @option( + name="target", + description="The member to challenge", + input_type=discord.Member, + required=True, + ) + async def roulette( + self, + ctx: discord.ApplicationContext, + target: discord.Member, + ): + author = ctx.author + self_test = author.id == target.id + + if self_test: + target = author + + game = { + "author": author, + "target": target, + "status": "pending", + "current_player": None, + "guild_id": ctx.guild_id, + "message": None, + "fired": set(), + } + + self.active_games[(ctx.guild_id, author.id, target.id)] = game + + await ctx.send_response( + f"{author.mention} challenged {target.mention} to a Russian roulette duel.\n" + f"The target has 60 seconds to accept or decline.", + view=self._build_view(game), + ) + + if self_test: + await ctx.send_followup( + "**Self-challenge enabled for testing.**\n" + "The game will start immediately and you can test the flow.", + ephemeral=True, + ) + + try: + game["message"] = await ctx.interaction.original_response() + except Exception: + game["message"] = None + + async def timeout_handler(): + try: + await asyncio.sleep(60) + except asyncio.CancelledError: + return + + if game.get("status") == "pending": + game["status"] = "timeout" + self.active_games.pop((ctx.guild_id, author.id, target.id), None) + + await self._mute_member(target, 60) + + if game.get("message"): + await game["message"].edit( + content=f"⏰ {target.mention} did not respond in time and was muted for 60 seconds.", + view=None, + ) + + asyncio.create_task(timeout_handler()) \ No newline at end of file diff --git a/main.py b/main.py index a783859..8d0f5b9 100644 --- a/main.py +++ b/main.py @@ -2,130 +2,33 @@ import os, discord, json, sys, random from argparse import ArgumentParser from datetime import datetime from asyncio import create_task -from dotenv import load_dotenv from discord import option from discord.ext import tasks, commands -from setup.bot import bot, reload_feature +from setup.bot import bot, reload_feature, register_cogs from setup.logger import LOGGER +from setup.config import Config + + PARSER = ArgumentParser(description='BotMafieux for Discord.') 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() - -# @tasks.loop(time=datetime.time(datetime.strptime('00:00','%H:%M'))) -# async def birthday_anouncements_task(): -# LOGGER.debug('birthday_anouncements_task started.') -# for guild in bot.guilds: -# birthday_settings = db.select('birthday_settings',f'guild_id = {guild.id}') -# if birthday_settings: -# if birthday_settings['is_enabled']: -# for user in guild.members: -# user_productivity_data = db.select('guild_user_productivity',f'user_id = {user.id} AND guild_id = {guild.id}') -# if user_productivity_data: -# if user_productivity_data['is_enabled']: -# user = guild.get_member(user.id) -# if user: -# birthday = user_productivity_data['birthday'] -# today = datetime.today().strftime('%d/%m') -# if birthday == today: -# channel = guild.get_channel(user_productivity_data['channel_id']) -# if channel: -# message = user_productivity_data['birthday_message'] -# if message: -# message = message.replace('{user}',user.mention) -# await channel.send(message) -# else: -# LOGGER.debug(f'Channel {user_productivity_data["channel_id"]} not found.') -# else: -# LOGGER.debug(f'No birthday today for user {user.name} ({user.id}).') -# else: -# LOGGER.debug(f'User {user.name} ({user.id}) not found.') -# else: -# LOGGER.debug(f'User {user.name} ({user.id}) has birthday announcements disabled.') -# else: -# LOGGER.debug(f'No birthday data found for user {user.name} ({user.id}).') -# else: -# LOGGER.debug(f'No productivity settings found for guild {guild.name} ({guild.id}).') -# LOGGER.debug('birthday_anouncements_task ended.') - -# @bot.slash_command(name='help',description='Displays the help message.') -# async def help(ctx:discord.ApplicationContext): -# """Displays this message.""" -# command_name = 'help' -# LOGGER.debug(f'{ctx.author.name} used /{command_name}.') -# embed = discord.Embed(title='Help',description='List of commands and their descriptions.',color=discord.Color.from_rgb(171,0,219)) -# embed.set_author(name=bot.user.name,icon_url=bot.user.avatar.url) -# for command in bot.commands: -# embed.add_field(name=command.name,value=command.description,inline=False) -# await ctx.send_response(embed=embed,ephemeral=True) - -# @bot.slash_command(name='reload',description='Reloads the bot.') -# @option(name='feature',description='The feature to reload.',required=True,choices=['birthday','productivity']) -# @commands.is_owner() -# async def reload(ctx:discord.ApplicationContext, feature:str): -# """Reloads the bot.""" -# command_name = 'reload' -# guild = ctx.guild -# channel = ctx.channel -# LOGGER.debug(f'{ctx.author.name} used /{command_name} {feature}.') -# try: -# reload_feature(feature) -# await ctx.send_response(f'{feature} reloaded!',ephemeral=True) -# except Exception as e: -# LOGGER.error(f'Error reloading feature {feature}: {e}') -# await ctx.send_response(f'Error reloading feature {feature}: {e}',ephemeral=True) - -# @bot.event -# async def on_member_join(member:discord.Member): -# if not member.bot: -# user_id = member.id -# user_name = member.name -# user_mention = member.mention -# db_user = db.select('global_user',f'WHERE user_id = {user_id}') -# if not db_user: -# db.insert('global_user',user_id=user_id,user_name=user_name,user_mention=user_mention) - -# @bot.event -# async def on_raw_member_remove(payload:discord.RawMemberRemoveEvent): -# user = payload.user -# if not user.bot: -# user_id = user.id -# guild_id = payload.guild_id -# db.delete('guild_user_birthday',f'user_id = {user_id} AND guild_id = {guild_id}') -# db.delete('guild_user_productivity',f'user_id = {user_id} AND guild_id = {guild_id}') - -# @bot.event -# async def on_guild_join(guild:discord.Guild): -# guild_id = guild.id -# guild_name = guild.name -# LOGGER.info(f'Bot joined guild {guild_name} ({guild_id}).') -# db.insert('guild',guild_id=guild_id) -# db.insert('birthday_settings',guild_id=guild_id,is_enabled=False,channel_id=random.choice(guild.text_channels).id) -# db.insert('productivity_settings',guild_id=guild_id,is_enabled=False) - -# @bot.event -# async def on_guild_remove(guild:discord.Guild): -# guild_id = guild.id -# guild_name = guild.name -# LOGGER.info(f'Bot left guild {guild_name} ({guild_id}).') -# db.delete('birthday_settings',f'guild_id = {guild_id}') -# db.delete('productivity_settings',f'guild_id = {guild_id}') -# db.delete('guild',f'guild_id = {guild_id}') +selected_cogs = EXEC_ARGS.cogs if EXEC_ARGS.cogs is not None else Config.get_cogs() +register_cogs(selected_cogs) @bot.event async def on_ready(): if bot.user: LOGGER.info(f'{bot.user.name} has connected to Discord!') - LOGGER.debug(f'Guilds: {','.join([guild.name for guild in bot.guilds])}') - + LOGGER.debug(f"Guilds: {','.join([guild.name for guild in bot.guilds])}") + if __name__ == '__main__': - load_dotenv() - TOKEN = os.getenv("TOKEN") - if TOKEN: - bot.run(TOKEN) + token = Config.TOKEN + if token: + bot.run(token) else: LOGGER.error('No token found.') sys.exit(1) - \ No newline at end of file diff --git a/setup/bot.py b/setup/bot.py index 116204d..bb0612d 100644 --- a/setup/bot.py +++ b/setup/bot.py @@ -4,6 +4,7 @@ from discord.ext import bridge # from cogs.productivity import ProductivityCog from cogs.cs import CSCog from cogs.insult import InsultsCog +from cogs.roulette import RouletteCog from setup.logger import LOGGER bot_intents = discord.Intents.default() @@ -15,15 +16,70 @@ bot_intents.reactions = True bot = bridge.Bot(command_prefix='$', intents=bot_intents) -COGS = [CSCog, InsultsCog] +COG_REGISTRY = { + 'cs': CSCog, + 'insult': InsultsCog, + 'roulette': RouletteCog +} -for cog in COGS: - bot.add_cog(cog(bot)) -def reload_feature(feature:str): +def _normalize_cog_names(cogs=None): + if cogs is None: + return list(COG_REGISTRY.keys()) + + if isinstance(cogs, str): + cogs = [cogs] + + selected_cogs = [] + for cog_name in cogs: + name = str(cog_name).strip().lower() + if not name: + continue + if name in {'all', '*'}: + return list(COG_REGISTRY.keys()) + if name in COG_REGISTRY: + selected_cogs.append(name) + else: + LOGGER.warning("Unknown cog '%s'. Available cogs: %s", cog_name, ', '.join(COG_REGISTRY.keys())) + + return list(dict.fromkeys(selected_cogs)) + + +def register_cogs(cogs=None): + selected_cogs = _normalize_cog_names(cogs) + + existing_cog_names = set(COG_REGISTRY.values()) + for existing_name in list(bot.cogs.keys()): + if existing_name in {cog_class.__name__ for cog_class in existing_cog_names}: + bot.remove_cog(existing_name) + + for cog_name in selected_cogs: + cog_class = COG_REGISTRY[cog_name] + bot.add_cog(cog_class(bot)) + + return selected_cogs + + + +def reload_feature(feature: str): + normalized_feature = str(feature).strip().lower() LOGGER.debug(bot.cogs) + + if normalized_feature in {'all', '*'}: + register_cogs() + LOGGER.debug('All cogs reloaded!') + return + + if normalized_feature not in COG_REGISTRY: + LOGGER.warning("Unknown cog '%s'. Available cogs: %s", feature, ', '.join(COG_REGISTRY.keys())) + return + LOGGER.debug(f'Reloading {feature}...') - bot.remove_cog(feature.capitalize()+'Cog') - cog = globals()[feature.capitalize()+'Cog'] - bot.add_cog(cog(bot)) + target_cog_name = COG_REGISTRY[normalized_feature].__name__ + for existing_name in list(bot.cogs.keys()): + if existing_name == target_cog_name: + bot.remove_cog(existing_name) + break + + bot.add_cog(COG_REGISTRY[normalized_feature](bot)) LOGGER.debug(f'{feature} reloaded!') diff --git a/setup/config.py b/setup/config.py new file mode 100644 index 0000000..2678ff2 --- /dev/null +++ b/setup/config.py @@ -0,0 +1,18 @@ +import os +from dotenv import load_dotenv + + +class Config: + load_dotenv() + + TOKEN = str(os.getenv("TOKEN", "")) + CONFIG_PATH = str(os.getenv("CONFIG_PATH")) + COGS = os.getenv("COGS", "") + + @classmethod + def get_cogs(cls): + raw_value = cls.COGS.strip() + if not raw_value: + return None + return [item.strip().lower() for item in raw_value.split(",") if item.strip()] +