feat: Telegram Emoji status rule

backend/api/pos/chat.py

@@ -214,6 +214,11 @@ class TelegramChatPremiumRuleCPO(BaseFDO):
     is_enabled: bool
 
 
+class TelegramChatEmojiRuleCPO(BaseFDO):
+    is_enabled: bool
+    emoji_id: str
+
+
 class ChatEligibilityRuleFDO(BaseFDO, ChatEligibilityRuleDTO):
     @field_serializer("expected", return_type=float | int)
     def preprocess_expected(self, v: int) -> float | int:

backend/api/routes/admin/chat/rule/__init__.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter
 
+from api.routes.admin.chat.rule.emoji import manage_emoji_rules_router
 from api.routes.admin.chat.rule.jetton import manage_jetton_rules_router
 from api.routes.admin.chat.rule.nft import manage_nft_collection_rules_router
 from api.routes.admin.chat.rule.premium import manage_premium_rules_router
@@ -15,6 +16,7 @@
 manage_rules_router.include_router(manage_nft_collection_rules_router)
 manage_rules_router.include_router(manage_toncoin_rules_router)
 manage_rules_router.include_router(manage_premium_rules_router)
+manage_rules_router.include_router(manage_emoji_rules_router)
 manage_rules_router.include_router(manage_sticker_rules_router)
 manage_rules_router.include_router(manage_whitelist_rules_router)
 manage_rules_router.include_router(manage_external_whitelist_rules_router)

backend/api/routes/admin/chat/rule/emoji.py

@@ -0,0 +1,118 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from starlette.requests import Request
+from starlette.status import HTTP_200_OK, HTTP_404_NOT_FOUND, HTTP_400_BAD_REQUEST
+
+from api.deps import get_db_session
+from api.pos.base import BaseExceptionFDO
+from api.pos.chat import ChatEligibilityRuleFDO, TelegramChatEmojiRuleCPO
+from core.actions.chat.rule.emoji import TelegramChatEmojiAction
+
+manage_emoji_rules_router = APIRouter(prefix="/emoji")
+
+
+@manage_emoji_rules_router.get(
+    "/{rule_id}",
+    responses={
+        HTTP_200_OK: {"model": ChatEligibilityRuleFDO},
+        HTTP_404_NOT_FOUND: {
+            "description": "Rule Not Found",
+            "model": BaseExceptionFDO,
+        },
+    },
+)
+async def get_emoji_rule(
+    request: Request,
+    slug: str,
+    rule_id: int,
+    db_session: Session = Depends(get_db_session),
+) -> ChatEligibilityRuleFDO:
+    action = TelegramChatEmojiAction(
+        requestor=request.state.user,
+        chat_slug=slug,
+        db_session=db_session,
+    )
+    return ChatEligibilityRuleFDO.model_validate(
+        action.read(rule_id=rule_id).model_dump()
+    )
+
+
+@manage_emoji_rules_router.post(
+    "",
+    responses={
+        HTTP_200_OK: {"model": ChatEligibilityRuleFDO},
+        HTTP_400_BAD_REQUEST: {
+            "description": "Occurs if Telegram Emoji rule already exists for the chat",
+            "model": BaseExceptionFDO,
+        },
+    },
+)
+async def add_emoji_rule(
+    request: Request,
+    slug: str,
+    rule: TelegramChatEmojiRuleCPO,
+    db_session: Session = Depends(get_db_session),
+) -> ChatEligibilityRuleFDO:
+    action = TelegramChatEmojiAction(
+        requestor=request.state.user,
+        chat_slug=slug,
+        db_session=db_session,
+    )
+    return ChatEligibilityRuleFDO.model_validate(
+        action.create(
+            emoji_id=rule.emoji_id,
+        ).model_dump()
+    )
+
+
+@manage_emoji_rules_router.put(
+    "/{rule_id}",
+    responses={
+        HTTP_200_OK: {"model": ChatEligibilityRuleFDO},
+        HTTP_404_NOT_FOUND: {
+            "description": "Rule Not Found",
+            "model": BaseExceptionFDO,
+        },
+    },
+)
+async def update_emoji_rule(
+    request: Request,
+    slug: str,
+    rule_id: int,
+    rule: TelegramChatEmojiRuleCPO,
+    db_session: Session = Depends(get_db_session),
+) -> ChatEligibilityRuleFDO:
+    action = TelegramChatEmojiAction(
+        requestor=request.state.user,
+        chat_slug=slug,
+        db_session=db_session,
+    )
+    return ChatEligibilityRuleFDO.model_validate(
+        action.update(
+            rule_id=rule_id, emoji_id=rule.emoji_id, is_enabled=rule.is_enabled
+        ).model_dump()
+    )
+
+
+@manage_emoji_rules_router.delete(
+    "/{rule_id}",
+    responses={
+        HTTP_200_OK: {"model": BaseExceptionFDO},
+        HTTP_404_NOT_FOUND: {
+            "description": "Rule Not Found",
+            "model": BaseExceptionFDO,
+        },
+    },
+)
+async def delete_emoji_rule(
+    request: Request,
+    slug: str,
+    rule_id: int,
+    db_session: Session = Depends(get_db_session),
+) -> None:
+    action = TelegramChatEmojiAction(
+        requestor=request.state.user,
+        chat_slug=slug,
+        db_session=db_session,
+    )
+    action.delete(rule_id=rule_id)

backend/core/actions/authorization.py

@@ -26,6 +26,7 @@
 )
 from core.models.rule import TelegramChatWhitelistExternalSource, TelegramChatWhitelist
 from core.services.chat import TelegramChatService
+from core.services.chat.rule.emoji import TelegramChatEmojiService
 from core.services.chat.rule.premium import TelegramChatPremiumService
 from core.services.chat.rule.sticker import TelegramChatStickerCollectionService
 from core.services.chat.rule.whitelist import (
@@ -79,6 +80,7 @@ def __init__(
         self.telegram_chat_sticker_collection_service = (
             TelegramChatStickerCollectionService(db_session)
         )
+        self.telegram_chat_emoji_service = TelegramChatEmojiService(db_session)
         self.telethon_service = TelethonService(client=telethon_client)
 
     def is_user_eligible_chat_member(
@@ -168,6 +170,9 @@ def get_eligibility_rules(
         all_premium_rules = self.telegram_chat_premium_service.get_all(
             chat_id, enabled_only=enabled_only
         )
+        all_emoji_rules = self.telegram_chat_emoji_service.get_all(
+            chat_id, enabled_only=enabled_only
+        )
         all_sticker_rules = self.telegram_chat_sticker_collection_service.get_all(
             chat_id, enabled_only=enabled_only
         )
@@ -179,6 +184,7 @@ def get_eligibility_rules(
             whitelist_external_sources=all_external_source_rules,
             whitelist_sources=all_whitelist_groups,
             premium=all_premium_rules,
+            emoji=all_emoji_rules,
         )
 
     def get_ineligible_chat_members(
@@ -422,6 +428,19 @@ def check_chat_member_eligibility(
                 for rule in eligibility_rules.stickers
             ]
         )
+        items.extend(
+            [
+                EligibilitySummaryInternalDTO(
+                    id=rule.id,
+                    type=EligibilityCheckType.EMOJI,
+                    expected=1,
+                    title="Emoji Status",
+                    actual=1,
+                    is_enabled=rule.is_enabled,
+                )
+                for rule in eligibility_rules.emoji
+            ]
+        )
         return RulesEligibilitySummaryInternalDTO(
             items=items,
             is_admin=bool(chat_member and chat_member.is_admin),

backend/core/actions/chat/__init__.py

@@ -170,7 +170,7 @@ async def _load_participants(self, chat_identifier: int) -> None:
 
     async def index(self, chat: ChatPeerType) -> None:
         """
-        Handles the process of creating and refreshing a Telegram chat invite link,
+        Handles the process of creating and refreshing a Telegram chat invite link
         and loading the participants for the given chat. If the chat already has an
         invite link, it skips the creation process.
 
@@ -285,19 +285,22 @@ async def refresh_all(self) -> None:
             Raised if a chat does not exist because it was deleted or the bot was
             removed from the chat.
         :raises TelegramChatNotSufficientPrivileges:
-            Raised if the bot lacks sufficient privileges to function in the chat.
+            Raised if the bot lacks enough privileges to function in the chat.
 
         :return: This function does not return a value as its primary purpose is to
             refresh all accessible chats.
         """
-        for chat in self.telegram_chat_service.get_all():
+        for chat in self.telegram_chat_service.get_all(
+            enabled_only=True,
+            sufficient_privileges_only=True,
+        ):
             try:
                 await self._refresh(chat)
-            except (
-                TelegramChatNotExists,  # happens when chat is deleted or bot is removed from the chat
-                TelegramChatNotSufficientPrivileges,  # happens when bot has no rights to function in the chat
-            ):
-                continue
+            except Exception as e:
+                logger.exception(
+                    f"Unexpected error occurred while refreshing chat {chat.id!r}",
+                    exc_info=e,
+                )
 
     async def _refresh(self, chat: TelegramChat) -> TelegramChat:
         """
@@ -499,6 +502,10 @@ async def get_with_eligibility_rules(self) -> TelegramChatWithRulesDTO:
                         StickerChatEligibilityRuleDTO.from_orm(rule)
                         for rule in eligibility_rules.stickers
                     ),
+                    *(
+                        ChatEligibilityRuleDTO.from_emoji_rule(rule)
+                        for rule in eligibility_rules.emoji
+                    ),
                 ],
                 key=lambda rule: (not rule.is_enabled, rule.type.value, rule.title),
             ),

backend/core/actions/chat/rule/emoji.py

@@ -0,0 +1,64 @@
+import logging
+
+from fastapi import HTTPException
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import Session
+from starlette.status import HTTP_404_NOT_FOUND, HTTP_400_BAD_REQUEST
+
+from core.actions.chat import ManagedChatBaseAction
+from core.dtos.chat.rules import ChatEligibilityRuleDTO
+from core.models.user import User
+from core.services.chat.rule.emoji import TelegramChatEmojiService
+
+
+logger = logging.getLogger(__name__)
+
+
+class TelegramChatEmojiAction(ManagedChatBaseAction):
+    def __init__(self, db_session: Session, requestor: User, chat_slug: str):
+        super().__init__(db_session, requestor, chat_slug)
+        self.service = TelegramChatEmojiService(db_session)
+
+    def read(self, rule_id: int) -> ChatEligibilityRuleDTO:
+        try:
+            rule = self.service.get(chat_id=self.chat.id, rule_id=rule_id)
+        except NoResultFound:
+            raise HTTPException(
+                detail="Rule not found",
+                status_code=HTTP_404_NOT_FOUND,
+            )
+        return ChatEligibilityRuleDTO.from_emoji_rule(rule)
+
+    def create(self, emoji_id: str) -> ChatEligibilityRuleDTO:
+        if self.service.exists(chat_id=self.chat.id):
+            raise HTTPException(
+                detail="Telegram Emoji rule already exists for that chat. Please, modify it instead.",
+                status_code=HTTP_400_BAD_REQUEST,
+            )
+
+        rule = self.service.create(chat_id=self.chat.id, emoji_id=emoji_id)
+        logger.info(f"New Telegram Emoji rule created for the chat {self.chat.id!r}.")
+        return ChatEligibilityRuleDTO.from_emoji_rule(rule)
+
+    def update(
+        self, rule_id: int, emoji_id: str, is_enabled: bool
+    ) -> ChatEligibilityRuleDTO:
+        try:
+            rule = self.service.get(chat_id=self.chat.id, rule_id=rule_id)
+        except NoResultFound:
+            raise HTTPException(
+                detail="Rule not found",
+                status_code=HTTP_404_NOT_FOUND,
+            )
+
+        self.service.update(rule=rule, emoji_id=emoji_id, is_enabled=is_enabled)
+        logger.info(
+            f"Updated Telegram Emoji rule {rule_id!r} for the chat {self.chat.id!r}."
+        )
+        return ChatEligibilityRuleDTO.from_emoji_rule(rule)
+
+    def delete(self, rule_id: int) -> None:
+        self.service.delete(chat_id=self.chat.id, rule_id=rule_id)
+        logger.info(
+            f"Deleted Telegram Emoji rule {rule_id!r} for the chat {self.chat.id!r}."
+        )

backend/core/dtos/chat/rules/__init__.py

@@ -18,6 +18,7 @@
     TelegramChatWhitelist,
     TelegramChatPremium,
     TelegramChatStickerCollection,
+    TelegramChatEmoji,
 )
 from core.models.rule import TelegramChatToncoin
 
@@ -30,6 +31,7 @@ class EligibilityCheckType(enum.Enum):
     WHITELIST = "whitelist"
     PREMIUM = "premium"
     STICKER_COLLECTION = "sticker_collection"
+    EMOJI = "emoji"
 
 
 @dataclasses.dataclass
@@ -41,6 +43,7 @@ class TelegramChatEligibilityRulesDTO:
     premium: list[TelegramChatPremium]
     whitelist_external_sources: list[TelegramChatWhitelistExternalSource]
     whitelist_sources: list[TelegramChatWhitelist]
+    emoji: list[TelegramChatEmoji]
 
 
 class ChatEligibilityRuleDTO(BaseModel):
@@ -134,6 +137,18 @@ def from_premium_rule(cls, rule: TelegramChatPremium) -> Self:
             is_enabled=rule.is_enabled,
         )
 
+    @classmethod
+    def from_emoji_rule(cls, rule: TelegramChatEmoji) -> Self:
+        return cls(
+            id=rule.id,
+            type=EligibilityCheckType.STICKER_COLLECTION,
+            title="Emoji Status",
+            expected=1,
+            photo_url=None,
+            blockchain_address=None,
+            is_enabled=rule.is_enabled,
+        )
+
 
 class TelegramChatWithRulesDTO(BaseModel):
     chat: TelegramChatDTO