feat: Add emoji logo support in Telegram Chat Emoji rules

backend/api/pos/chat.py

@@ -249,7 +249,7 @@ class TelegramChatPremiumRuleCPO(BaseFDO):
 
 class TelegramChatEmojiRuleCPO(BaseFDO):
     is_enabled: bool
-    emoji_id: str
+    emoji_id: int
 
 
 class ChatEligibilityRuleFDO(BaseFDO, ChatEligibilityRuleDTO):
@@ -300,7 +300,7 @@ class GiftChatEligibilitySummaryFDO(BaseFDO, GiftChatEligibilitySummaryDTO):
 
 
 class EmojiChatEligibilityRuleFDO(BaseFDO, EmojiChatEligibilityRuleDTO):
-    ...
+    photo_url: CDNImageField
 
 
 class EmojiChatEligibilitySummaryFDO(BaseFDO, EmojiChatEligibilitySummaryDTO):

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

@@ -58,11 +58,8 @@ async def add_emoji_rule(
         chat_slug=slug,
         db_session=db_session,
     )
-    return EmojiChatEligibilityRuleFDO.model_validate(
-        action.create(
-            emoji_id=rule.emoji_id,
-        ).model_dump()
-    )
+    result = await action.create(emoji_id=rule.emoji_id)
+    return EmojiChatEligibilityRuleFDO.model_validate(result.model_dump())
 
 
 @manage_emoji_rules_router.put(
@@ -87,11 +84,10 @@ async def update_emoji_rule(
         chat_slug=slug,
         db_session=db_session,
     )
-    return EmojiChatEligibilityRuleFDO.model_validate(
-        action.update(
-            rule_id=rule_id, emoji_id=rule.emoji_id, is_enabled=rule.is_enabled
-        ).model_dump()
+    result = await action.update(
+        rule_id=rule_id, emoji_id=rule.emoji_id, is_enabled=rule.is_enabled
     )
+    return EmojiChatEligibilityRuleFDO.model_validate(result.model_dump())
 
 
 @manage_emoji_rules_router.delete(

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

@@ -1,23 +1,94 @@
 import logging
+from tempfile import NamedTemporaryFile
 
 from fastapi import HTTPException
+from pydantic import BaseModel
 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.emoji import EmojiChatEligibilityRuleDTO
 from core.models.user import User
+from core.services.cdn import CDNService
 from core.services.chat.rule.emoji import TelegramChatEmojiService
-
+from core.services.superredis import RedisService
+from core.services.supertelethon import TelethonService
+from core.settings import core_settings
 
 logger = logging.getLogger(__name__)
 
 
+class StickerSetDTO(BaseModel):
+    id: int
+    access_hash: int
+    title: str
+    short_name: str
+    text_color: bool
+
+
+class EmojiItemDTO(BaseModel):
+    id: int
+    access_hash: int
+    mime_type: str
+    logo_url: str
+    sticker_set: StickerSetDTO
+
+
+EMOJI_CACHE_KEY_TEMPLATE = "emoji-metadata:{emoji_id}"
+EMOJI_CACHE_TTL = 60 * 60 * 24 * 30 * 12 * 10  # 10 years
+
+
 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)
+        self.telethon_service = TelethonService(
+            bot_token=core_settings.telegram_bot_token
+        )
+        self.redis_service = RedisService()
+        self.cdn_service = CDNService()
+
+    async def get_cached_emoji_data(self, emoji_id: int) -> EmojiItemDTO:
+        emoji_cache_key = EMOJI_CACHE_KEY_TEMPLATE.format(emoji_id=emoji_id)
+        if cached_data := self.redis_service.get(emoji_cache_key):
+            return EmojiItemDTO.model_validate_json(cached_data)
+
+        await self.telethon_service.start()
+
+        emoji, sticker_set = await self.telethon_service.index_emoji(emoji_id=emoji_id)
+        with NamedTemporaryFile(mode="w+b", delete=True) as f:
+            logo_path = await self.telethon_service.download_emoji(
+                emoji=emoji, target_location=f
+            )
+            if logo_path:
+                await self.cdn_service.upload_file(
+                    file_path=f.name,
+                    object_name=logo_path,
+                )
+                logger.info(f"Uploaded new emoji logo to {logo_path!r}.")
+
+        await self.telethon_service.stop()
+
+        emoji_item = EmojiItemDTO(
+            id=emoji.id,
+            access_hash=emoji.access_hash,
+            mime_type=emoji.mime_type,
+            logo_url=logo_path,
+            sticker_set=StickerSetDTO(
+                id=sticker_set.id,
+                access_hash=sticker_set.access_hash,
+                title=sticker_set.title,
+                short_name=sticker_set.short_name,
+                text_color=sticker_set.text_color,
+            ),
+        )
+        self.redis_service.set(
+            emoji_cache_key,
+            emoji_item.model_dump_json(),
+            EMOJI_CACHE_TTL,
+        )
+        return emoji_item
 
     def read(self, rule_id: int) -> EmojiChatEligibilityRuleDTO:
         try:
@@ -29,19 +100,22 @@ def read(self, rule_id: int) -> EmojiChatEligibilityRuleDTO:
             )
         return EmojiChatEligibilityRuleDTO.from_orm(rule)
 
-    def create(self, emoji_id: str) -> EmojiChatEligibilityRuleDTO:
+    async def create(self, emoji_id: int) -> EmojiChatEligibilityRuleDTO:
         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)
+        emoji_data = await self.get_cached_emoji_data(emoji_id=emoji_id)
+        rule = self.service.create(
+            chat_id=self.chat.id, emoji_id=emoji_id, logo_url=emoji_data.logo_url
+        )
         logger.info(f"New Telegram Emoji rule created for the chat {self.chat.id!r}.")
         return EmojiChatEligibilityRuleDTO.from_orm(rule)
 
-    def update(
-        self, rule_id: int, emoji_id: str, is_enabled: bool
+    async def update(
+        self, rule_id: int, emoji_id: int, is_enabled: bool
     ) -> EmojiChatEligibilityRuleDTO:
         try:
             rule = self.service.get(chat_id=self.chat.id, rule_id=rule_id)
@@ -51,7 +125,13 @@ def update(
                 status_code=HTTP_404_NOT_FOUND,
             )
 
-        self.service.update(rule=rule, emoji_id=emoji_id, is_enabled=is_enabled)
+        emoji_data = await self.get_cached_emoji_data(emoji_id=emoji_id)
+        self.service.update(
+            rule=rule,
+            emoji_id=emoji_id,
+            is_enabled=is_enabled,
+            logo_url=emoji_data.logo_url,
+        )
         logger.info(
             f"Updated Telegram Emoji rule {rule_id!r} for the chat {self.chat.id!r}."
         )

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

@@ -15,7 +15,7 @@ def from_orm(cls, obj: TelegramChatEmoji) -> Self:
             type=EligibilityCheckType.EMOJI,
             title=obj.emoji_id,
             expected=1,
-            photo_url=None,
+            photo_url=obj.logo_url,
             blockchain_address=None,
             is_enabled=obj.is_enabled,
             emoji_id=obj.emoji_id,

backend/core/migrations/versions/1749582058-a02acc5ddb6e-emoji_logo_url.py

@@ -0,0 +1,29 @@
+"""emoji logo url
+
+Revision ID: a02acc5ddb6e
+Revises: 53b283880538
+Create Date: 2025-06-10 19:00:58.556112
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "a02acc5ddb6e"
+down_revision: Union[str, None] = "53b283880538"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "telegram_chat_emoji",
+        sa.Column("logo_url", sa.String(length=255), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("telegram_chat_emoji", "logo_url")

backend/core/models/rule.py

@@ -45,6 +45,7 @@ class TelegramChatEmoji(Base):
         nullable=False,
     )
     emoji_id = mapped_column(String(255), nullable=False)
+    logo_url = mapped_column(String(255), nullable=True)
     is_enabled = mapped_column(Boolean, nullable=False, default=False)
     created_at = mapped_column(
         DateTime(timezone=True), server_default=func.now(), nullable=False

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

@@ -32,23 +32,30 @@ def exists(self, chat_id: int) -> bool:
             > 0
         )
 
-    def create(self, chat_id: int, emoji_id: str) -> TelegramChatEmoji:
+    def create(
+        self, chat_id: int, emoji_id: int, logo_url: str | None
+    ) -> TelegramChatEmoji:
         if self.exists(chat_id):
             raise TelegramChatRuleExists(
                 "Telegram Chat rule of that type for that chat already exists."
             )
         new_rule = TelegramChatEmoji(
-            chat_id=chat_id, emoji_id=emoji_id, is_enabled=True
+            chat_id=chat_id, emoji_id=str(emoji_id), is_enabled=True, logo_url=logo_url
         )
         self.db_session.add(new_rule)
         self.db_session.commit()
         return new_rule
 
     def update(
-        self, rule: TelegramChatEmoji, emoji_id: str, is_enabled: bool
+        self,
+        rule: TelegramChatEmoji,
+        emoji_id: int,
+        is_enabled: bool,
+        logo_url: str | None,
     ) -> TelegramChatEmoji:
         rule.emoji_id = emoji_id
         rule.is_enabled = is_enabled
+        rule.logo_url = logo_url
         self.db_session.commit()
         return rule
 

backend/core/services/supertelethon.py

@@ -30,6 +30,7 @@
     Document,
     DocumentAttributeCustomEmoji,
     StickerSet,
+    DocumentAttributeFilename,
 )
 from telethon.tl.types.payments import SavedStarGifts
 
@@ -229,6 +230,24 @@ async def download_unique_gift_thumbnail(
         )
         return f"{entity.slug}-preview.png"
 
+    async def download_emoji(
+        self,
+        emoji: Document,
+        target_location: BinaryIO | IO[bytes],
+    ) -> str:
+        """
+        Download the emoji document and save it to the specified location.
+        """
+        filename = next(
+            filter(lambda a: isinstance(a, DocumentAttributeFilename), emoji.attributes)
+        ).file_name
+        suffix = Path(filename).suffix
+        await self.client.download_media(
+            message=emoji,
+            file=target_location,  # type: ignore
+        )
+        return f"{emoji.id}{suffix}"
+
     async def promote_user(
         self, chat_id: int, telegram_user_id: int, custom_title: str
     ) -> None:
@@ -317,7 +336,10 @@ async def index_emoji(self, emoji_id: int) -> tuple[Document, StickerSet]:
         sticker_set = await self.client(
             GetStickerSetRequest(stickerset=sticker_set_input, hash=0)
         )
-        return emoji, sticker_set
+        logger.info(
+            f"Indexed emoji {emoji.id!r} and sticker set {sticker_set.set.id!r}."
+        )
+        return emoji, sticker_set.set
 
     async def index_gift(self, slug: str, number: int) -> StarGiftUnique:
         """