forked from rejnronuz/muzovkant-cogdump
108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
import discord
|
|
from discord.ext import commands, tasks
|
|
import json
|
|
import random
|
|
import logging
|
|
from typing import Optional, Dict, Any
|
|
import asyncio
|
|
import os
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class StatusRotator(commands.Cog):
|
|
def __init__(self, bot, *, status_file: str = os.path.join(os.path.dirname(__file__), 'statuses.json')):
|
|
self.bot = bot
|
|
self.statuses: list[Dict[str, Any]] = []
|
|
self.current_index = 0
|
|
self.status_file = status_file
|
|
|
|
async def cog_load(self):
|
|
await self.load_statuses()
|
|
asyncio.ensure_future(self._startup())
|
|
|
|
async def _startup(self):
|
|
await self.bot.wait_until_ready()
|
|
self.rotate_status.start()
|
|
|
|
async def cog_unload(self):
|
|
self.rotate_status.cancel()
|
|
|
|
async def load_statuses(self):
|
|
try:
|
|
with open(self.status_file, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
self.statuses = data.get('statuses', [])
|
|
|
|
interval = data.get('interval_minutes', 1.0)
|
|
self.rotate_status.change_interval(minutes=interval)
|
|
|
|
logger.info("Loaded %d statuses with a %.1f minute interval.", len(self.statuses), interval)
|
|
except FileNotFoundError:
|
|
logger.error("Status file not found: %s", self.status_file)
|
|
except json.JSONDecodeError:
|
|
logger.error("Failed to parse JSON in: %s", self.status_file)
|
|
|
|
def get_next_status(self) -> Optional[Dict[str, Any]]:
|
|
if not self.statuses:
|
|
return None
|
|
status = self.statuses[self.current_index]
|
|
self.current_index = (self.current_index + 1) % len(self.statuses)
|
|
return status
|
|
|
|
def get_random_status(self) -> Optional[Dict[str, Any]]:
|
|
if not self.statuses:
|
|
return None
|
|
return random.choice(self.statuses)
|
|
|
|
def create_activity(self, status_data: Dict[str, Any]) -> discord.Activity:
|
|
"""Helper to convert JSON data into a discord.Activity object."""
|
|
act_type_str = status_data.get('type', 'playing').lower()
|
|
text = status_data.get('text', 'Unknown Status')
|
|
|
|
activity_types = {
|
|
'playing': discord.ActivityType.playing,
|
|
'watching': discord.ActivityType.watching,
|
|
'listening': discord.ActivityType.listening,
|
|
'streaming': discord.ActivityType.streaming,
|
|
'competing': discord.ActivityType.competing,
|
|
'custom': discord.ActivityType.custom
|
|
}
|
|
|
|
act_type = activity_types.get(act_type_str, discord.ActivityType.playing)
|
|
|
|
# streaming requires a url parameter to show up correctly
|
|
if act_type == discord.ActivityType.streaming:
|
|
url = status_data.get('url', 'https://twitch.tv/twitch')
|
|
return discord.Activity(type=act_type, name=text, url=url)
|
|
|
|
return discord.Activity(type=act_type, name=text)
|
|
|
|
async def update_status(self, status_data: Optional[Dict[str, Any]] = None):
|
|
if not self.statuses:
|
|
logger.warning("No statuses loaded, skipping update")
|
|
return
|
|
|
|
if status_data is None:
|
|
status_data = self.get_next_status()
|
|
|
|
activity = self.create_activity(status_data)
|
|
|
|
try:
|
|
await self.bot.change_presence(activity=activity)
|
|
logger.debug("Status updated to: [%s] %s", activity.type.name, activity.name)
|
|
except Exception as e:
|
|
logger.error("Failed to update status: %s", e)
|
|
|
|
|
|
@tasks.loop(minutes=1.0)
|
|
async def rotate_status(self):
|
|
await self.update_status()
|
|
|
|
@rotate_status.before_loop
|
|
async def before_rotate_status(self):
|
|
await self.bot.wait_until_ready()
|
|
|
|
|
|
async def setup(bot):
|
|
await bot.add_cog(StatusRotator(bot)) |