feat: add activities.py module for mood-based Discord presence

New module that loads activities.yaml and provides:
- Weighted random activity selection per mood
- Discord presence update (Listening/Playing)
- File mtime caching for hot-reload
- Validation for CRUD operations
- Fallback for moods with no activities defined
This commit is contained in:
2026-04-24 13:30:54 +03:00
parent e30316f383
commit a5916645df

178
bot/utils/activities.py Normal file
View File

@@ -0,0 +1,178 @@
# utils/activities.py
"""
Mood-based Discord activity status system.
Loads activity definitions from activities.yaml and provides functions to:
- Pick a weighted-random activity for a given mood
- Update the bot's Discord presence (Listening/Playing)
- Get/set activity data for the Web UI
"""
import os
import random
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")
# Cache: (data_dict, file_mtime)
_activities_cache = None
_cache_mtime = 0.0
def _load_activities(force=False):
"""Load activities.yaml with file-mtime-based caching.
Returns the full dict: {"normal": {...}, "evil": {...}}
"""
global _activities_cache, _cache_mtime
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 _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}")
return data
except Exception as e:
logger.error(f"Failed to load activities file: {e}")
return _activities_cache or {"normal": {}, "evil": {}}
def save_activities(data: dict):
"""Write the full activities dict back to YAML."""
global _activities_cache, _cache_mtime
try:
with open(ACTIVITIES_FILE, "w", encoding="utf-8") as f:
yaml.dump(data, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
# Update cache immediately
_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
def get_all_activities() -> dict:
"""Return the full activities dict (normal + evil sections)."""
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}
Raises:
ValueError: if validation fails
"""
# Validate
valid_types = {"listening", "playing"}
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_types:
raise ValueError(f"Entry {i} has invalid type '{entry.get('type')}', must be 'listening' or 'playing'")
if not entry.get("name") or not isinstance(entry["name"], str):
raise ValueError(f"Entry {i} must have a non-empty string 'name'")
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")
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)
def pick_activity_for_mood(mood_name: str, is_evil: bool = False):
"""Pick a weighted-random activity for a mood.
Returns:
tuple: (activity_type, name) e.g. ("listening", "World is Mine")
Returns ("listening", "Vocaloid") as fallback if mood has no entries.
"""
activities = get_activities_for_mood(mood_name, is_evil)
if not activities:
logger.debug(f"No activities defined for {'evil/' if is_evil else ''}{mood_name}, using fallback")
return ("listening", "Vocaloid")
# Weighted random selection
weights = [entry.get("weight", 1) for entry in activities]
chosen = random.choices(activities, weights=weights, k=1)[0]
return (chosen["type"], chosen["name"])
async def update_bot_presence(mood_name: str, is_evil: bool = False):
"""Update the bot's Discord presence based on the current mood.
- asleep: shows idle status, no activity
- Other moods: shows "Listening to..." or "Playing..." with weighted-random pick
"""
if not globals.client or not globals.client.is_ready():
logger.debug("Bot not ready, skipping presence update")
return
try:
if mood_name == "asleep":
# While asleep: idle status, no activity
await globals.client.change_presence(
status=discord.Status.idle,
activity=None
)
logger.info("Set presence: idle (asleep)")
return
activity_type, name = pick_activity_for_mood(mood_name, is_evil)
if activity_type == "listening":
activity = discord.Activity(type=discord.ActivityType.listening, name=name)
log_label = f"Listening to {name}"
else:
activity = discord.Game(name=name)
log_label = f"Playing {name}"
await globals.client.change_presence(activity=activity)
logger.info(f"Set presence: {log_label} (mood={'evil/' if is_evil else ''}{mood_name})")
except Exception as e:
logger.error(f"Failed to update bot presence: {e}")
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:
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}")