109 lines
5.6 KiB
Python
109 lines
5.6 KiB
Python
import asyncio
|
|
import os
|
|
import time
|
|
import aiohttp
|
|
import dotenv
|
|
import revolt
|
|
from dotenv import load_dotenv
|
|
from revolt.ext import commands
|
|
from cogs.info import Info
|
|
from cogs.moderation import Moderation
|
|
from utils.embed import CustomEmbed
|
|
|
|
# This code reads the variables set in the bot's '.env' file.
|
|
env = dotenv.find_dotenv()
|
|
load_dotenv(env)
|
|
token = os.getenv('TOKEN')
|
|
api_url = os.getenv('API_URL')
|
|
prefix = os.getenv('PREFIX')
|
|
message_logging_channel = os.getenv('MESSAGE_LOGGING_CHANNEL')
|
|
app_url = os.getenv('APP_URL')
|
|
|
|
class Client(commands.CommandsClient):
|
|
"""This class contains all of the commands/methods required for core functionality. Everything else will be relegated to cogs (eventually)."""
|
|
async def get_prefix(self, message: revolt.Message, input: str | None = None): # pylint: disable=W0622
|
|
if input is None:
|
|
return prefix
|
|
return input
|
|
|
|
async def on_ready(self):
|
|
for server in self.servers:
|
|
Moderation.create_server_table(self, server)
|
|
print(f"Logged in as {self.user.name}#{self.user.discriminator} ({self.user.id})")
|
|
|
|
async def on_server_join(self, server: revolt.Server):
|
|
print(f"Joined server: {server.name} ({server.id})")
|
|
Moderation.create_server_table(self, server)
|
|
|
|
async def on_message_delete(self, message: revolt.Message):
|
|
if isinstance(message.author, revolt.Member):
|
|
if message.author.bot is True:
|
|
return
|
|
truncated_content = message.content[:1500] + "..."
|
|
embed = [CustomEmbed(description=f"## Message Deleted in {message.channel.mention}\n**Author:** {message.author.name}#{message.author.discriminator} ({message.author.id})\n**Content:** {message.content}", colour="#5d82d1"), CustomEmbed(description=f"## Message Deleted in {message.channel.mention}\n**Author:** {message.author.name}#{message.author.discriminator} ({message.author.id})\n**Content:** {truncated_content}\n\n*Message content is over the character limit.*", colour="#5d82d1")]
|
|
embed[0].set_footer(f"Message ID: {message.id}")
|
|
embed[1].set_footer(f"Message ID: {message.id}")
|
|
channel = self.get_channel(message_logging_channel)
|
|
try:
|
|
try:
|
|
await channel.send(embed=embed[0])
|
|
except LookupError:
|
|
print("Message logging channel not found for server ID: " + message.server.id)
|
|
except(revolt.errors.HTTPError):
|
|
try:
|
|
await channel.send(embed=embed[1])
|
|
except LookupError:
|
|
print("Message logging channel not found for server ID: " + message.server.id)
|
|
else:
|
|
print(f"{message.author.name}#{message.author.discriminator} ({message.author.id}): {message.content}\n ⤷ Deleted from Direct Messages\n Message ID: {message.id}")
|
|
|
|
async def on_message_update(self, before: revolt.Message, after: revolt.Message):
|
|
if isinstance(after.author, revolt.Member):
|
|
if after.author.bot is True:
|
|
return
|
|
message_link = f"{app_url}/server/{after.server.id}/channel/{after.channel.id}/message/{after.id}"
|
|
embeds = [CustomEmbed(description=f"## Message Edited in {after.channel.mention}\n**Author:** {after.author.name}#{after.author.discriminator} ({after.author.id})\n**Message ID:** [{after.id}]({message_link})", colour="#5d82d1"), CustomEmbed(title="Old Content", description=before.content, colour="#5d82d1"), CustomEmbed(title="New Content", description=after.content, colour="#5d82d1")]
|
|
channel = self.get_channel(message_logging_channel)
|
|
try:
|
|
await channel.send(embed=embeds[0])
|
|
await channel.send(embed=embeds[1])
|
|
await channel.send(embed=embeds[2])
|
|
except LookupError:
|
|
print("Message logging channel not found for server ID: " + after.server.id)
|
|
else:
|
|
print(f"{after.author.name}#{after.author.discriminator} ({after.author.id}): {before.content}\n ⤷ Edited in Direct Messages\n New Content: {after.content}")
|
|
|
|
@commands.command()
|
|
async def ping(self, ctx: commands.Context):
|
|
"""This command checks the bot's latency."""
|
|
before = time.monotonic()
|
|
msg = await ctx.message.reply("🏓")
|
|
ping = (time.monotonic() - before) * 1000
|
|
embed = [CustomEmbed(title="🏓 Pong!", description=f"`\n{int(ping)} ms`", colour="#5d82d1")]
|
|
await msg.edit(content=" ", embeds=embed)
|
|
print(f'Ping {int(ping)}ms')
|
|
|
|
async def prefix_change(self, message: revolt.Message, new_prefix: str, silent: bool | None = False):
|
|
dotenv.set_key(env, 'PREFIX', new_prefix)
|
|
if silent is not True:
|
|
await message.reply(f"Prefix has been changed from `{prefix}` to `{new_prefix}`!")
|
|
print(f"Prefix changed: {prefix} → {new_prefix}")
|
|
await Client.get_prefix(message, new_prefix)
|
|
|
|
@commands.command()
|
|
async def prefix(self, ctx: commands.Context, new_prefix: str = None):
|
|
"""This command sets the bot's prefix. CURRENTLY BROKEN"""
|
|
if new_prefix is not None and ctx.author.id == ctx.client.user.owner_id:
|
|
await Client.prefix_change(self=self, message=ctx.message, new_prefix=new_prefix)
|
|
else:
|
|
await ctx.message.reply(f"The prefix is currently set to `{prefix}`.")
|
|
|
|
async def main():
|
|
"""This function logs into the bot user."""
|
|
async with aiohttp.ClientSession() as session:
|
|
client = Client(session, token, api_url=api_url)
|
|
client.add_cog(Moderation(client))
|
|
client.add_cog(Info(client))
|
|
await client.start()
|
|
|
|
asyncio.run(main())
|