from modules.globalvars import * # Command: Retrain the Markov model from memory @bot.hybrid_command(description=f"{(_('command_desc_retrain'))}") async def retrain(ctx: commands.Context) -> None: if ctx.author.id != ownerid: return message_ref: MessageReference = await send_message(ctx, f"{(_('command_markov_retrain'))}") try: with open(MEMORY_FILE, 'r') as f: memory: List[str] = json.load(f) except FileNotFoundError: await send_message(ctx, f"{(_('command_markov_memory_not_found'))}") return except json.JSONDecodeError: await send_message(ctx, f"{(_('command_markov_memory_is_corrupt'))}") return data_size: int = len(memory) processed_data: int = 0 processing_message_ref: MessageReference = await send_message(ctx, f"{(_('command_markov_retraining')).format(processed_data=processed_data, data_size=data_size)}") start_time: float = time.time() for i, data in enumerate(memory): processed_data += 1 global markov_model markov_model = train_markov_model(memory) save_markov_model(markov_model) await send_message(ctx, f"{_('command_markov_retrain_successful').format(data_size=data_size)}", edit=True, message_reference=processing_message_ref) # Command: Generate a sentence using the Markov model @bot.hybrid_command(description=f"{(_('command_desc_talk'))}") async def talk(ctx: commands.Context, sentence_size: int = 5) -> None: if not markov_model: await send_message(ctx, f"{(_('command_talk_insufficent_text'))}") return response: Optional[str] = None for _ in range(20): if sentence_size == 1: response = markov_model.make_short_sentence(max_chars=100, tries=100) if response: response = response.split()[0] else: response = markov_model.make_sentence(tries=100, max_words=sentence_size) if response and response not in generated_sentences: if sentence_size > 1: response = improve_sentence_coherence(response) generated_sentences.add(response) break if response: cleaned_response: str = re.sub(r'[^\w\s]', '', response).lower() coherent_response: str = rephrase_for_coherence(cleaned_response) if random.random() < 0.9 and is_positive(coherent_response): gif_url: str = random.choice(positive_gifs) combined_message: str = f"{coherent_response}\n[jif]({gif_url})" else: combined_message: str = coherent_response logger.info(combined_message) os.environ['gooberlatestgen'] = combined_message await send_message(ctx, combined_message) else: await send_message(ctx, f"{(_('command_talk_generation_fail'))}") # Command: Generate an image @bot.hybrid_command(description=f"{(_('command_desc_help'))}") async def impact(ctx: commands.Context, text: Optional[str] = None) -> None: assets_folder: str = "assets/images" temp_input: Optional[str] = None def get_random_asset_image() -> Optional[str]: files: List[str] = [f for f in os.listdir(assets_folder) if f.lower().endswith(('.png', '.jpg', '.jpeg', '.webp'))] if not files: return None return os.path.join(assets_folder, random.choice(files)) if ctx.message.attachments: attachment: discord.Attachment = ctx.message.attachments[0] if attachment.content_type and attachment.content_type.startswith("image/"): ext: str = os.path.splitext(attachment.filename)[1] temp_input = f"tempy{ext}" await attachment.save(temp_input) input_path: str = temp_input else: fallback_image: Optional[str] = get_random_asset_image() if fallback_image is None: await ctx.reply(_('no_image_available')) return temp_input = tempfile.mktemp(suffix=os.path.splitext(fallback_image)[1]) shutil.copy(fallback_image, temp_input) input_path = temp_input else: fallback_image = get_random_asset_image() if fallback_image is None: await ctx.reply(_('no_image_available')) return temp_input = tempfile.mktemp(suffix=os.path.splitext(fallback_image)[1]) shutil.copy(fallback_image, temp_input) input_path = temp_input output_path: Optional[str] = await gen_meme(input_path, custom_text=text) if output_path is None or not os.path.isfile(output_path): if temp_input and os.path.exists(temp_input): os.remove(temp_input) await ctx.reply(_('failed_generate_image')) return await ctx.send(file=discord.File(output_path)) if temp_input and os.path.exists(temp_input): os.remove(temp_input) # New demotivator command @bot.hybrid_command(description="Generate a demotivator poster with two lines of text") async def demotivator(ctx: commands.Context) -> None: assets_folder: str = "assets/images" temp_input: Optional[str] = None def get_random_asset_image() -> Optional[str]: files: List[str] = [f for f in os.listdir(assets_folder) if f.lower().endswith(('.png', '.jpg', '.jpeg', '.webp'))] if not files: return None return os.path.join(assets_folder, random.choice(files)) if ctx.message.attachments: attachment: discord.Attachment = ctx.message.attachments[0] if attachment.content_type and attachment.content_type.startswith("image/"): ext: str = os.path.splitext(attachment.filename)[1] temp_input = f"tempy{ext}" await attachment.save(temp_input) input_path: str = temp_input else: fallback_image: Optional[str] = get_random_asset_image() if fallback_image is None: await ctx.reply(_('no_image_available')) return temp_input = tempfile.mktemp(suffix=os.path.splitext(fallback_image)[1]) shutil.copy(fallback_image, temp_input) input_path = temp_input else: fallback_image = get_random_asset_image() if fallback_image is None: await ctx.reply(_('no_image_available')) return temp_input = tempfile.mktemp(suffix=os.path.splitext(fallback_image)[1]) shutil.copy(fallback_image, temp_input) input_path = temp_input output_path: Optional[str] = await gen_demotivator(input_path) if output_path is None or not os.path.isfile(output_path): if temp_input and os.path.exists(temp_input): os.remove(temp_input) await ctx.reply("Failed to generate demotivator.") return await ctx.send(file=discord.File(output_path)) if temp_input and os.path.exists(temp_input): os.remove(temp_input) bot.remove_command('help') # Command: Show help information @bot.hybrid_command(description=f"{(_('command_desc_help'))}") async def help(ctx: commands.Context) -> None: embed: discord.Embed = discord.Embed( title=f"{(_('command_help_embed_title'))}", description=f"{(_('command_help_embed_desc'))}", color=Colour(0x000000) ) command_categories: Dict[str, List[str]] = { f"{(_('command_help_categories_general'))}": ["mem", "talk", "about", "ping", "impact", "demotivator", "help"], f"{(_('command_help_categories_admin'))}": ["stats", "retrain", "setlanguage"] } custom_commands: List[str] = [] for cog_name, cog in bot.cogs.items(): for command in cog.get_commands(): if command.name not in command_categories[f"{(_('command_help_categories_general'))}"] and command.name not in command_categories[f"{(_('command_help_categories_admin'))}"]: custom_commands.append(command.name) if custom_commands: embed.add_field(name=f"{(_('command_help_categories_custom'))}", value="\n".join([f"{PREFIX}{command}" for command in custom_commands]), inline=False) for category, commands_list in command_categories.items(): commands_in_category: str = "\n".join([f"{PREFIX}{command}" for command in commands_list]) embed.add_field(name=category, value=commands_in_category, inline=False) await send_message(ctx, embed=embed) @bot.hybrid_command(description=f"{(_('command_desc_setlang'))}") @app_commands.describe(locale="Choose your language") async def setlanguage(ctx: commands.Context, locale: str) -> None: if ctx.author.id != ownerid: await ctx.send(":thumbsdown:") return await ctx.defer() set_language(locale) await ctx.send(":thumbsup:")