Загрузить файлы в «cogs/status_rotator»

cogs/status_rotator/config.json

@@ -0,0 +1,5 @@
+{
+  "description": "Rotates bot statuses",
+  "usages": [],
+  "hidden": true
+}

cogs/status_rotator/status_rotator.py

@@ -0,0 +1,108 @@
+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))

cogs/status_rotator/statuses.json

@@ -0,0 +1,13 @@
+{
+  "interval_minutes": 1.0,
+  "statuses": [
+    {
+      "type": "playing",
+      "text": "Example 1"
+    },
+    {
+        "type": "listening",
+        "text": "Example 2"
+    }
+  ]
+}