# utils/activities.py """ Mood-based Discord activity status system. Activity display is driven by the autonomous engine's mood energy profiles: - High-energy moods (excited, bubbly) → almost always show an activity - Low-energy moods (sleepy, melancholy) → mostly idle, occasionally active - Manual override via Web UI bypasses automatic behavior Supports 5 activity types: listening, playing, watching, competing, streaming. """ import os import random import tempfile import threading import time import yaml import discord import globals from utils.logger import get_logger logger = get_logger('activity') ACTIVITIES_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), "activities.yaml") # Discord activity name character limit DISCORD_ACTIVITY_NAME_MAX = 128 # All valid activity types VALID_ACTIVITY_TYPES = {"listening", "playing", "watching", "competing", "streaming"} # ── Activity probability per mood (derived from autonomous engine energy profiles) ── # Value = probability that the bot WILL have an activity (vs being idle). ACTIVITY_PROBABILITY = { # Normal moods "asleep": 0.00, "sleepy": 0.15, "melancholy": 0.25, "shy": 0.30, "irritated": 0.40, "neutral": 0.45, "serious": 0.50, "romantic": 0.55, "curious": 0.60, "angry": 0.60, "flirty": 0.65, "silly": 0.75, "bubbly": 0.80, "excited": 0.85, # Evil moods "melancholic": 0.25, "bored": 0.35, "contemptuous": 0.45, "evil_neutral": 0.50, "sarcastic": 0.55, "jealous": 0.60, "cunning": 0.65, "aggressive": 0.70, "playful_cruel": 0.70, "manic": 0.85, } # ── Thread lock for all shared mutable state ── _state_lock = threading.Lock() # ── Manual override state ── _manual_override = False _manual_override_until = 0.0 # Unix timestamp; 0 = no override MANUAL_OVERRIDE_DURATION = 1800 # 30 minutes # ── Current activity tracking ── _current_activity = None # dict: {type, name, state, url} or None # Cache: (data_dict, file_mtime) _activities_cache = None _cache_mtime = 0.0 # ══════════════════════════════════════════════════════════════════════════════ # YAML Loading / Saving # ══════════════════════════════════════════════════════════════════════════════ def _load_activities(force=False): """Load activities.yaml with file-mtime-based caching. Returns a deep copy.""" global _activities_cache, _cache_mtime with _state_lock: try: mtime = os.path.getmtime(ACTIVITIES_FILE) except OSError: logger.warning(f"Activities file not found: {ACTIVITIES_FILE}") return {"normal": {}, "evil": {}} if not force and _activities_cache is not None and mtime == _cache_mtime: # Return a deep copy so callers cannot mutate the cache import copy return copy.deepcopy(_activities_cache) try: with open(ACTIVITIES_FILE, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} _activities_cache = data _cache_mtime = mtime logger.debug(f"Loaded activities from {ACTIVITIES_FILE}") import copy return copy.deepcopy(data) except Exception as e: logger.error(f"Failed to load activities file: {e}") if _activities_cache is not None: import copy return copy.deepcopy(_activities_cache) return {"normal": {}, "evil": {}} def save_activities(data: dict): """Write the full activities dict back to YAML using atomic write (temp + rename).""" global _activities_cache, _cache_mtime with _state_lock: try: # Atomic write: write to temp file in same directory, then rename dir_name = os.path.dirname(ACTIVITIES_FILE) fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".yaml.tmp") try: with os.fdopen(fd, "w", encoding="utf-8") as f: yaml.dump(data, f, default_flow_style=False, allow_unicode=True, sort_keys=False) os.replace(tmp_path, ACTIVITIES_FILE) except BaseException: # Clean up temp file on failure try: os.unlink(tmp_path) except OSError: pass raise _activities_cache = data _cache_mtime = os.path.getmtime(ACTIVITIES_FILE) logger.info(f"Saved activities to {ACTIVITIES_FILE}") except Exception as e: logger.error(f"Failed to save activities file: {e}") raise # ══════════════════════════════════════════════════════════════════════════════ # CRUD for activity data (used by Web UI) # ══════════════════════════════════════════════════════════════════════════════ def get_all_activities() -> dict: """Return the full activities dict (normal + evil sections). Returns a deep copy.""" return _load_activities() def get_activities_for_mood(mood_name: str, is_evil: bool = False) -> list: """Return the activity list for a specific mood. Returns empty list if not found.""" section = "evil" if is_evil else "normal" data = _load_activities() return data.get(section, {}).get(mood_name, []) def set_activities_for_mood(mood_name: str, is_evil: bool, activities: list): """Validate and save updated activity list for a mood. Args: mood_name: mood key (e.g. "bubbly", "aggressive") is_evil: True for evil section, False for normal activities: list of dicts with keys {type, name, weight, [state], [url]} Raises: ValueError: if validation fails """ for i, entry in enumerate(activities): if not isinstance(entry, dict): raise ValueError(f"Entry {i} must be a dict, got {type(entry).__name__}") if entry.get("type") not in VALID_ACTIVITY_TYPES: raise ValueError( f"Entry {i} has invalid type '{entry.get('type')}', " f"must be one of: {', '.join(sorted(VALID_ACTIVITY_TYPES))}" ) if not entry.get("name") or not isinstance(entry["name"], str): raise ValueError(f"Entry {i} must have a non-empty string 'name'") if len(entry["name"]) > DISCORD_ACTIVITY_NAME_MAX: raise ValueError(f"Entry {i} name exceeds {DISCORD_ACTIVITY_NAME_MAX} characters") if not isinstance(entry.get("weight", 0), int) or entry.get("weight", 0) < 1: raise ValueError(f"Entry {i} weight must be a positive integer") if "state" in entry and entry["state"] is not None and not isinstance(entry["state"], str): raise ValueError(f"Entry {i} 'state' must be a string if provided") if "url" in entry and entry["url"] is not None and not isinstance(entry["url"], str): raise ValueError(f"Entry {i} 'url' must be a string if provided") if entry.get("type") == "streaming" and not entry.get("url"): raise ValueError(f"Entry {i} is streaming type but has no url") section = "evil" if is_evil else "normal" data = _load_activities() if section not in data: data[section] = {} data[section][mood_name] = activities save_activities(data) # ══════════════════════════════════════════════════════════════════════════════ # Activity Selection # ══════════════════════════════════════════════════════════════════════════════ def pick_activity_for_mood(mood_name: str, is_evil: bool = False): """Pick a weighted-random activity for a mood. Validates entries and skips malformed ones with a warning. Returns: dict: {"type": ..., "name": ..., "state": ..., "url": ...} state and url may be None. Returns None if mood has no valid entries. """ activities = get_activities_for_mood(mood_name, is_evil) if not activities: return None # Validate entries, skipping malformed ones valid = [] weights = [] for i, entry in enumerate(activities): if not isinstance(entry, dict): logger.warning(f"Skipping non-dict entry {i} in {'evil/' if is_evil else ''}{mood_name}") continue if "type" not in entry or "name" not in entry: logger.warning(f"Skipping entry {i} missing 'type' or 'name' in {'evil/' if is_evil else ''}{mood_name}: {entry}") continue if entry["type"] not in VALID_ACTIVITY_TYPES: logger.warning(f"Skipping entry {i} with unrecognized type '{entry['type']}' in {'evil/' if is_evil else ''}{mood_name}") continue w = entry.get("weight", 1) if not isinstance(w, int) or w < 1: logger.warning(f"Skipping entry {i} with invalid weight {w} in {'evil/' if is_evil else ''}{mood_name}") continue valid.append(entry) weights.append(w) if not valid: logger.warning(f"No valid entries for {'evil/' if is_evil else ''}{mood_name}") return None chosen = random.choices(valid, weights=weights, k=1)[0] return { "type": chosen["type"], "name": chosen["name"], "state": chosen.get("state"), "url": chosen.get("url"), } def should_have_activity(mood_name: str) -> bool: """Decide whether the bot should show an activity for this mood. Based on mood energy: high-energy moods are more likely to be active, low-energy moods are more likely to be idle. """ probability = ACTIVITY_PROBABILITY.get(mood_name, 0.45) return random.random() < probability # ══════════════════════════════════════════════════════════════════════════════ # Manual Override # ══════════════════════════════════════════════════════════════════════════════ def is_manual_override_active() -> bool: """Check if a manual override is in effect (hasn't expired). Thread-safe.""" with _state_lock: global _manual_override if not _manual_override: return False if _manual_override_until > 0 and time.time() > _manual_override_until: _manual_override = False logger.info("Manual override expired, returning to automatic mode") return False return True def set_manual_override(duration: int = MANUAL_OVERRIDE_DURATION): """Activate manual override for the given duration (seconds). Thread-safe.""" with _state_lock: global _manual_override, _manual_override_until _manual_override = True expiry = time.time() + duration _manual_override_until = expiry logger.info(f"Manual override activated for {duration}s (expires at {time.strftime('%H:%M:%S', time.localtime(expiry))})") def clear_manual_override(): """Deactivate manual override immediately. Thread-safe.""" with _state_lock: global _manual_override, _manual_override_until _manual_override = False _manual_override_until = 0.0 logger.info("Manual override cleared") # ══════════════════════════════════════════════════════════════════════════════ # Current Activity Tracking # ══════════════════════════════════════════════════════════════════════════════ def get_current_activity(): """Return the current activity dict or None if idle. Thread-safe.""" with _state_lock: return _current_activity def _set_current_activity(activity_dict): """Update the tracked current activity. Thread-safe.""" global _current_activity with _state_lock: _current_activity = activity_dict # ══════════════════════════════════════════════════════════════════════════════ # Discord Presence Updates # ══════════════════════════════════════════════════════════════════════════════ def _build_activity(payload: dict): """Build a discord.Activity (or discord.Streaming) from a payload dict. Logs a warning if the activity type is unrecognized (falls back to playing). """ atype = payload["type"] name = payload["name"] state = payload.get("state") or None # normalize empty string to None url = payload.get("url") if atype == "streaming" and url: return discord.Streaming(name=name, url=url) type_map = { "listening": discord.ActivityType.listening, "playing": discord.ActivityType.playing, "watching": discord.ActivityType.watching, "competing": discord.ActivityType.competing, "streaming": discord.ActivityType.streaming, # fallback without url } resolved_type = type_map.get(atype) if resolved_type is None: logger.warning(f"Unrecognized activity type '{atype}', falling back to 'playing'") resolved_type = discord.ActivityType.playing return discord.Activity( type=resolved_type, name=name, state=state, ) def _activity_label(payload: dict) -> str: """Human-readable label for logging.""" atype = payload["type"] name = payload["name"] prefixes = { "listening": "Listening to", "playing": "Playing", "watching": "Watching", "competing": "Competing in", "streaming": "Streaming", } label = f"{prefixes.get(atype, 'Playing')} {name}" state = payload.get("state") if state: label += f" ({state})" return label async def update_bot_presence(mood_name: str, is_evil: bool = False, force: bool = False): """Update the bot's Discord presence based on the current mood. - asleep: idle status, no activity - Manual override active: skip (unless force=True) - Energy-based probability: may choose to be idle instead of showing an activity - force=True bypasses both manual override and probability (used by on_ready and manual set) Args: mood_name: current mood key is_evil: whether evil mode is active force: bypass manual override and probability checks """ if not globals.client or not globals.client.is_ready(): logger.debug("Bot not ready, skipping presence update") return try: # asleep → always idle if mood_name == "asleep": _set_current_activity(None) await globals.client.change_presence( status=discord.Status.idle, activity=None ) logger.info("Set presence: idle (asleep)") return # Check manual override (skip unless forced) if not force and is_manual_override_active(): logger.debug("Manual override active, skipping automatic presence update") return # Energy-based probability: should we show an activity at all? if not force and not should_have_activity(mood_name): await clear_bot_presence() logger.info(f"Decided to be idle (mood={'evil/' if is_evil else ''}{mood_name})") return # Pick a random activity for this mood chosen = pick_activity_for_mood(mood_name, is_evil) if not chosen: # No activities defined for this mood → idle await clear_bot_presence() logger.info(f"No activities for {'evil/' if is_evil else ''}{mood_name}, staying idle") return activity = _build_activity(chosen) label = _activity_label(chosen) _set_current_activity(chosen) await globals.client.change_presence(status=discord.Status.online, activity=activity) logger.info(f"Set presence: {label} (mood={'evil/' if is_evil else ''}{mood_name})") except Exception as e: logger.error(f"Failed to update bot presence: {e}", exc_info=True) async def set_activity_manual(activity_type: str, name: str, state: str = None, url: str = None): """Manually set the bot's activity (bypasses mood system). Raises: ValueError: if activity_type is invalid, name too long, or streaming lacks url RuntimeError: if bot is not ready """ if activity_type not in VALID_ACTIVITY_TYPES: raise ValueError(f"Invalid type '{activity_type}', must be one of: {', '.join(sorted(VALID_ACTIVITY_TYPES))}") if not name or not isinstance(name, str): raise ValueError("name must be a non-empty string") if len(name) > DISCORD_ACTIVITY_NAME_MAX: raise ValueError(f"name exceeds {DISCORD_ACTIVITY_NAME_MAX} characters") if activity_type == "streaming" and not url: raise ValueError("streaming type requires a url") if not globals.client or not globals.client.is_ready(): raise RuntimeError("Bot is not ready") payload = {"type": activity_type, "name": name, "state": state, "url": url} activity = _build_activity(payload) label = _activity_label(payload) _set_current_activity(payload) set_manual_override() await globals.client.change_presence(status=discord.Status.online, activity=activity) logger.info(f"Set presence (manual): {label}") async def clear_bot_presence(): """Clear the bot's activity (set to online with no activity).""" if not globals.client or not globals.client.is_ready(): return try: _set_current_activity(None) await globals.client.change_presence(status=discord.Status.online, activity=None) logger.info("Cleared bot presence") except Exception as e: logger.error(f"Failed to clear bot presence: {e}") async def clear_activity_manual(): """Manually clear the bot's activity and activate manual override.""" set_manual_override() await clear_bot_presence() logger.info("Cleared presence (manual override)") async def release_manual_override(): """Release manual override and immediately recalculate presence from current mood. Uses force=True so the bot always gets an activity instead of potentially going idle right away (which would be confusing UX after clicking "Return to Auto"). """ clear_manual_override() try: if globals.EVIL_MODE: mood = globals.EVIL_DM_MOOD is_evil = True else: mood = globals.DM_MOOD is_evil = False await update_bot_presence(mood, is_evil=is_evil, force=True) logger.info(f"Released manual override, set presence for mood={'evil/' if is_evil else ''}{mood}") except Exception as e: logger.error(f"Failed to recalculate presence after releasing override: {e}")