diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 0aa0dde..e69de29 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -1,42 +0,0 @@ -# This workflow will install Python dependencies, run tests and lint with a single version of Python -# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python - -name: Python application - -on: - push: - branches: [ "master" ] - pull_request: - branches: [ "master" ] - -permissions: - contents: read - -jobs: - build: - - runs-on: prod - - steps: - - uses: actions/checkout@v4 - - name: Set up Python 3.10 - uses: actions/setup-python@v3 - with: - python-version: "3.10" - - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install flake8 pytest - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - - name: Lint with flake8 - run: | - # stop the build if there are Python syntax errors or undefined names - flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics - # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide - flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics - - name: Test with pytest - run: | - pytest - - name: Run Bot - run: | - python3 ./index.py diff --git a/cogs/admin.py b/cogs/admin.py deleted file mode 100644 index bc50286..0000000 --- a/cogs/admin.py +++ /dev/null @@ -1,92 +0,0 @@ -import os - -from typing import List - -from discord import Interaction, Object -from discord.app_commands import Choice, Group, autocomplete -from discord.ext.commands import Bot, Cog - -class Admin(Cog): - def __init__(self, bot: Bot): - self.bot = bot - - async def unloaded_extension_list( - self, - interaction: Interaction, - current: str - ) -> List[Choice[str]]: - return [ - Choice( - name = extension, - value = extension - ) - for extension in [ - extension[:-3] - for extension in os.listdir("./cogs") if extension.endswith(".py") - ] if (current.lower() in extension.lower()) and (extension not in self.bot.loaded_extension_list) - ] - - async def loaded_extension_list( - self, - interaction: Interaction, - current: str - ) -> List[Choice[str]]: - return [ - Choice( - name = extension, - value = extension - ) - for extension in [ - extension[:-3] - for extension in os.listdir("./cogs") if extension.endswith(".py") - ] if (current.lower() in extension.lower()) and (extension in self.bot.loaded_extension_list) - ] - - cog = Group(name="cog", description = "Group of commands to manage cogs.") - - @cog.command(name = "load", description = "Load a cog.") - @autocomplete(extension = unloaded_extension_list) - async def load(self, interaction: Interaction, extension: str): - try: - await self.bot.load_extension(f"cogs.{extension}") - await interaction.response.send_message( - f"{os.getenv('EMOJI_SUCCESS')} Loaded `cogs.{extension}` extension" - ) - except Exception as error: - await interaction.response.send_message( - f"{os.getenv('EMOJI_FAIL')} Unable to load `cogs.{extension}`\n```python\n{error}\n```", - ephemeral = True - ) - - @cog.command(name = "unload", description = "Unload a cog.") - @autocomplete(extension = loaded_extension_list) - async def unload(self, interaction: Interaction, extension: str): - if extension == "admin": - return await interaction.response.send_message(f"{os.getenv('EMOJI_FAIL')} Unloading `cogs.{extension}` is disallowed") - try: - await self.bot.unload_extension(f"cogs.{extension}") - await interaction.response.send_message( - f"{os.getenv('EMOJI_SUCCESS')} Unloaded `cogs.{extension}` extension" - ) - except Exception as error: - await interaction.response.send_message( - f"{os.getenv('EMOJI_FAIL')} Unable to unload `cogs.{extension}`\n```python\n{error}\n```", - ephemeral = True - ) - - @cog.command(name = "reload", description = "Reload a cog.") - @autocomplete(extension = loaded_extension_list) - async def reload(self, interaction: Interaction, extension: str): - try: - await self.bot.reload_extension(f"cogs.{extension}") - await interaction.response.send_message( - f"{os.getenv('EMOJI_SUCCESS')} Reloaded `cogs.{extension}` extension" - ) - except Exception as error: - await interaction.response.send_message( - f"{os.getenv('EMOJI_FAIL')} Unable to reload `cogs.{extension}`\n```python\n{error}\n```", - ephemeral = True - ) - -async def setup(bot: Bot): - await bot.add_cog(Admin(bot), guild = Object(os.getenv("GUILD"))) \ No newline at end of file diff --git a/cogs/copyright.py b/cogs/copyright.py index c419dfd..5db115e 100644 --- a/cogs/copyright.py +++ b/cogs/copyright.py @@ -18,10 +18,10 @@ async def copyright(self, interaction: Interaction): colour = 0x2B2D31 ).set_author( name = "Copyright Notice", - icon_url = "http://cdn.uwitz.org/r/justice.png" + icon_url = "https://cdn.uwitz.org/r/justice.png" ).set_footer( text = "Uwitz Federation", - icon_url = "http://cdn.uwitz.org/r/shield-logo.png" + icon_url = "https://cdn.uwitz.org/r/shield-logo.png" ) await interaction.response.send_message( embed = embed, diff --git a/cogs/developer.py b/cogs/developer.py new file mode 100644 index 0000000..e2ef7da --- /dev/null +++ b/cogs/developer.py @@ -0,0 +1,135 @@ +import os + +from typing import List + +from discord import Embed, Interaction, Object +from discord.app_commands import Choice, Group, autocomplete, command +from discord.ext.commands import Bot, Cog + +class Developer(Cog): + def __init__(self, bot: Bot): + self.bot = bot + + async def unloaded_extension_list( + self, + interaction: Interaction, + current: str + ) -> List[Choice[str]]: + return [ + Choice( + name = extension, + value = extension + ) + for extension in [ + extension[:-3] + for extension in os.listdir("./cogs") if extension.endswith(".py") + ] if (current.lower() in extension.lower()) and (extension not in self.bot.loaded_extension_list) + ] + + async def loaded_extension_list( + self, + interaction: Interaction, + current: str + ) -> List[Choice[str]]: + return [ + Choice( + name = extension, + value = extension + ) + for extension in [ + extension[:-3] + for extension in os.listdir("./cogs") if extension.endswith(".py") + ] if (current.lower() in extension.lower()) and (extension in self.bot.loaded_extension_list) + ] + + cog = Group(name="cog", description = "Group of commands to manage cogs.") + + @cog.command(name = "enable", description = "Load a cog to the bot's runtime") + @autocomplete(extension = unloaded_extension_list) + async def enable(self, interaction: Interaction, extension: str): + try: + await self.bot.load_extension(f"cogs.{extension}") + await interaction.response.send_message( + f"{os.getenv('EMOJI_SUCCESS')} Loaded `cogs.{extension}` extension" + ) + except Exception as error: + await interaction.response.send_message( + f"{os.getenv('EMOJI_FAIL')} Unable to load `cogs.{extension}`\n```python\n{error}\n```", + ephemeral = True + ) + + @cog.command(name = "disable", description = "Unload a cog from the bot's runtime") + @autocomplete(extension = loaded_extension_list) + async def disable(self, interaction: Interaction, extension: str): + if extension == "developer": + return await interaction.response.send_message(f"{os.getenv('EMOJI_FAIL')} Unloading `cogs.{extension}` is disallowed") + try: + await self.bot.unload_extension(f"cogs.{extension}") + self.bot.loaded_extension_list.remove(extension) + await interaction.response.send_message( + f"{os.getenv('EMOJI_SUCCESS')} Unloaded `cogs.{extension}` extension" + ) + except Exception as error: + await interaction.response.send_message( + f"{os.getenv('EMOJI_FAIL')} Unable to unload `cogs.{extension}`\n```python\n{error}\n```", + ephemeral = True + ) + + @cog.command(name = "restart", description = "Unload a cog from bots runtime to load updated code") + @autocomplete(extension = loaded_extension_list) + async def restart(self, interaction: Interaction, extension: str): + try: + await self.bot.reload_extension(f"cogs.{extension}") + await interaction.response.send_message( + f"{os.getenv('EMOJI_SUCCESS')} Reloaded `cogs.{extension}` extension" + ) + except Exception as error: + await interaction.response.send_message( + f"{os.getenv('EMOJI_FAIL')} Unable to reload `cogs.{extension}`\n```python\n{error}\n```", + ephemeral = True + ) + + @command(name = "health", description = "Check the bot's developer information.") + async def health(self, interaction: Interaction): + unloaded_extensions = [ + extension + for extension in [ + extension[:-3] + for extension in os.listdir("./cogs") if extension.endswith(".py") + ] if extension not in self.bot.loaded_extension_list + ] + unloaded_extensions = ["null"] if len(unloaded_extensions) == 0 else unloaded_extensions + + ping = round(self.bot.latency * 1000) + efficiency_description = "peak" if ping <= 50 and len(unloaded_extensions) == 0 else ("critical" if self.bot.internal_error_occured else "degraded") + status = "critical-health" if self.bot.internal_error_occured else ("degraded-health" if len(unloaded_extensions) > 0 or ping >= 125 else "good-health") + ping_emoji = os.getenv('EMOJI_GOODPING') if ping <= 50 else (os.getenv('EMOJI_MODERATEPING') if ping <= 125 else os.getenv('EMOJI_BADPING')) + + embed = Embed( + description = f"Running `v{self.bot.version}` with `{efficiency_description}` performance", + colour = 0x2B2D31 + ).add_field( + name = "> Ping", + value = f"{ping_emoji} `{ping}ms`", + inline = True + ).add_field( + name = "> Servers", + value = f"`{len(self.bot.guilds)}` *(`{len(self.bot.users)}` members)*", + inline = True + ).add_field( + name = "> Loaded", + value = f"```diff\n+ {'\n+ '.join(self.bot.loaded_extension_list)}\n```", + inline = False + ).add_field( + name = "> Unloaded", + value = f"```diff\n- {'\n- '.join(unloaded_extensions)}\n```", + inline = True + ).set_author( + name = "Health Status", + icon_url = f"https://cdn.uwitz.org/r/{status}.png" + ) + + await interaction.response.send_message(embed = embed) + +async def setup(bot: Bot): + await bot.add_cog(Developer(bot), guild = Object(os.getenv("GUILD"))) \ No newline at end of file diff --git a/cogs/error_handler.py b/cogs/error_handler.py new file mode 100644 index 0000000..4544fc9 --- /dev/null +++ b/cogs/error_handler.py @@ -0,0 +1,13 @@ +from discord.ext.commands import Bot, Cog + +class ErrorHandler(Cog): + def __init__(self, bot: Bot): + self.bot = bot + + @Cog.listener("on_error") + async def error_event(self, exception): + print(f"ERROR:\n{exception}") + self.bot.internal_error_occured = True + +async def setup(bot: Bot): + await bot.add_cog(ErrorHandler(bot)) \ No newline at end of file diff --git a/cogs/mod.py b/cogs/mod.py index 39652b5..c3e4096 100644 --- a/cogs/mod.py +++ b/cogs/mod.py @@ -118,6 +118,74 @@ class Mod(Cog): def __init__(self, bot): self.bot = bot + # @command( + # name = "ban", + # description = "Banish a member with a valid reason.", + # ) + # @describe(member = "The member you'd like banned.") + # @describe(reason = "Why should the member be banned?") + # async def ban(self, interaction: Interaction, member: Member, reason: str): + # guild_config = await self.bot.database["config"].find_one( + # { + # "_id": interaction.guild.id + # } + # ) + # report_channel = self.bot.get_channel(os.getenv("ENFORCEMENT")) + # if interaction.user.id in + + # report_channel = self.bot.get_channel(os.getenv("ENFORCEMENT")) + # if not interaction.user.id == int(os.getenv("OWNER_ID")): + # guild_config = await self.bot.database["config"].find_one( + # { + # "_id": interaction.guild.id + # } + # ) + # for role in guild_config.get("ADMINISTRATOR_ROLES"): + # if (interaction.guild.get_role(role) in interaction.user.roles) or (interaction.user.guild_permissions.administrator): + # if interaction.user.guild_permissions.administrator or ( + # (os.getenv("ADMINISTRATOR_ROLE"), os.getenv("MODERATOR_ROLE")) in [role.id for role in interaction.user.roles] + # ): + # await interaction.response.send(f"{os.getenv("EMOJI_FAIL")} You are not authorised to moderate another person in authority.") + # break + # report = Embed( + # description = f"Sudo Banned by <@!{interaction.user.id}>\n\n**Reason:**\n```diff\n - {reason}```", + # timestamp = datetime.now(), + # color = 0xFF7A7A + # ).set_author( + # name = member.display_name, + # icon_url = member.display_avatar.url + # ) + + # try: + # await member.ban( + # reason = reason, + # delete_message_seconds = 60 + # ) + # except: + # return await interaction.response.send_message(f"{os.getenv("EMOJI_FAIL")} The Royal Defence is not authorised to ban this user.") + + # await report_channel.send(embed = report) + # return await interaction.response.send_message(f"{os.getenv("EMOJI_SUCCESS")} Successfully Sudo Banished User.", ephemeral = True) + + # confirmation_embed = Embed( + # description = f"Would you like to confirm that you want to send a ban report about <@!{member.id}> for:\n```diff\n- {reason}\n```", + # timestamp = datetime.now(), + # colour = 0xFF7A7A + # ).set_author( + # name = member.display_name, + # icon_url = member.display_avatar.url + # ) + + # await interaction.response.send_message( + # embed = confirmation_embed, + # view = Prompt( + # target_user = member, + # reason = reason, + # report_channel = report_channel + # ) + # ) + # break + @command( name = "mute", description = "Mute a member with a valid reason." diff --git a/cogs/sync.py b/cogs/sync.py new file mode 100644 index 0000000..1aad6f8 --- /dev/null +++ b/cogs/sync.py @@ -0,0 +1,105 @@ +import re + +from discord import Message, WebhookMessage +from discord import Webhook +from discord.ext.commands import Cog + +from typing import List +from aiohttp import ClientSession + +def _clean_message(message: str) -> str: + return re.sub(r'@(everyone|here|[!&]?[0-9]{17,21})', '@\u200b\\1', message) + +class Sync(Cog): + def __init__(self, bot): + self.bot = bot + + @Cog.listener("on_message") + async def message_sync(self, message: Message): + if message.author.bot: return + current_guild_config = await self.bot.chatsync_db["config"].find_one( + { + "_id": message.guild.id + } + ) + if current_guild_config.get("linked") and message.channel.id == current_guild_config.get("sync_channel"): + guild_config_list: List[dict] = self.bot.chatsync_db["config"].find( + { + "linked": True + } + ) + message_content = _clean_message(message.content) + message_data = { + "_id": message.id, + "message_author": { + "id": message.author.id, + "username": message.author.display_name, + "name": message.author.name, + "avatar_icon": message.author.display_avatar.url + }, + "content": message.content, + "guild_messages": { + f"{message.guild.id}": message.id + }, + "attachments": message.attachments + } + async with ClientSession() as session: + async for guild in guild_config_list: + if guild.get("_id") != message.guild.id: + webhook = Webhook.from_url( + guild.get("sync_webhook"), + session = session + ) + webhook_message: WebhookMessage = await webhook.send( + username = f"{message.author.display_name} ({message.guild.name})", + avatar_url = message.author.display_avatar.url, + content = message_content, + # files = message.attachments, + allowed_mentions = False, + wait = True + ) + message_data["guild_messages"][f"{guild.get('_id')}"] = webhook_message.id + + await self.bot.chatsync_db["messages"].insert_one(message_data) + + @Cog.listener("on_message_edit") + async def edit_sync(self, message_before: Message, message_after: Message): + if message_after.author.bot: return + current_guild_config = await self.bot.chatsync_db["config"].find_one( + { + "_id": message_after.guild.id + } + ) + if current_guild_config.get("linked") and message_after.channel.id == current_guild_config.get("sync_channel"): + guild_config_list: List[dict] = self.bot.chatsync_db["config"].find( + { + "linked": True + } + ) + original_message = await self.bot.chatsync_db["messages"].find_one( + { + "_id": message_after.id + } + ) + guild_messages = original_message.get("guild_messages") + async with ClientSession() as session: + async for guild in guild_config_list: + if guild.get("_id") == message_after.guild.id: continue + webhook = Webhook.from_url( + guild.get("sync_webhook"), + session = session + ) + await webhook.edit_message(guild_messages[f"{guild.get("_id")}"], content = message_after.content) + await self.bot.chatsync_db["messages"].update_one( + { + "_id": message_after.id + }, + { + "$set": { + "content": message_after.content + } + } + ) + +async def setup(bot): + await bot.add_cog(Sync(bot)) \ No newline at end of file diff --git a/index.py b/index.py index ade683d..e358d38 100644 --- a/index.py +++ b/index.py @@ -12,8 +12,10 @@ class System(Bot): def __init__(self): + self.version = "0.0.1a" self.loaded_extension_list = [] self.unloaded_extension_list = [] + self.internal_error_occured = False intents = Intents.all() super().__init__(intents = intents, command_prefix = "/") @@ -31,16 +33,18 @@ async def setup_hook(self) -> Coroutine[Any, Any, None]: self.encryption.load_credentials() except FileNotFoundError: self.encryption.generate_credentials() - - if os.getenv("MONGO_TLS") == "True": - self.database = AsyncIOMotorClient( - os.getenv("MONGO"), - tls = True, - tlsCertificateKeyFile = "mongo_cert.pem" - )["disect"] - else: - self.database = AsyncIOMotorClient(os.getenv("MONGO"))["disect"] - + + self.database = AsyncIOMotorClient( + os.getenv("MONGO"), + tls = True, + tlsCertificateKeyFile = "mongo_cert.pem" + )["disect"] + self.chatsync_db = AsyncIOMotorClient( + os.getenv("MONGO"), + tls = True, + tlsCertificateKeyFile = "mongo_cert.pem" + )["channelsync"] + for file in os.listdir("./cogs"): if file.endswith(".py"): try: @@ -56,4 +60,4 @@ async def setup_hook(self) -> Coroutine[Any, Any, None]: if __name__ == "__main__": load_dotenv(find_dotenv()) bot = System() - bot.run(os.getenv("TOKEN")) \ No newline at end of file + bot.run(os.getenv("TOKEN"))