import os import json import logging import asyncio import disnake from disnake.ext import commands from typing import Tuple, Optional, List from dotenv import load_dotenv # Logger instance kuby_logger = logging.getLogger("KubyBot") # Project paths UTILS_DIR = os.path.dirname(os.path.abspath(__file__)) PROJECT_ROOT = os.path.dirname(UTILS_DIR) # Explicitly load .env file from the project root load_dotenv(os.path.join(PROJECT_ROOT, ".env")) # Whitelist file location WHITELIST_PARTNERS_FILE = os.path.join(PROJECT_ROOT, "whitelist_partners.json") async def load_partners_whitelist() -> List[int]: """Asynchronously loads the whitelist partners JSON file. Returns: List[int]: A list of whitelisted guild IDs. """ def read_file(): if not os.path.exists(WHITELIST_PARTNERS_FILE): # Create a default empty file if it doesn't exist try: with open(WHITELIST_PARTNERS_FILE, "w", encoding="utf-8") as f: json.dump({"partners": []}, f, indent=4) return [] except Exception as e: kuby_logger.error(f"Failed to create default whitelist file: {e}") return [] try: with open(WHITELIST_PARTNERS_FILE, "r", encoding="utf-8") as f: data = json.load(f) return data.get("partners", []) except (json.JSONDecodeError, IOError) as e: kuby_logger.error(f"Failed to read whitelist file at {WHITELIST_PARTNERS_FILE}: {e}") return [] return await asyncio.to_thread(read_file) def get_sku_ids() -> Tuple[Optional[int], Optional[int], Optional[int]]: """Loads and returns the SKU IDs from the environment variables. Returns: Tuple[Optional[int], Optional[int], Optional[int]]: Starter, Medium, and Large SKU IDs. """ def parse_env_int(key: str) -> Optional[int]: val = os.getenv(key) if not val: return None try: return int(val.strip()) except ValueError: kuby_logger.error(f"Invalid environment variable format for {key}: '{val}' is not an integer.") return None starter = parse_env_int("STARTER_SKU_ID") medium = parse_env_int("MEDIUM_SKU_ID") # Supports both LARGE_SKU_ID and MAX_SKU_ID large = parse_env_int("LARGE_SKU_ID") or parse_env_int("MAX_SKU_ID") return starter, medium, large def check_premium_tier(): """A custom disnake command check that enforces premium tier limitations based on server size and Discord Entitlements. IMPORTANT: - Set DISABLE_PREMIUM_SKU_CHECK=true to bypass SKU entitlement checks (useful for dev/test). """ async def predicate(interaction: disnake.ApplicationCommandInteraction) -> bool: # Dev/test bypass: allow commands even when SKU env vars aren't configured. if os.getenv("DISABLE_PREMIUM_SKU_CHECK", "").lower() in {"1", "true", "yes", "y", "on"}: kuby_logger.warning( "[PREMIUM CHECK] Bypass enabled via DISABLE_PREMIUM_SKU_CHECK. " f"guild_id={getattr(interaction.guild, 'id', None)}" ) return True # Debugging Log debug_guild_id = interaction.guild.id if interaction.guild else None debug_member_count = interaction.guild.member_count if interaction.guild else None try: debug_entitlements = [{"sku_id": ent.sku_id, "active": ent.is_active()} for ent in interaction.entitlements] except Exception as e: debug_entitlements = f"Error: {e}" # Resolve SKU IDs from env for easier diagnosis starter_id, medium_id, large_id = get_sku_ids() kuby_logger.debug( "[DEBUG PREMIUM CHECK] " f"Guild ID: {debug_guild_id} | " f"Member Count: {debug_member_count} | " f"Entitlements: {debug_entitlements} | " f"Env SKUs: starter={starter_id} medium={medium_id} large={large_id}" ) # 1. Refuse access if the command was run in DMs if interaction.guild is None: kuby_logger.warning("Access Denied: Command run in DMs.") await interaction.response.send_message( "❌ Cette commande ne peut être exécutée que sur un serveur Discord.", ephemeral=True ) return False guild_id = interaction.guild.id # 2. Check the Partners Whitelist try: partners = await load_partners_whitelist() if guild_id in partners: kuby_logger.info(f"Access Granted: Guild {guild_id} ({interaction.guild.name}) is in whitelist.") return True except Exception as e: # Safe fail-closed pattern: Log error and continue to premium checks if whitelist loading crashed. kuby_logger.error(f"Error checking partners whitelist for guild {guild_id}: {e}", exc_info=True) # 3. Retrieve and validate the guild member count member_count = interaction.guild.member_count if member_count is None or not isinstance(member_count, int) or member_count <= 0: kuby_logger.error(f"Fail-Safe triggered: Invalid member count {member_count} for guild {guild_id}.") await interaction.response.send_message( "❌ Impossible de vérifier la taille du serveur par sécurité. Veuillez réessayer plus tard.", ephemeral=True ) return False # 4. Check active entitlements and SKUs try: active_skus = {ent.sku_id for ent in interaction.entitlements if ent.is_active()} except Exception as e: kuby_logger.error(f"Fail-Closed: Error retrieving entitlements for guild {guild_id}: {e}", exc_info=True) await interaction.response.send_message( "❌ Erreur lors de la vérification de votre abonnement auprès de Discord. Veuillez réessayer plus tard.", ephemeral=True ) return False # Identify required tier & allowed SKUs if member_count < 500: required_tier_name = "Starter" allowed_skus = [starter_id, medium_id, large_id] elif 500 <= member_count <= 2500: required_tier_name = "Medium" allowed_skus = [medium_id, large_id] else: required_tier_name = "Large" allowed_skus = [large_id] # Filter out unconfigured SKUs (None values) allowed_skus = [sku for sku in allowed_skus if sku is not None] # If no valid SKU ID is configured in the environment for this tier, block access if not allowed_skus: kuby_logger.error( f"Configuration Error: No SKU IDs configured in .env for tier '{required_tier_name}' " f"(Starter: {starter_id}, Medium: {medium_id}, Large: {large_id}). Access blocked." ) await interaction.response.send_message( "❌ Une erreur de configuration empêche l'accès à cette commande premium. " "Veuillez contacter le support de l'association Omega Kube.", ephemeral=True ) return False # Verify entitlement matching one of the allowed SKU IDs has_valid_sku = any(sku in active_skus for sku in allowed_skus) if has_valid_sku: kuby_logger.info( f"Access Granted: Guild {guild_id} ({interaction.guild.name}) " f"passed premium check (Members: {member_count}, Tier: {required_tier_name})." ) return True # 5. Block access and send polite ephemeral instruction message kuby_logger.info( f"Access Denied: Guild {guild_id} ({interaction.guild.name}) " f"has {member_count} members but no valid SKU. Active SKUs: {active_skus}." ) await interaction.response.send_message( f"✨ **Fonctionnalité Premium** ✨\n\n" f"Merci d'utiliser le bot **Kuby** ! Cette commande nécessite un abonnement actif.\n" f"📊 **Statistiques de votre serveur :**\n" f"- Nombre de membres : `{member_count}`\n" f"- Palier requis : **Palier {required_tier_name}**\n\n" f"💡 **Comment s'abonner ?**\n" f"Vous pouvez vous abonner directement depuis le profil du bot Discord ou via l'App Directory.\n\n" f"Si vous êtes un partenaire officiel de l'association **Omega Kube**, " f"veuillez contacter un administrateur pour être ajouté à la Whitelist.", ephemeral=True ) return False return commands.check(predicate)