fix(invites): vanity invite tracking + auto sync

core/bot_classes/axobot.py

@@ -246,6 +246,13 @@ async def get_config(self, guild_id: discord.Guild | int, option: str):
         options_list = await self.get_options_list()
         return options_list.get(option, {"default": None})["default"]
 
+    async def get_guilds_with_config(self, option_name: str, option_value: str):
+        """Get a list of guilds with a specific config option set to a specific value"""
+        cog = self.get_cog("ServerConfig")
+        if cog and self.database_online:
+            return await cog.db_get_guilds_with_value(option_name, option_value)
+        return []
+
     async def get_recipient(self, channel: discord.DMChannel) -> discord.User | None:
         """Get the recipient of the given DM channel
 

modules/invites_tracker/invites_tracker.py

@@ -3,7 +3,7 @@
 
 import discord
 from discord import app_commands
-from discord.ext import commands
+from discord.ext import commands, tasks
 
 from core.arguments.args import GuildInviteArgument
 from core.bot_classes import Axobot
@@ -19,13 +19,20 @@ def __init__(self, bot: Axobot):
         self.bot = bot
         self.file = "invites_tracker"
 
+    async def cog_load(self):
+        self.sync_all_guilds_invites.start() # pylint: disable=no-member
+
+    async def cog_unload(self):
+        self.sync_all_guilds_invites.cancel() # pylint: disable=no-member
+
     async def db_get_invites(self, guild_id: int) -> list[TrackedInvite]:
         "Get a list of tracked invites associated to a guild"
         query = "SELECT * FROM `invites_tracker` WHERE `guild_id` = %s AND `beta` = %s"
         async with self.bot.db_main.read(query, (guild_id, self.bot.beta)) as query_result:
             return query_result
 
-    async def db_add_invite(self, guild_id: int, invite_id: str, user_id: int | None, creation_date: datetime, usage_count: int):
+    async def db_add_invite(self, guild_id: int, invite_id: str, user_id: int | None, creation_date: datetime | None,
+                            usage_count: int):
         "Insert a tracked invite in the database, or update it if it already exists"
         query = (
             "INSERT INTO `invites_tracker` "\
@@ -58,8 +65,11 @@ async def db_delete_invite(self, guild_id: int, invite_id: str):
     async def _get_guild_invites_with_vanity(self, guild: discord.Guild):
         "Get the guild invites, including the vanity invite if it exists"
         guild_invites = await guild.invites()
-        if vanity_invite := await guild.vanity_invite():
-            guild_invites.append(vanity_invite)
+        try:
+            if vanity_invite := await guild.vanity_invite():
+                guild_invites.append(vanity_invite)
+        except discord.Forbidden:
+            pass
         return guild_invites
 
     async def sync_guild_invites(self, guild: discord.Guild):
@@ -104,6 +114,34 @@ async def is_tracker_enabled(self, guild_id: int) -> bool:
         return await self.bot.get_config(guild_id, "enable_invites_tracking")
 
 
+    @tasks.loop(hours=6)
+    async def sync_all_guilds_invites(self):
+        "Sync the tracked invites of all guilds"
+        synced_invites = 0
+        for guild_id in await self.bot.get_guilds_with_config("enable_invites_tracking", str(True)):
+            try:
+                guild = self.bot.get_guild(guild_id)
+                if guild and guild.me.guild_permissions.manage_guild:
+                    synced_invites += await self.sync_guild_invites(guild)
+            except Exception as err:
+                self.bot.dispatch("error", err)
+        self.bot.log.info(f"Synced {synced_invites} invites")
+        try:
+            emb = discord.Embed(
+                description=f"{synced_invites} **synced invitations**",
+                color=0x6699ff,
+                timestamp=self.bot.utcnow()
+            )
+            emb.set_author(name=self.bot.user, icon_url=self.bot.user.display_avatar)
+            await self.bot.send_embed(emb)
+        except Exception as err:
+            self.bot.dispatch("error", err)
+
+    @sync_all_guilds_invites.before_loop
+    async def before_printer(self):
+        """Wait until the bot is ready"""
+        await self.bot.wait_until_ready()
+
     @commands.Cog.listener()
     async def on_invite_create(self, invite: discord.Invite):
         "Update the tracked invite when a new invite is created"

modules/serverconfig/serverconfig.py

@@ -232,6 +232,16 @@ async def db_get_value(self, guild_id: int, option_name: str) -> str | None:
                 return None
             return query_results["value"]
 
+    async def db_get_guilds_with_value(self, option_name: str, value: str) -> list[int]:
+        "Get a list of guilds with a specific value"
+        if option_name not in (await self.get_options_list()):
+            raise ValueError(f"Option {option_name} does not exist")
+        if not self.bot.database_online:
+            raise RuntimeError("Database is offline")
+        query = "SELECT `guild_id` FROM `serverconfig` WHERE `option_name` = %s AND `value` = %s AND `beta` = %s"
+        async with self.bot.db_main.read(query, (option_name, value, self.bot.beta)) as query_results:
+            return [row["guild_id"] for row in query_results]
+
     async def db_get_guild(self, guild_id: int) -> dict[str, str] | None:
         "Get a guild from the database"
         if not self.bot.database_online: