Koko210/miku-discord
1
0
Fork 0
Code Issues 34 Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor activity system: energy-based probability, manual override, all 5 activity types

Browse Source 
- Rewrite utils/activities.py with mood energy-driven activity probability
  (high-energy moods like excited/bubbly show activity ~80-85% of the time,
  low-energy moods like sleepy/melancholy only ~15-25%)
- Add manual override system with 30-min auto-expiry for Web UI control
- Support all 5 Discord activity types: listening, playing, watching,
  competing, streaming (with purple LIVE badge via discord.Streaming)
- Add current activity tracking (get_current_activity)
- Add force=True param to update_bot_presence for on_ready (bot.py)
- Add 4 new API routes for manual override:
  GET/POST/DELETE /activities/current, POST /activities/current/auto
- Expand activities.yaml from 139 to 157 entries, adding watching,
  competing, and streaming entries across 11 moods
- Update Web UI: activity type dropdown with all 5 types, conditional
  URL field for streaming, 'Current Activity' override panel with
  set/clear/auto controls, type-aware icons and labels
This commit is contained in:
koko210Serve
2026-04-27 23:39:18 +03:00
parent 9bc618b526
commit d6cdb89e42
5 changed files with 556 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

306
bot/utils/activities.py
View File

@@ -2,14 +2,17 @@
"""
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
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 time
import yaml
import discord
import globals
@@ -19,16 +22,59 @@ logger = get_logger('activity')
ACTIVITIES_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), "activities.yaml")
# 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,
}
# ── 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 the full dict: {"normal": {...}, "evil": {...}}
"""
"""Load activities.yaml with file-mtime-based caching."""
global _activities_cache, _cache_mtime
try:
@@ -60,7 +106,6 @@ def save_activities(data: dict):
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}")
@@ -69,6 +114,10 @@ def save_activities(data: dict):
raise
# ══════════════════════════════════════════════════════════════════════════════
# CRUD for activity data (used by Web UI)
# ══════════════════════════════════════════════════════════════════════════════
def get_all_activities() -> dict:
"""Return the full activities dict (normal + evil sections)."""
return _load_activities()
@@ -83,28 +132,31 @@ def get_activities_for_mood(mood_name: str, is_evil: bool = False) -> list:
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}
activities: list of dicts with keys {type, name, weight, [state], [url]}
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 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 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")
section = "evil" if is_evil else "normal"
data = _load_activities()
@@ -114,38 +166,157 @@ def set_activities_for_mood(mood_name: str, is_evil: bool, activities: list):
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.
Returns:
tuple: (activity_type, name, state) e.g. ("listening", "World is Mine", "by ryo (supercell)")
state may be None if not defined. Fallback: ("listening", "Vocaloid", None)
dict: {"type": ..., "name": ..., "state": ..., "url": ...}
state and url may be None.
Returns None 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", None)
return None
# 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"], chosen.get("state"))
return {
"type": chosen["type"],
"name": chosen["name"],
"state": chosen.get("state"),
"url": chosen.get("url"),
}
async def update_bot_presence(mood_name: str, is_evil: bool = False):
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)."""
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)."""
global _manual_override, _manual_override_until
_manual_override = True
_manual_override_until = time.time() + duration
logger.info(f"Manual override activated for {duration}s")
def clear_manual_override():
"""Deactivate manual override immediately."""
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."""
return _current_activity
def _set_current_activity(activity_dict):
"""Update the tracked current activity."""
global _current_activity
_current_activity = activity_dict
# ══════════════════════════════════════════════════════════════════════════════
# Discord Presence Updates
# ══════════════════════════════════════════════════════════════════════════════
def _build_activity(payload: dict):
"""Build a discord.Activity (or discord.Streaming) from a payload dict."""
atype = payload["type"]
name = payload["name"]
state = payload.get("state")
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
}
return discord.Activity(
type=type_map.get(atype, discord.ActivityType.playing),
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: shows idle status, no activity
- Other moods: shows "Listening to..." or "Playing..." with weighted-random pick
- 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":
# While asleep: idle status, no activity
_set_current_activity(None)
await globals.client.change_presence(
status=discord.Status.idle,
activity=None
@@ -153,31 +324,90 @@ async def update_bot_presence(mood_name: str, is_evil: bool = False):
logger.info("Set presence: idle (asleep)")
return
activity_type, name, state = pick_activity_for_mood(mood_name, is_evil)
# Check manual override (skip unless forced)
if not force and is_manual_override_active():
logger.debug("Manual override active, skipping automatic presence update")
return
if activity_type == "listening":
activity = discord.Activity(type=discord.ActivityType.listening, name=name, state=state)
log_label = f"Listening to {name}"
else:
activity = discord.Activity(type=discord.ActivityType.playing, name=name, state=state)
log_label = f"Playing {name}"
# 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
if state:
log_label += f" ({state})"
# 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: {log_label} (mood={'evil/' if is_evil else ''}{mood_name})")
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}")
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 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 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."""
clear_manual_override()
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=False)
logger.info(f"Released manual override, recalculated for mood={'evil/' if is_evil else ''}{mood}")