Koko210/miku-discord
1
0
Fork 0
Code Issues 34 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
20891179eee237acb940828ccdefd323cdaba67f
miku-discord/bot/utils/activities.py
koko210Serve 6080fe170f Fix all activity system edge cases 
Critical fixes:
- Add threading.Lock for all shared mutable state (override, cache, current activity)
- Atomic YAML writes (temp file + os.replace) to prevent corruption on crash
- Deep-copy cache on reads to prevent callers from mutating shared state

High-severity fixes:
- Validate entries in pick_activity_for_mood() — skip/log malformed instead of KeyError
- Log warning on unrecognized activity type fallback
- Normalize empty-string state to None (avoid 'None' display)
- release_manual_override() now uses force=True so bot always shows activity
- Add try/except in release_manual_override() to handle failures gracefully

Medium fixes:
- Remove dead 'test' mood from activities.yaml
- Validate name length (128 char Discord limit) in CRUD and manual set
- Validate streaming entries have URL in CRUD path
- Add JSON parse error handling in API routes
- on_ready preserves active manual override instead of overwriting
- Log override expiry timestamp (HH:MM:SS) for easier debugging
- exc_info=True on presence update errors for full stack traces

Low fixes:
- JS activitySetFromEntry() shows notification on parse error
2026-04-28 00:18:25 +03:00

493 lines
20 KiB
Python
Raw Blame History

# 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}")
Reference in New Issue View Git Blame Copy Permalink