4f57cccea5
Enhance the bot's architecture by introducing sharding capabilities, allowing for better scalability and performance. Update environment files to include sharding options, and modify the bot's core logic to handle shard connections and latencies. Implement shard-specific logging and API synchronization, ensuring that only the primary shard performs certain tasks. Update API schemas and routes to reflect shard information, improving the overall status reporting and monitoring of the bot's performance across multiple shards.
150 lines
6.4 KiB
Python
150 lines
6.4 KiB
Python
# ╔══════════════════════════════════════════════════════════════════╗
|
|
# ║ ║
|
|
# ║ +-+-+-+-+-+-+-+-+ ║
|
|
# ║ |H|e|x|a|H|o|s|t| ║
|
|
# ║ +-+-+-+-+-+-+-+-+ ║
|
|
# ║ ║
|
|
# ║ © 2026 HexaHost — All Rights Reserved ║
|
|
# ║ ║
|
|
# ║ discord ── https://discord.gg/hexahost ║
|
|
# ║ github ── https://github.com/theoneandonlymace ║
|
|
# ║ ║
|
|
# ╚══════════════════════════════════════════════════════════════════╝
|
|
|
|
from __future__ import annotations
|
|
from utils.db_paths import db_path
|
|
from discord.ext import commands, tasks
|
|
import discord
|
|
import aiohttp
|
|
import typing
|
|
from typing import List
|
|
import aiosqlite
|
|
from utils.config import OWNER_IDS, BotName, get_shard_kwargs
|
|
from utils import getConfig, updateConfig
|
|
from utils.Tools import setup_db
|
|
from .Context import Context
|
|
from colorama import Fore, Style, init
|
|
import importlib
|
|
import inspect
|
|
|
|
init(autoreset=True)
|
|
|
|
# Corrected the extensions list
|
|
extensions: List[str] = [
|
|
"cogs"
|
|
]
|
|
|
|
class Axiom(commands.AutoShardedBot):
|
|
def __init__(self, *arg, **kwargs):
|
|
intents = discord.Intents.all()
|
|
intents.presences = True
|
|
intents.members = True
|
|
shard_kwargs = get_shard_kwargs()
|
|
# Allow callers to override shard settings via kwargs
|
|
for key in ("shard_count", "shard_ids"):
|
|
if key in kwargs:
|
|
shard_kwargs[key] = kwargs.pop(key)
|
|
super().__init__(command_prefix=self.get_prefix,
|
|
case_insensitive=True,
|
|
intents=intents,
|
|
# The status is already set to Do Not Disturb here
|
|
status=discord.Status.do_not_disturb,
|
|
strip_after_prefix=True,
|
|
owner_ids=OWNER_IDS,
|
|
allowed_mentions=discord.AllowedMentions(
|
|
everyone=False, replied_user=False, roles=False),
|
|
**shard_kwargs)
|
|
self.status_index = 0
|
|
self.status_list = []
|
|
self._slash_synced = False
|
|
|
|
async def setup_hook(self):
|
|
await setup_db()
|
|
await self.load_extensions()
|
|
self.status_task.start()
|
|
|
|
async def load_extensions(self):
|
|
for extension in extensions:
|
|
try:
|
|
await self.load_extension(extension)
|
|
print(Fore.GREEN + Style.BRIGHT + f"Loaded extension: {extension}")
|
|
except Exception as e:
|
|
print(f"{Fore.RED}{Style.BRIGHT}Failed to load extension {extension}. {e}")
|
|
print(Fore.GREEN + Style.BRIGHT + "*" * 20)
|
|
|
|
@tasks.loop(seconds=30)
|
|
async def status_task(self):
|
|
await self.wait_until_ready()
|
|
if not self.guilds:
|
|
return
|
|
|
|
guild = self.guilds[0] # Use first available guild for prefix
|
|
try:
|
|
config = await getConfig(guild.id)
|
|
prefix = config.get("prefix", ">")
|
|
except:
|
|
prefix = ">"
|
|
|
|
user_count = sum(g.member_count or 0 for g in self.guilds)
|
|
guild_count = len(self.guilds)
|
|
|
|
self.status_list = [
|
|
(discord.ActivityType.playing, f"{prefix}help | Security in your Server"),
|
|
(discord.ActivityType.watching, f"{user_count} users"),
|
|
(discord.ActivityType.watching, f"{guild_count} servers"),
|
|
(discord.ActivityType.listening, "Killing Nukers"),
|
|
(discord.ActivityType.playing, f"Protector {BotName}"),
|
|
]
|
|
|
|
current = self.status_list[self.status_index % len(self.status_list)]
|
|
# This task only changes the activity, not the online status (dnd, idle, etc.)
|
|
await self.change_presence(activity=discord.Activity(type=current[0], name=current[1]))
|
|
self.status_index += 1
|
|
|
|
async def send_raw(self, channel_id: int, content: str, **kwargs) -> typing.Optional[discord.Message]:
|
|
await self.http.send_message(channel_id, content, **kwargs)
|
|
|
|
async def invoke_help_command(self, ctx: Context) -> None:
|
|
return await ctx.send_help(ctx.command)
|
|
|
|
async def fetch_message_by_channel(self, channel: discord.TextChannel, messageID: int) -> typing.Optional[discord.Message]:
|
|
async for msg in channel.history(limit=1, before=discord.Object(messageID + 1), after=discord.Object(messageID - 1)):
|
|
return msg
|
|
|
|
async def get_prefix(self, message: discord.Message):
|
|
if message.guild:
|
|
guild_id = message.guild.id
|
|
async with aiosqlite.connect(db_path('np.db')) as db:
|
|
async with db.execute("SELECT id FROM np WHERE id = ?", (message.author.id,)) as cursor:
|
|
row = await cursor.fetchone()
|
|
data = await getConfig(guild_id)
|
|
prefix = data["prefix"]
|
|
if row:
|
|
return commands.when_mentioned_or(prefix, '')(self, message)
|
|
else:
|
|
return commands.when_mentioned_or(prefix)(self, message)
|
|
else:
|
|
async with aiosqlite.connect(db_path('np.db')) as db:
|
|
async with db.execute("SELECT id FROM np WHERE id = ?", (message.author.id,)) as cursor:
|
|
row = await cursor.fetchone()
|
|
if row:
|
|
return commands.when_mentioned_or('?', '')(self, message)
|
|
else:
|
|
return commands.when_mentioned_or('')(self, message)
|
|
|
|
async def on_message_edit(self, before, after):
|
|
ctx: Context = await self.get_context(after, cls=Context)
|
|
if before.content != after.content:
|
|
if after.guild is None or after.author.bot:
|
|
return
|
|
if ctx.command is None:
|
|
return
|
|
if type(ctx.channel) == "public_thread":
|
|
return
|
|
await self.invoke(ctx)
|
|
|
|
def setup_bot():
|
|
intents = discord.Intents.all()
|
|
bot = Axiom(intents=intents)
|
|
return bot
|