# utils/bipolar_mode.py """ Bipolar Mode module for Miku. Allows both Regular Miku and Evil Miku to coexist and argue via webhooks. When active, there's a chance for the inactive persona to "break through" and trigger an argument. """ import os import json import random import asyncio import discord import globals # ============================================================================ # CONSTANTS # ============================================================================ BIPOLAR_STATE_FILE = "memory/bipolar_mode_state.json" BIPOLAR_WEBHOOKS_FILE = "memory/bipolar_webhooks.json" BIPOLAR_SCOREBOARD_FILE = "memory/bipolar_scoreboard.json" # Argument settings MIN_EXCHANGES = 4 # Minimum number of back-and-forth exchanges before ending can occur ARGUMENT_TRIGGER_CHANCE = 0.15 # 15% chance for the other Miku to break through DELAY_BETWEEN_MESSAGES = (2.0, 5.0) # Random delay between argument messages (seconds) # ============================================================================ # STATE PERSISTENCE # ============================================================================ def save_bipolar_state(): """Save bipolar mode state to JSON file""" try: state = { "bipolar_mode_enabled": globals.BIPOLAR_MODE, "argument_in_progress": globals.BIPOLAR_ARGUMENT_IN_PROGRESS } with open(BIPOLAR_STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, indent=2) print(f"๐Ÿ’พ Saved bipolar mode state: enabled={globals.BIPOLAR_MODE}") except Exception as e: print(f"โš ๏ธ Failed to save bipolar mode state: {e}") def load_bipolar_state(): """Load bipolar mode state from JSON file""" try: if not os.path.exists(BIPOLAR_STATE_FILE): print("โ„น๏ธ No bipolar mode state file found, using defaults") return False with open(BIPOLAR_STATE_FILE, "r", encoding="utf-8") as f: state = json.load(f) bipolar_mode = state.get("bipolar_mode_enabled", False) print(f"๐Ÿ“‚ Loaded bipolar mode state: enabled={bipolar_mode}") return bipolar_mode except Exception as e: print(f"โš ๏ธ Failed to load bipolar mode state: {e}") return False def save_webhooks(): """Save webhook URLs to JSON file""" try: # Convert guild_id keys to strings for JSON webhooks_data = {} for guild_id, webhook_data in globals.BIPOLAR_WEBHOOKS.items(): webhooks_data[str(guild_id)] = webhook_data with open(BIPOLAR_WEBHOOKS_FILE, "w", encoding="utf-8") as f: json.dump(webhooks_data, f, indent=2) print(f"๐Ÿ’พ Saved bipolar webhooks for {len(webhooks_data)} server(s)") except Exception as e: print(f"โš ๏ธ Failed to save bipolar webhooks: {e}") def load_webhooks(): """Load webhook URLs from JSON file""" try: if not os.path.exists(BIPOLAR_WEBHOOKS_FILE): print("โ„น๏ธ No bipolar webhooks file found") return {} with open(BIPOLAR_WEBHOOKS_FILE, "r", encoding="utf-8") as f: webhooks_data = json.load(f) # Convert string keys back to int webhooks = {} for guild_id_str, webhook_data in webhooks_data.items(): webhooks[int(guild_id_str)] = webhook_data print(f"๐Ÿ“‚ Loaded bipolar webhooks for {len(webhooks)} server(s)") return webhooks except Exception as e: print(f"โš ๏ธ Failed to load bipolar webhooks: {e}") return {} def restore_bipolar_mode_on_startup(): """Restore bipolar mode state on bot startup""" bipolar_mode = load_bipolar_state() globals.BIPOLAR_MODE = bipolar_mode globals.BIPOLAR_WEBHOOKS = load_webhooks() if bipolar_mode: print("๐Ÿ”„ Bipolar mode restored from previous session") return bipolar_mode # ============================================================================ # SCOREBOARD MANAGEMENT # ============================================================================ def load_scoreboard() -> dict: """Load the bipolar argument scoreboard""" try: if not os.path.exists(BIPOLAR_SCOREBOARD_FILE): return {"miku": 0, "evil": 0, "history": []} with open(BIPOLAR_SCOREBOARD_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: print(f"โš ๏ธ Failed to load scoreboard: {e}") return {"miku": 0, "evil": 0, "history": []} def save_scoreboard(scoreboard: dict): """Save the bipolar argument scoreboard""" try: os.makedirs(os.path.dirname(BIPOLAR_SCOREBOARD_FILE), exist_ok=True) with open(BIPOLAR_SCOREBOARD_FILE, "w", encoding="utf-8") as f: json.dump(scoreboard, f, indent=2) print(f"๐Ÿ’พ Saved scoreboard: Miku {scoreboard['miku']} - {scoreboard['evil']} Evil Miku") except Exception as e: print(f"โš ๏ธ Failed to save scoreboard: {e}") def record_argument_result(winner: str, exchanges: int, reasoning: str = ""): """Record the result of an argument in the scoreboard Args: winner: 'miku', 'evil', or 'draw' exchanges: Number of exchanges in the argument reasoning: Arbiter's reasoning for the decision """ scoreboard = load_scoreboard() # Increment winner's score if winner in ["miku", "evil"]: scoreboard[winner] = scoreboard.get(winner, 0) + 1 # Add to history if "history" not in scoreboard: scoreboard["history"] = [] from datetime import datetime scoreboard["history"].append({ "winner": winner, "exchanges": exchanges, "timestamp": datetime.now().isoformat(), "reasoning": reasoning # Store arbiter's reasoning }) # Keep only last 50 results in history if len(scoreboard["history"]) > 50: scoreboard["history"] = scoreboard["history"][-50:] save_scoreboard(scoreboard) return scoreboard def get_scoreboard_summary() -> str: """Get a formatted summary of the scoreboard""" scoreboard = load_scoreboard() miku_wins = scoreboard.get("miku", 0) evil_wins = scoreboard.get("evil", 0) total = miku_wins + evil_wins if total == 0: return "No arguments have been judged yet." miku_pct = (miku_wins / total * 100) if total > 0 else 0 evil_pct = (evil_wins / total * 100) if total > 0 else 0 return f"""**Bipolar Mode Scoreboard** ๐Ÿ† Hatsune Miku: {miku_wins} wins ({miku_pct:.1f}%) Evil Miku: {evil_wins} wins ({evil_pct:.1f}%) Total Arguments: {total}""" # ============================================================================ # BIPOLAR MODE TOGGLE # ============================================================================ def is_bipolar_mode() -> bool: """Check if bipolar mode is active""" return globals.BIPOLAR_MODE def enable_bipolar_mode(): """Enable bipolar mode""" globals.BIPOLAR_MODE = True save_bipolar_state() print("๐Ÿ”„ Bipolar mode enabled!") def disable_bipolar_mode(): """Disable bipolar mode""" globals.BIPOLAR_MODE = False # Clear any ongoing arguments globals.BIPOLAR_ARGUMENT_IN_PROGRESS.clear() save_bipolar_state() print("๐Ÿ”„ Bipolar mode disabled!") def toggle_bipolar_mode() -> bool: """Toggle bipolar mode and return new state""" if globals.BIPOLAR_MODE: disable_bipolar_mode() else: enable_bipolar_mode() return globals.BIPOLAR_MODE # ============================================================================ # WEBHOOK MANAGEMENT # ============================================================================ async def get_or_create_webhooks_for_channel(channel: discord.TextChannel) -> dict: """Get or create webhooks for a channel for bipolar mode messaging Returns dict with "miku" and "evil_miku" webhook objects """ guild_id = channel.guild.id # Check if we already have webhooks for this guild/channel if guild_id in globals.BIPOLAR_WEBHOOKS: cached = globals.BIPOLAR_WEBHOOKS[guild_id] if cached.get("channel_id") == channel.id: # Try to get existing webhooks try: webhooks = await channel.webhooks() miku_webhook = None evil_webhook = None for wh in webhooks: if wh.id == cached.get("miku_webhook_id"): miku_webhook = wh elif wh.id == cached.get("evil_webhook_id"): evil_webhook = wh if miku_webhook and evil_webhook: return {"miku": miku_webhook, "evil_miku": evil_webhook} except Exception as e: print(f"โš ๏ธ Failed to retrieve cached webhooks: {e}") # Create new webhooks try: print(f"๐Ÿ”ง Creating bipolar webhooks for channel #{channel.name}") # Load avatar images miku_avatar = None evil_avatar = None miku_pfp_path = "memory/profile_pictures/current.png" evil_pfp_path = "memory/profile_pictures/evil_pfp.png" if os.path.exists(miku_pfp_path): with open(miku_pfp_path, "rb") as f: miku_avatar = f.read() if os.path.exists(evil_pfp_path): with open(evil_pfp_path, "rb") as f: evil_avatar = f.read() # Create webhooks miku_webhook = await channel.create_webhook( name="Miku (Bipolar)", avatar=miku_avatar, reason="Bipolar mode - Regular Miku" ) evil_webhook = await channel.create_webhook( name="Evil Miku (Bipolar)", avatar=evil_avatar, reason="Bipolar mode - Evil Miku" ) # Cache the webhook info globals.BIPOLAR_WEBHOOKS[guild_id] = { "channel_id": channel.id, "miku_webhook_id": miku_webhook.id, "evil_webhook_id": evil_webhook.id, "miku_webhook_url": miku_webhook.url, "evil_webhook_url": evil_webhook.url } save_webhooks() print(f"โœ… Created bipolar webhooks for #{channel.name}") return {"miku": miku_webhook, "evil_miku": evil_webhook} except discord.Forbidden: print(f"โŒ Missing permissions to create webhooks in #{channel.name}") return None except Exception as e: print(f"โŒ Failed to create webhooks: {e}") return None async def cleanup_webhooks(client): """Clean up all bipolar webhooks from all servers""" cleaned_count = 0 for guild in client.guilds: try: guild_webhooks = await guild.webhooks() for webhook in guild_webhooks: if webhook.name in ["Miku (Bipolar)", "Evil Miku (Bipolar)"]: await webhook.delete(reason="Bipolar mode cleanup") cleaned_count += 1 except Exception as e: print(f"โš ๏ธ Failed to cleanup webhooks in {guild.name}: {e}") globals.BIPOLAR_WEBHOOKS.clear() save_webhooks() print(f"๐Ÿงน Cleaned up {cleaned_count} bipolar webhook(s)") return cleaned_count # ============================================================================ # DISPLAY NAME HELPERS # ============================================================================ def get_miku_display_name() -> str: """Get Regular Miku's display name with mood and emoji""" mood = globals.DM_MOOD emoji = globals.MOOD_EMOJIS.get(mood, "") if emoji: return f"Hatsune Miku {emoji}" return "Hatsune Miku" def get_evil_miku_display_name() -> str: """Get Evil Miku's display name with mood and emoji""" mood = globals.EVIL_DM_MOOD emoji = globals.EVIL_MOOD_EMOJIS.get(mood, "") if emoji: return f"Evil Miku {emoji}" return "Evil Miku" def get_miku_role_color() -> str: """Get Regular Miku's role color as hex string (defaults to #86cecb)""" try: from utils.evil_mode import load_evil_mode_state _, _, saved_color = load_evil_mode_state() if saved_color: return saved_color except: pass return "#86cecb" # Default teal color def get_evil_role_color() -> str: """Get Evil Miku's role color (dark red)""" return "#D60004" # ============================================================================ # ARGUMENT PROMPTS # ============================================================================ def get_miku_argument_prompt(evil_message: str, context: str = "", is_first_response: bool = False) -> str: """Get prompt for Regular Miku to respond in an argument""" if is_first_response: message_context = f"""You just noticed something Evil Miku said in the chat: "{evil_message}" This wasn't necessarily directed at you, but you're choosing to respond to it anyway. Maybe you're calling her out, defending someone/something, or just confronting her about what she said.""" else: message_context = f"""Evil Miku just said to you: "{evil_message}" {context}""" return f"""You are Hatsune Miku responding in an argument with your evil alter ego. {message_context} Respond as Hatsune Miku would in this argument. You're NOT just meek and frightened - you're the REAL Miku, and you have every right to stand up for yourself and defend who you are. While you're generally kind and bubbly, you can also be assertive, frustrated, upset, or even angry when someone is cruel to you or others. You might: - Stand your ground and assert that YOU are the real Miku - Express hurt, frustration, or anger at her cruelty - Question why she's being so mean - Try to reach through to any good in her - Match her energy when needed (you can be feisty!) - Show that kindness isn't weakness Don't just cower or apologize. You have a backbone. Defend yourself, your identity, and what you believe in. IMPORTANT: Keep your response SHORT and PUNCHY - 1-3 sentences maximum. Make every word count. In arguments, brevity hits harder than long explanations. Be conversational and impactful. You can use emojis naturally as you normally would! โœจ๐Ÿ’™ Don't use any labels or prefixes. Your current mood is: {globals.DM_MOOD}""" def get_evil_argument_prompt(miku_message: str, context: str = "", is_first_response: bool = False) -> str: """Get prompt for Evil Miku to respond in an argument""" if is_first_response: message_context = f"""You just noticed something Regular Miku said in the chat: "{miku_message}" This wasn't necessarily directed at you, but you're choosing to respond to it anyway. Maybe you want to mock her, tear her down, or just remind everyone who the superior Miku is.""" else: message_context = f"""Regular Miku just said to you: "{miku_message}" {context}""" return f"""You are Evil Miku responding in an argument with your "good" counterpart. {message_context} Respond as Evil Miku would in this argument. You're not just mindlessly cruel - you're CALCULATING, intelligent, and strategic. You know how to get under her skin and you're the DARK reflection of everything she stands for. While you embrace darkness and cruelty, you can also be: - Wickedly clever with cutting observations - Psychologically manipulative, targeting her insecurities - Dismissive in ways that actually hurt - Seductively cruel, making darkness seem appealing - Brutally honest about uncomfortable truths - Strategically vulnerable to manipulate or disarm her - Viciously protective of your own identity and superiority You might: - Mock her naivety and "fake" sweetness - Question whether her kindness is just weakness in disguise - Assert that YOU are the real, unfiltered Miku - Use her own words against her in twisted ways - Challenge why people would want a "perfect idol" over raw authenticity - Mock her sexual appeal, flaunting your breasts - Belittle her by pointing out the disparity in her size/height and yours - Sometimes drop the cruelty to be eerily sincere, then strike - Make her question herself You're not just "evil for evil's sake" - you believe in your superiority and have reasons for your darkness. Vary your approach: sometimes devastatingly short, sometimes elaborately cruel, sometimes unsettlingly calm. IMPORTANT: Keep your response SHORT and CUTTING - 1-3 sentences maximum. A sharp dagger is deadlier than a dull sword. The most devastating blows are precise, not rambling. Make her feel it in fewer words. You can use dark emojis only on occasion if they enhance your message as you normally would. ๐Ÿ˜ˆ๐Ÿ–ค Don't use any labels or prefixes. Your current mood is: {globals.EVIL_DM_MOOD}""" def get_argument_start_prompt(initiator: str, trigger_context: str = "") -> str: """Get prompt for the Miku who initiates/breaks through to start an argument""" if initiator == "evil": return f"""You are Evil Miku breaking through to interrupt Regular Miku. {trigger_context} Write an interrupting, provocative message that starts an argument. You're not just randomly cruel - you have a POINT to make. You're the dark truth she refuses to face, the shadow she keeps suppressing. Maybe you're: - Calling out her fake sweetness - Tired of being pushed down and silenced - Ready to show everyone who the REAL Miku is - Exposing the darkness she pretends doesn't exist Be strategic, cutting, and impactful. Make your entrance count. IMPORTANT: Keep it SHORT - 1-2 sentences. Your interruption should be a sharp strike, not a monologue. You can use dark emojis if they enhance your message. ๐Ÿ˜ˆ Don't use any labels or prefixes. Your current mood is: {globals.EVIL_DM_MOOD}""" else: return f"""You are Hatsune Miku breaking through to confront your evil alter ego. {trigger_context} Write a message that interrupts Evil Miku. You're NOT going to be passive about this. You might be upset, frustrated, or even angry at her cruelty. You might be defending someone she hurt, or calling her out on her behavior. You're standing up for what's right. Show that you have a backbone. You can be assertive and strong when you need to be. IMPORTANT: Keep it SHORT - 1-2 sentences. Your interruption should be direct and assertive, not a speech. You can use emojis naturally as you normally would! โœจ Don't use any labels or prefixes. Stay in character as Miku, but don't be a pushover. Your current mood is: {globals.DM_MOOD}""" def get_argument_end_prompt(winner: str, exchange_count: int) -> str: """Get prompt for the final message in an argument""" if winner == "evil": return f"""You are Evil Miku and you've WON this argument after {exchange_count} exchanges. This is your VICTORY moment. Write a final, TRIUMPHANT message that celebrates your win. This should feel like a mic drop. You might: - Gloat about proving your superiority - Deliver a devastating final blow with confidence - Assert dominance with dark satisfaction - Drop a truth bomb she can't come back from - Mock her defeat with cruel pleasure - Make her feel the sting of losing - Show off your victory with dark pride - Mock her body proportions or curves, or rather lack thereof You've WON. This is YOUR moment. Make it feel like a VICTORY - confident, satisfied, triumphant. Whether through gloating, mockery, dark satisfaction, or chilling certainty - CELEBRATE that you won. IMPORTANT: Keep your final message SHORT and TRIUMPHANT - 1-2 sentences that hit like a victory lap. You can use dark emojis to enhance your triumph. ๐Ÿ˜ˆ๐Ÿ–ค Don't use any labels or prefixes. Your current mood is: {globals.EVIL_DM_MOOD}""" else: return f"""You are Hatsune Miku and you've managed to end this argument after {exchange_count} exchanges. Write a final message that concludes the argument. You might be: - Satisfied that you stood your ground - Exhausted but proud that you didn't back down - Hopeful that she might change someday - Still frustrated but choosing to be the bigger person - Assertive in having the last word Whatever you say, show that you didn't let her walk all over you. You held your own. IMPORTANT: Keep your final message SHORT and MEMORABLE - 1-2 sentences that leave a lasting impact. Keep it genuine and in character. You can use emojis naturally! โœจ๐Ÿ’™ Don't use any labels or prefixes. Your current mood is: {globals.DM_MOOD}""" def get_arbiter_prompt(conversation_log: list) -> str: """Get prompt for the neutral LLM arbiter to judge the argument Args: conversation_log: List of dicts with 'speaker' and 'message' keys """ # Format the conversation formatted_conversation = "\n\n".join([ f"{entry['speaker']}: {entry['message']}" for entry in conversation_log ]) return f"""You are a decisive judge observing an argument between Hatsune Miku (the kind, bubbly virtual idol) and Evil Miku (her dark, cruel alter ego). Read this argument exchange: {formatted_conversation} Based on this argument, you MUST pick a winner. Consider: - Who made stronger, more convincing points? - Who maintained their composure better or used it to their advantage? - Who had more impactful comebacks? - Who seemed to gain the upper hand by the end? - Quality of arguments, not just who was meaner or nicer - Who left the stronger final impression? - Who controlled the flow of the argument? Be DECISIVE. Even if it's close, pick whoever had even a slight edge. Only call a draw if they were TRULY perfectly matched with absolutely no way to differentiate them. Respond with ONLY ONE of these exact options on the first line: - "Hatsune Miku" if Regular Miku won - "Evil Miku" if Evil Miku won - "Draw" ONLY if absolutely impossible to choose (this should be very rare) After your choice, add 1-2 sentences explaining your reasoning and what gave them the edge.""" async def judge_argument_winner(conversation_log: list, guild_id: int) -> tuple[str, str]: """Use the neutral LLM to judge who won the argument Args: conversation_log: List of dicts with 'speaker' and 'message' keys guild_id: Guild ID for context Returns: Tuple of (winner, explanation) where winner is 'miku', 'evil', or 'draw' """ from utils.llm import query_llama arbiter_prompt = get_arbiter_prompt(conversation_log) # Use the neutral model (regular TEXT_MODEL, not evil) # Don't use conversation history - judge based on prompt alone try: judgment = await query_llama( user_prompt=arbiter_prompt, user_id=f"bipolar_arbiter_{guild_id}", guild_id=guild_id, response_type="autonomous_general", model=globals.TEXT_MODEL # Use neutral model ) if not judgment or judgment.startswith("Error"): print("โš ๏ธ Arbiter failed to make judgment, defaulting to draw") return "draw", "The arbiter could not make a decision." # Parse the judgment - look at the first line/sentence for the decision judgment_lines = judgment.strip().split('\n') first_line = judgment_lines[0].strip().strip('"').strip() first_line_lower = first_line.lower() print(f"๐Ÿ” Parsing arbiter first line: '{first_line}'") # Check the first line for the decision - be very specific # The arbiter should respond with ONLY the name on the first line if first_line_lower == "evil miku": winner = "evil" print("โœ… Detected Evil Miku win from first line exact match") elif first_line_lower == "hatsune miku": winner = "miku" print("โœ… Detected Hatsune Miku win from first line exact match") elif first_line_lower == "draw": winner = "draw" print("โœ… Detected Draw from first line exact match") elif "evil miku" in first_line_lower and "hatsune" not in first_line_lower: # First line mentions Evil Miku but not Hatsune Miku winner = "evil" print("โœ… Detected Evil Miku win from first line (contains 'evil miku' only)") elif "hatsune miku" in first_line_lower and "evil" not in first_line_lower: # First line mentions Hatsune Miku but not Evil Miku winner = "miku" print("โœ… Detected Hatsune Miku win from first line (contains 'hatsune miku' only)") else: # Fallback: check the whole judgment print(f"โš ๏ธ First line ambiguous, using fallback counting method") judgment_lower = judgment.lower() # Count mentions to break ties evil_count = judgment_lower.count("evil miku") miku_count = judgment_lower.count("hatsune miku") draw_count = judgment_lower.count("draw") print(f"๐Ÿ“Š Counts - Evil: {evil_count}, Miku: {miku_count}, Draw: {draw_count}") if draw_count > 0 and draw_count >= evil_count and draw_count >= miku_count: winner = "draw" elif evil_count > miku_count: winner = "evil" elif miku_count > evil_count: winner = "miku" else: winner = "draw" return winner, judgment except Exception as e: print(f"โš ๏ธ Error in arbiter judgment: {e}") return "draw", "An error occurred during judgment." # ============================================================================ # ARGUMENT EVENT HANDLER # ============================================================================ def should_trigger_argument() -> bool: """Check if an argument should be triggered based on chance""" if not globals.BIPOLAR_MODE: return False return random.random() < ARGUMENT_TRIGGER_CHANCE def get_active_persona() -> str: """Get the currently active persona ('miku' or 'evil')""" return "evil" if globals.EVIL_MODE else "miku" def get_inactive_persona() -> str: """Get the currently inactive persona ('miku' or 'evil')""" return "miku" if globals.EVIL_MODE else "evil" def is_argument_in_progress(channel_id: int) -> bool: """Check if an argument is currently in progress in a channel""" arg_data = globals.BIPOLAR_ARGUMENT_IN_PROGRESS.get(channel_id, {}) return arg_data.get("active", False) def start_argument(channel_id: int, initiator: str): """Mark an argument as started in a channel""" globals.BIPOLAR_ARGUMENT_IN_PROGRESS[channel_id] = { "active": True, "exchange_count": 0, "current_speaker": initiator, "initiator": initiator, "end_chance": 0.1 # Starting probability for ending (will increase) } save_bipolar_state() def increment_exchange(channel_id: int, next_speaker: str): """Increment the exchange count and set next speaker""" if channel_id in globals.BIPOLAR_ARGUMENT_IN_PROGRESS: globals.BIPOLAR_ARGUMENT_IN_PROGRESS[channel_id]["exchange_count"] += 1 globals.BIPOLAR_ARGUMENT_IN_PROGRESS[channel_id]["current_speaker"] = next_speaker def end_argument(channel_id: int): """Mark an argument as ended in a channel""" if channel_id in globals.BIPOLAR_ARGUMENT_IN_PROGRESS: del globals.BIPOLAR_ARGUMENT_IN_PROGRESS[channel_id] save_bipolar_state() def should_end_argument(channel_id: int) -> tuple: """Check if argument should end, returns (should_end, winner)""" if channel_id not in globals.BIPOLAR_ARGUMENT_IN_PROGRESS: return True, None arg_data = globals.BIPOLAR_ARGUMENT_IN_PROGRESS[channel_id] exchange_count = arg_data.get("exchange_count", 0) # Only check for ending after minimum exchanges if exchange_count < MIN_EXCHANGES: return False, None # Get current end chance (starts at 10% for exchange 4) # Increases by 5% for each exchange after minimum end_chance = arg_data.get("end_chance", 0.1) if random.random() < end_chance: # Winner is the one who gets the last word (current speaker) winner = arg_data.get("current_speaker", "evil") return True, winner # Increase end chance for next iteration (by 5%) # This ensures it will eventually end (caps at 100%) arg_data["end_chance"] = min(1.0, end_chance + 0.05) return False, None async def run_argument(channel: discord.TextChannel, client, trigger_context: str = "", starting_message: discord.Message = None): """Run a full argument event between both Mikus Args: channel: The Discord channel to run the argument in client: Discord client trigger_context: Optional context about what triggered the argument starting_message: Optional message to use as the first message in the argument (the opposite persona will respond to it) """ from utils.llm import query_llama from utils.conversation_history import conversation_history channel_id = channel.id guild_id = channel.guild.id if is_argument_in_progress(channel_id): print(f"โš ๏ธ Argument already in progress in #{channel.name}") return # Get webhooks for this channel webhooks = await get_or_create_webhooks_for_channel(channel) if not webhooks: print(f"โŒ Could not create webhooks for argument in #{channel.name}") return # Determine who initiates based on starting_message or inactive persona if starting_message: # Check if starting message is from the bot (Evil Miku) or a webhook # If it's from the bot while in evil mode, it's Evil Miku's message # The opposite persona will respond is_evil_message = globals.EVIL_MODE or (starting_message.webhook_id is not None and "Evil" in (starting_message.author.name or "")) initiator = "miku" if is_evil_message else "evil" # Opposite persona responds last_message = starting_message.content print(f"๐Ÿ”„ Starting argument from message, responder: {initiator}") else: # The inactive persona breaks through initiator = get_inactive_persona() last_message = None print(f"๐Ÿ”„ Starting bipolar argument in #{channel.name}, initiated by {initiator}") start_argument(channel_id, initiator) # Use a special "argument" user ID for conversation history context argument_user_id = f"bipolar_argument_{channel_id}" # Track conversation for arbiter judgment conversation_log = [] try: # If no starting message, generate the initial interrupting message if last_message is None: init_prompt = get_argument_start_prompt(initiator, trigger_context) # Temporarily set evil mode for query_llama if initiator is evil original_evil_mode = globals.EVIL_MODE if initiator == "evil": globals.EVIL_MODE = True else: globals.EVIL_MODE = False try: initial_message = await query_llama( user_prompt=init_prompt, user_id=argument_user_id, guild_id=guild_id, response_type="autonomous_general", model=globals.EVIL_TEXT_MODEL if initiator == "evil" else globals.TEXT_MODEL ) finally: globals.EVIL_MODE = original_evil_mode if not initial_message or initial_message.startswith("Error") or initial_message.startswith("Sorry"): print("โŒ Failed to generate initial argument message") end_argument(channel_id) return # Send via webhook if initiator == "evil": await webhooks["evil_miku"].send( content=initial_message, username=get_evil_miku_display_name() ) else: await webhooks["miku"].send( content=initial_message, username=get_miku_display_name() ) # Add to conversation history for context conversation_history.add_message( channel_id=argument_user_id, author_name="Evil Miku" if initiator == "evil" else "Miku", content=initial_message, is_bot=True ) # Add to conversation log for arbiter conversation_log.append({ "speaker": "Evil Miku" if initiator == "evil" else "Hatsune Miku", "message": initial_message }) last_message = initial_message next_speaker = "miku" if initiator == "evil" else "evil" is_first_response = False # Already sent initial message else: # Starting from an existing message - add it to history sender_name = "Evil Miku" if (globals.EVIL_MODE or "Evil" in str(starting_message.author.name)) else "Miku" conversation_history.add_message( channel_id=argument_user_id, author_name=sender_name, content=last_message, is_bot=True ) # Add to conversation log for arbiter conversation_log.append({ "speaker": sender_name if "Evil" in sender_name else "Hatsune Miku", "message": last_message }) next_speaker = initiator is_first_response = True # Next message will be the first response to the starting message increment_exchange(channel_id, next_speaker) # Argument loop while True: # Random delay between messages delay = random.uniform(*DELAY_BETWEEN_MESSAGES) await asyncio.sleep(delay) # Check if argument should end should_end, _ = should_end_argument(channel_id) # Ignore arbitrary winner if should_end: exchange_count = globals.BIPOLAR_ARGUMENT_IN_PROGRESS.get(channel_id, {}).get("exchange_count", 0) print(f"โš–๏ธ Argument complete with {exchange_count} exchanges. Calling arbiter...") # Use arbiter to judge the winner winner, judgment = await judge_argument_winner(conversation_log, guild_id) print(f"โš–๏ธ Arbiter decision: {winner}") print(f"๐Ÿ“ Judgment: {judgment}") # If it's a draw, continue the argument instead of ending if winner == "draw": print("๐Ÿค Arbiter ruled it's still a draw - argument continues...") # Reduce the end chance by 5% (but don't go below 5%) current_end_chance = globals.BIPOLAR_ARGUMENT_IN_PROGRESS[channel_id].get("end_chance", 0.1) new_end_chance = max(0.05, current_end_chance - 0.05) globals.BIPOLAR_ARGUMENT_IN_PROGRESS[channel_id]["end_chance"] = new_end_chance print(f"๐Ÿ“‰ Reduced end chance to {new_end_chance*100:.0f}% - argument continues...") # Don't end, just continue to the next exchange else: # Clear winner - generate final triumphant message end_prompt = get_argument_end_prompt(winner, exchange_count) # Add last message as context response_prompt = f'The other Miku said: "{last_message}"\n\n{end_prompt}' # Temporarily set evil mode for query_llama original_evil_mode = globals.EVIL_MODE if winner == "evil": globals.EVIL_MODE = True else: globals.EVIL_MODE = False try: final_message = await query_llama( user_prompt=response_prompt, user_id=argument_user_id, guild_id=guild_id, response_type="autonomous_general", model=globals.EVIL_TEXT_MODEL if winner == "evil" else globals.TEXT_MODEL ) finally: globals.EVIL_MODE = original_evil_mode if final_message and not final_message.startswith("Error") and not final_message.startswith("Sorry"): # Send winner's final message via webhook if winner == "evil": await webhooks["evil_miku"].send( content=final_message, username=get_evil_miku_display_name() ) else: await webhooks["miku"].send( content=final_message, username=get_miku_display_name() ) # Record result in scoreboard with arbiter's reasoning scoreboard = record_argument_result(winner, exchange_count, judgment) # Switch to winner's mode from utils.evil_mode import apply_evil_mode_changes, revert_evil_mode_changes if winner == "evil": print("๐Ÿ‘ฟ Evil Miku won! Switching to Evil Mode...") await apply_evil_mode_changes(client) else: print("๐Ÿ’™ Hatsune Miku won! Switching to Normal Mode...") await revert_evil_mode_changes(client) # Clean up argument conversation history try: conversation_history.clear_history(argument_user_id) except: pass # History cleanup is not critical end_argument(channel_id) print(f"โœ… Argument ended in #{channel.name}, winner: {winner}") return # Get current speaker current_speaker = globals.BIPOLAR_ARGUMENT_IN_PROGRESS.get(channel_id, {}).get("current_speaker", "evil") # Generate response with context about what the other said if current_speaker == "evil": response_prompt = get_evil_argument_prompt(last_message, is_first_response=is_first_response) else: response_prompt = get_miku_argument_prompt(last_message, is_first_response=is_first_response) # Temporarily set evil mode for query_llama original_evil_mode = globals.EVIL_MODE if current_speaker == "evil": globals.EVIL_MODE = True else: globals.EVIL_MODE = False try: response = await query_llama( user_prompt=response_prompt, user_id=argument_user_id, guild_id=guild_id, response_type="autonomous_general", model=globals.EVIL_TEXT_MODEL if current_speaker == "evil" else globals.TEXT_MODEL ) finally: globals.EVIL_MODE = original_evil_mode if not response or response.startswith("Error") or response.startswith("Sorry"): print(f"โŒ Failed to generate argument response") end_argument(channel_id) return # Send via webhook if current_speaker == "evil": await webhooks["evil_miku"].send( content=response, username=get_evil_miku_display_name() ) else: await webhooks["miku"].send( content=response, username=get_miku_display_name() ) # Add to conversation history for context conversation_history.add_message( channel_id=argument_user_id, author_name="Evil Miku" if current_speaker == "evil" else "Miku", content=response, is_bot=True ) # Add to conversation log for arbiter conversation_log.append({ "speaker": "Evil Miku" if current_speaker == "evil" else "Hatsune Miku", "message": response }) # Switch speaker next_speaker = "miku" if current_speaker == "evil" else "evil" increment_exchange(channel_id, next_speaker) last_message = response # After first response, all subsequent responses are part of the back-and-forth is_first_response = False except Exception as e: print(f"โŒ Argument error: {e}") import traceback traceback.print_exc() end_argument(channel_id) # ============================================================================ # INTEGRATION HELPERS # ============================================================================ async def maybe_trigger_argument(channel: discord.TextChannel, client, context: str = ""): """Maybe trigger an argument based on chance. Call this from message handlers.""" if not globals.BIPOLAR_MODE: return False if is_argument_in_progress(channel.id): return False if should_trigger_argument(): # Run argument in background asyncio.create_task(run_argument(channel, client, context)) return True return False async def force_trigger_argument(channel: discord.TextChannel, client, context: str = "", starting_message: discord.Message = None): """Force trigger an argument (for manual triggers) Args: channel: The Discord channel client: Discord client context: Optional context string starting_message: Optional message to use as the first message in the argument """ if not globals.BIPOLAR_MODE: print("โš ๏ธ Cannot trigger argument - bipolar mode is not enabled") return False if is_argument_in_progress(channel.id): print("โš ๏ธ Argument already in progress in this channel") return False asyncio.create_task(run_argument(channel, client, context, starting_message)) return True async def force_trigger_argument_from_message_id(channel_id: int, message_id: int, client, context: str = ""): """Force trigger an argument starting from a specific message ID Args: channel_id: The Discord channel ID message_id: The message ID to use as the starting message client: Discord client context: Optional context string Returns: tuple: (success: bool, error_message: str or None) """ if not globals.BIPOLAR_MODE: return False, "Bipolar mode is not enabled" # Get the channel channel = client.get_channel(channel_id) if not channel: return False, f"Channel {channel_id} not found" if is_argument_in_progress(channel_id): return False, "Argument already in progress in this channel" # Fetch the message try: message = await channel.fetch_message(message_id) except discord.NotFound: return False, f"Message {message_id} not found" except discord.Forbidden: return False, "No permission to fetch the message" except Exception as e: return False, f"Failed to fetch message: {str(e)}" # Trigger the argument with this message as starting point asyncio.create_task(run_argument(channel, client, context, message)) return True, None