fix: improve admin status detection for chat participants

backend/community_manager/actions/chat.py

@@ -18,6 +18,7 @@
 from community_manager.dtos.chat import TargetChatMembersDTO
 from community_manager.events import ChatAdminChangeEventBuilder
 from community_manager.settings import community_manager_settings
+from community_manager.utils import is_chat_participant_manager_admin
 from core.actions.authorization import AuthorizationAction
 from core.actions.base import BaseAction
 from core.actions.user import UserAction
@@ -144,7 +145,9 @@ async def _load_participants(
             self.telegram_chat_user_service.create_or_update(
                 chat_id=chat_identifier,
                 user_id=user.id,
-                is_admin=hasattr(participant_user.participant, "admin_rights"),
+                is_admin=is_chat_participant_manager_admin(
+                    participant_user.participant
+                ),
                 is_managed=False,
             )
             processed_user_ids.append(user.id)
@@ -1100,12 +1103,13 @@ async def kick_chat_member(self, chat_member: TelegramChatUser) -> None:
                 f"Failed to kick user {chat_member.user.telegram_id!r} from chat {chat_member.chat_id!r} as bot user lacks admin privileges",
                 exc_info=e,
             )
-            self.telegram_chat_service.set_insufficient_privileges(
-                chat_id=chat_member.chat_id, value=True
-            )
-            logger.info(
-                f"Set insufficient privileges flag for chat {chat_member.chat_id!r}."
-            )
+            # TODO: Fix this logic after telegram_chat_user.is_manager_admin/is_admin is properly indexed
+            # self.telegram_chat_service.set_insufficient_privileges(
+            #     chat_id=chat_member.chat_id, value=True
+            # )
+            # logger.info(
+            #     f"Set insufficient privileges flag for chat {chat_member.chat_id!r}."
+            # )
         except RPCError as e:
             logger.error(
                 f"Failed to kick user {chat_member.user.telegram_id!r} from chat {chat_member.chat_id!r}",

backend/community_manager/events/__init__.py

@@ -10,11 +10,14 @@
     ChannelParticipantBanned,
     ChannelParticipantSelf,
     User as TelethonUser,
-    ChannelParticipantCreator,
     ChatInviteExported,
 )
 
-from core.constants import REQUIRED_ADMIN_PRIVILEGES, REQUIRED_BOT_PRIVILEGES
+from community_manager.utils import (
+    is_chat_participant_admin,
+    is_chat_participant_manager_admin,
+)
+from core.constants import REQUIRED_BOT_PRIVILEGES
 
 logger = logging.getLogger(__name__)
 
@@ -121,18 +124,15 @@ def user(self) -> TelethonUser | None:
 
         @property
         def is_demoted(self) -> bool:
-            return isinstance(
-                self.prev_participant, ChannelParticipantAdmin
-            ) and not isinstance(
-                self.new_participant,
-                (ChannelParticipantAdmin, ChannelParticipantCreator),
-            )
+            return is_chat_participant_admin(
+                chat_participant=self.prev_participant
+            ) and not is_chat_participant_admin(chat_participant=self.new_participant)
 
         @property
         def is_promoted(self) -> bool:
-            return not isinstance(
-                self.prev_participant, ChannelParticipantAdmin
-            ) and isinstance(self.new_participant, ChannelParticipantAdmin)
+            return not is_chat_participant_admin(
+                chat_participant=self.prev_participant
+            ) and is_chat_participant_admin(chat_participant=self.new_participant)
 
         @property
         def has_enough_rights(self) -> bool:
@@ -142,28 +142,7 @@ def has_enough_rights(self) -> bool:
                 )
                 return False
 
-            if not isinstance(self.new_participant, ChannelParticipantAdmin):
-                logger.debug(
-                    "User %d is not an admin in the chat %d",
-                    self.user.id,
-                    self.original_update.channel_id,
-                )
-                return False
-            elif not all(
-                [
-                    getattr(self.new_participant.admin_rights, right)
-                    for right in REQUIRED_ADMIN_PRIVILEGES
-                ]
-            ):
-                logger.warning(
-                    "User %d has insufficient permissions in the chat %d: %s",
-                    self.user.id,
-                    self.original_update.channel_id,
-                    self.new_participant.admin_rights,
-                )
-                return False
-
-            return True
+            return is_chat_participant_manager_admin(self.new_participant)
 
         @property
         def sufficient_bot_privileges(self) -> bool:

backend/community_manager/utils.py

@@ -0,0 +1,40 @@
+from telethon.tl.types import (
+    ChannelParticipantAdmin,
+    ChannelParticipantCreator,
+    ChannelParticipant,
+)
+
+from core.constants import REQUIRED_ADMIN_PRIVILEGES
+
+
+def is_chat_participant_admin(
+    chat_participant: ChannelParticipant,
+) -> bool:
+    """
+    Check if the given chat participant is an admin or creator.
+    :param chat_participant: The chat participant to check.
+    :return: True if the participant is an admin or creator, False otherwise.
+    """
+    return isinstance(
+        chat_participant,
+        (ChannelParticipantAdmin, ChannelParticipantCreator),
+    )
+
+
+def is_chat_participant_manager_admin(
+    chat_participant: ChannelParticipant
+    | ChannelParticipantAdmin
+    | ChannelParticipantCreator,
+) -> bool:
+    """
+    Check if the given chat participant is an admin with manager privileges.
+    :param chat_participant: The chat participant to check.
+    :return: True if the participant is an admin with manager privileges, False otherwise.
+    """
+    if not is_chat_participant_admin(chat_participant):
+        return False
+
+    return all(
+        getattr(chat_participant.admin_rights, right)
+        for right in REQUIRED_ADMIN_PRIVILEGES
+    )