1
Fork 0
mirror of https://github.com/wlinator/luminara.git synced 2024-10-02 18:23:12 +00:00

Fix owner fetching and add sql commands

This commit is contained in:
wlinator 2024-08-29 09:24:26 -04:00
parent f94e818522
commit aa9aae2657
3 changed files with 92 additions and 10 deletions

View file

@ -1,10 +1,14 @@
import asyncio import asyncio
import os
import platform
from typing import Any from typing import Any
import discord
from discord.ext import commands from discord.ext import commands
from loguru import logger from loguru import logger
from db.database import run_migrations from db.database import run_migrations
from lib.const import CONST
from lib.loader import CogLoader from lib.loader import CogLoader
@ -13,8 +17,20 @@ class Luminara(commands.Bot):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
self.is_shutting_down: bool = False self.is_shutting_down: bool = False
self.setup_task: asyncio.Task[None] = asyncio.create_task(self.setup()) self.setup_task: asyncio.Task[None] = asyncio.create_task(self.setup())
self.strip_after_prefix = True
self.case_insensitive = True async def on_ready(self) -> None:
logger.success(f"{CONST.TITLE} v{CONST.VERSION}")
logger.success(f"Logged in with ID {self.user.id if self.user else 'Unknown'}")
logger.success(f"discord.py API version: {discord.__version__}")
logger.success(f"Python version: {platform.python_version()}")
logger.success(f"Running on: {platform.system()} {platform.release()} ({os.name})")
if self.owner_ids:
for owner in self.owner_ids:
logger.info(f"Added bot administrator: {owner}")
if not self.setup_task.done():
await self.setup_task
async def setup(self) -> None: async def setup(self) -> None:
try: try:
@ -26,16 +42,8 @@ class Luminara(commands.Bot):
await self.load_cogs() await self.load_cogs()
async def load_cogs(self) -> None: async def load_cogs(self) -> None:
logger.debug("Loading cogs...")
await CogLoader.setup(bot=self) await CogLoader.setup(bot=self)
@commands.Cog.listener()
async def on_ready(self) -> None:
logger.success(f"Logged in as {self.user}.")
if not self.setup_task.done():
await self.setup_task
@commands.Cog.listener() @commands.Cog.listener()
async def on_disconnect(self) -> None: async def on_disconnect(self) -> None:
logger.warning("Disconnected from Discord.") logger.warning("Disconnected from Discord.")

View file

@ -28,6 +28,8 @@ async def main() -> None:
intents=discord.Intents.all(), intents=discord.Intents.all(),
command_prefix=get_prefix, command_prefix=get_prefix,
allowed_mentions=discord.AllowedMentions(everyone=False), allowed_mentions=discord.AllowedMentions(everyone=False),
case_insensitive=True,
strip_after_prefix=True,
) )
try: try:

72
modules/admin/admin.py Normal file
View file

@ -0,0 +1,72 @@
import mysql.connector
from discord.ext import commands
from db import database
from lib.const import CONST
from lib.format import shorten
from ui.embeds import Builder
class Sql(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
@commands.command(name="sqlselect", aliases=["sqls"])
@commands.is_owner()
async def select_cmd(self, ctx: commands.Context[commands.Bot], *, query: str) -> None:
if query.lower().startswith("select "):
query = query[7:]
try:
results = database.select_query(f"SELECT {query}")
embed = Builder.create_embed(
user_name=ctx.author.name,
author_text=CONST.STRINGS["admin_sql_select_title"],
description=CONST.STRINGS["admin_sql_select_description"].format(
shorten(query, 200),
shorten(str(results), 200),
),
hide_name_in_description=True,
)
except mysql.connector.Error as error:
embed = Builder.create_embed(
user_name=ctx.author.name,
author_text=CONST.STRINGS["admin_sql_select_error_title"],
description=CONST.STRINGS["admin_sql_select_error_description"].format(
shorten(query, 200),
shorten(str(error), 200),
),
hide_name_in_description=True,
)
await ctx.send(embed=embed, ephemeral=True)
@commands.command(name="sqlinject", aliases=["sqli"])
@commands.is_owner()
async def inject_cmd(self, ctx: commands.Context[commands.Bot], *, query: str) -> None:
try:
database.execute_query(query)
embed = Builder.create_embed(
user_name=ctx.author.name,
author_text=CONST.STRINGS["admin_sql_inject_title"],
description=CONST.STRINGS["admin_sql_inject_description"].format(
shorten(query, 200),
),
hide_name_in_description=True,
)
except mysql.connector.Error as error:
embed = Builder.create_embed(
user_name=ctx.author.name,
author_text=CONST.STRINGS["admin_sql_inject_error_title"],
description=CONST.STRINGS["admin_sql_inject_error_description"].format(
shorten(query, 200),
shorten(str(error), 200),
),
hide_name_in_description=True,
)
await ctx.send(embed=embed, ephemeral=True)
async def setup(bot: commands.Bot) -> None:
await bot.add_cog(Sql(bot))