"""Byte Bot."""
from __future__ import annotations
import contextlib
import discord
import httpx
from anyio import run
from discord import Activity, Forbidden, Intents, Member, Message, NotFound
from discord.ext.commands import Bot, CommandError, Context, ExtensionAlreadyLoaded
from dotenv import load_dotenv
from byte.lib import settings
from byte.lib.log import get_logger
__all__ = [
"Byte",
"run_bot",
]
logger = get_logger()
load_dotenv()
async def on_member_join(member: Member) -> None:
"""Handle member join event.
Args:
member: Member object.
"""
await member.send(
f"Welcome to {member.guild.name}! Please make sure to read the rules if you haven't already. "
f"Feel free to ask any questions you have in the help channel."
)
[docs]
class Byte(Bot):
"""Byte Bot Base Class."""
[docs]
def __init__(self, command_prefix: list[str], intents: Intents, activity: Activity) -> None:
"""Initialize the bot.
Args:
command_prefix (str): Command prefix for the bot.
intents (discord.Intents): Intents for the bot.
activity (discord.Activity): Activity for the bot.
"""
super().__init__(command_prefix=command_prefix, intents=intents, activity=activity)
[docs]
async def setup_hook(self) -> None:
"""Any setup we need can be here."""
# Load cogs before syncing the tree.
await self.load_cogs()
dev_guild = discord.Object(id=settings.discord.DEV_GUILD_ID)
self.tree.copy_global_to(guild=dev_guild)
await self.tree.sync(guild=dev_guild)
[docs]
async def load_cogs(self) -> None:
"""Load cogs."""
cogs = [
cog
for plugins_dir in settings.discord.PLUGINS_DIRS
for cog in plugins_dir.rglob("*.py")
if cog.stem != "__init__"
]
cogs_import_path = [".".join(cog.parts[cog.parts.index("src") : -1]) + "." + cog.stem for cog in cogs]
for cog in cogs_import_path:
with contextlib.suppress(ExtensionAlreadyLoaded):
await self.load_extension(cog)
[docs]
async def on_ready(self) -> None:
"""Handle bot ready event."""
logger.info("%s has connected to Discord!", self.user)
[docs]
async def on_message(self, message: Message) -> None:
"""Handle message events.
Args:
message: Message object.
"""
await self.process_commands(message)
[docs]
async def on_command_error(self, ctx: Context, error: CommandError) -> None:
"""Handle command errors.
Args:
ctx: Context object.
error: Error object.
"""
err = error.original if hasattr(error, "original") else error
if isinstance(err, Forbidden | NotFound):
return
embed = discord.Embed(title="Command Error", description=str(error), color=discord.Color.red())
embed.set_thumbnail(url=ctx.author.avatar.url)
embed.add_field(name="Command", value=ctx.command)
embed.add_field(name="Message", value=ctx.message.content)
embed.add_field(name="Channel", value=ctx.channel.mention)
embed.add_field(name="Author", value=ctx.author.mention)
embed.add_field(name="Guild", value=ctx.guild.name)
embed.add_field(name="Location", value=f"[Jump]({ctx.message.jump_url})")
embed.set_footer(text=f"Time: {ctx.message.created_at.strftime('%Y-%m-%d %H:%M:%S')}")
await ctx.send(embed=embed, ephemeral=True)
[docs]
@staticmethod
async def on_member_join(member: Member) -> None:
"""Handle member join event.
Args:
member: Member object.
"""
await member.send(
f"Welcome to {member.guild.name}! Please make sure to read the rules if you haven't already. "
f"Feel free to ask any questions you have in the help channel."
)
[docs]
async def on_guild_join(self, guild: discord.Guild) -> None:
"""Handle guild join event.
Args:
guild: Guild object.
"""
await self.tree.sync(guild=guild)
api_url = f"http://0.0.0.0:8000/api/guilds/create?guild_id={guild.id}&guild_name={guild.name}"
async with httpx.AsyncClient() as client:
response = await client.post(api_url)
if response.status_code == httpx.codes.CREATED:
logger.info("successfully added guild %s (ID: %s)", guild.name, guild.id)
else:
logger.error("%s joined guild '%s' but was not added to database", self.user.name, guild.name)
[docs]
def run_bot() -> None:
"""Run the bot."""
intents = discord.Intents.default()
intents.message_content = True
intents.members = True
presence = discord.Activity(
name="!help",
type=discord.ActivityType.custom,
state="Serving Developers",
details="!help",
url=settings.discord.PRESENCE_URL,
)
bot = Byte(command_prefix=settings.discord.COMMAND_PREFIX, intents=intents, activity=presence)
async def start_bot() -> None:
"""Start the bot."""
await bot.start(settings.discord.TOKEN)
run(start_bot)
if __name__ == "__main__":
run_bot()
