Add channel deduplication

breadtube_bot/bot.py

@@ -32,6 +32,7 @@ class Bot:
         DELETE_MESSAGES = 1
         SCAN_BOT_CHANNEL = 2
         REFRESH_SUBS = 3
+        DELETE_DUPLICATE_CHANNELS = 4
 
     @staticmethod
     def _get_code_version() -> str:
@@ -75,6 +76,7 @@ class Bot:
 
         categories, text_channel = self.discord_manager.list_channels(
             self.guild_id, request_timeout=self.config.request_timeout)
+
         self.guild_text_channels: list[TextChannel] = text_channel
         self.guild_categories: list[ChannelCategory] = categories
         self.init_message: Message | None = None
@@ -95,7 +97,9 @@ class Bot:
         self.tasks.append((
             self.Task.SCAN_BOT_CHANNEL, time.time() + self.config.bot_channel_scan_interval, None))
         self.tasks = list(filter(lambda t: t[0] != Bot.Task.REFRESH_SUBS, self.tasks))
-        self.tasks.append((Bot.Task.REFRESH_SUBS, time.time() + 1, None))
+        if self.config.delete_duplicate_channels:
+            self.tasks.append((Bot.Task.DELETE_DUPLICATE_CHANNELS, time.time(), None))
+        self.tasks.append((Bot.Task.REFRESH_SUBS, time.time(), None))
         self.logger.info('Bot initialized')
 
     def init_bot_channel(self) -> TextChannel | None:
@@ -122,6 +126,20 @@ class Bot:
                                 deny=Permissions.NONE)]},
             request_timeout=self.config.request_timeout)
 
+    def _delete_duplicate_text_channels(self):
+        self.logger.debug('Deleting duplicated text channels')
+        channel_names: list[str] = [c.name for c in self.guild_text_channels if c.name is not None]
+        unique_names: set[str] = set(channel_names)
+        for name in unique_names:
+            occurence = 0
+            for channel in self.guild_text_channels:
+                if channel.name == name:
+                    occurence += 1
+                    if occurence > 1:
+                        self.logger.debug('Deleting duplicate channel: %d -> %s', channel.id, name)
+                        self.discord_manager.delete_text_channel(
+                            channel.id, request_timeout=self.config.request_timeout)
+
     def _get_all_channel_messages(self, channel: TextChannel) -> list[Message]:
         messages_id_delete_task: set[int] = set()
         for task_type, _, task_params in self.tasks:
@@ -554,3 +572,9 @@ class Bot:
                     self.tasks = list(filter(lambda t: t[0] != Bot.Task.REFRESH_SUBS, self.tasks))
                     self.tasks.append((
                         self.Task.REFRESH_SUBS, time.time() + self.config.youtube_channel_refresh_interval, None))
+                case Bot.Task.DELETE_DUPLICATE_CHANNELS:
+                    try:
+                        self._delete_duplicate_text_channels()
+                    except Exception as error:
+                        self.logger.error('Error deleting duplicate channels : %s -> %s',
+                                          error, traceback.format_exc().replace('\n', ' | '))

breadtube_bot/config.py

@@ -10,6 +10,7 @@ class Config:
     bot_channel_scan_interval: float = 30.
     bot_channel_init_retries: int = 3
     bot_message_duration: float = 150.
+    delete_duplicate_channels: bool = True
     request_timeout: float = 3.
     unmanaged_categories: str = ''
     youtube_channel_refresh_interval: float = 600

tests/test_youtube_subscription.py

@@ -6,6 +6,7 @@ from breadtube_bot.youtube_subscription import SubscriptionHelper
 def test_helper():
     subscriptions = SubscriptionHelper.read_text(Path('tests/data/subscriptions.csv').read_bytes())
     assert subscriptions['UUlDye0T4xTEq46sA5evq9fA'].name == 'a-gauche'
+    assert subscriptions['UUkKYLbCdxz3KyNKDzfrL20g'].name == 'abal'
     assert subscriptions['UUFrDDP81MX_QfOHrRZOgD4g'].name == 'l214'
     assert subscriptions['UUe2Lt08V5TVudUa86OzRm3w'].name == 'irascible'
     assert subscriptions['UUHXyS9njDTc-HbnfRr1k6uA'].name == 'praxis-par-francois-boulo'