147 lines
6.5 KiB
Python
147 lines
6.5 KiB
Python
# ╔══════════════════════════════════════════════════════════════════╗
|
|
# ║ ║
|
|
# ║ ░█▀▀░█▀█░█▀▄░█▀▀░█░█ ░█▀▄░█▀▀░█░█░█▀▀ ║
|
|
# ║ ░█░░░█░█░█░█░█▀▀░▄▀▄ ░█░█░█▀▀░▀▄▀░▀▀█ ║
|
|
# ║ ░▀▀▀░▀▀▀░▀▀░░▀▀▀░▀░▀ ░▀▀░░▀▀▀░░▀░░▀▀▀ ║
|
|
# ║ ║
|
|
# ║ © 2026 CodeX Devs — All Rights Reserved ║
|
|
# ║ ║
|
|
# ║ discord ── https://discord.gg/codexdev ║
|
|
# ║ youtube ── https://youtube.com/@CodeXDevs ║
|
|
# ║ github ── https://github.com/RayExo ║
|
|
# ║ ║
|
|
# ╚══════════════════════════════════════════════════════════════════╝
|
|
|
|
from __future__ import annotations
|
|
from discord.ext import commands, tasks
|
|
import discord
|
|
import aiohttp
|
|
import json
|
|
import jishaku
|
|
import asyncio
|
|
import typing
|
|
from typing import List
|
|
import aiosqlite
|
|
from utils.config import OWNER_IDS, BotName
|
|
from utils import getConfig, updateConfig
|
|
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 zyrox(commands.AutoShardedBot):
|
|
def __init__(self, *arg, **kwargs):
|
|
intents = discord.Intents.all()
|
|
intents.presences = True
|
|
intents.members = True
|
|
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),
|
|
sync_commands_debug=True,
|
|
sync_commands=True,
|
|
shard_count=1)
|
|
self.status_index = 0
|
|
self.status_list = []
|
|
|
|
async def setup_hook(self):
|
|
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/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/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 = zyrox(intents=intents)
|
|
return bot
|