Merge pull request #51 from OpenBuilders/feat/stickerdom-integration

feat: StickerDom intergration

Author: danoctua

backend/api/pos/sticker.py

@@ -0,0 +1,8 @@
+from api.pos.base import BaseFDO
+from core.dtos.sticker import MinimalStickerCollectionWithCharactersDTO
+
+
+class MinimalStickerCollectionWithCharactersFDO(
+    BaseFDO, MinimalStickerCollectionWithCharactersDTO
+):
+    ...

backend/api/routes/admin/resource.py

@@ -13,10 +13,10 @@
     NftCollectionFDO,
 )
 from api.pos.common import StatusFDO, CategoriesFDO
+from api.pos.sticker import MinimalStickerCollectionWithCharactersFDO
 from core.actions.jetton import JettonAction
 from core.actions.nft_collection import NftCollectionAction
 from core.actions.sticker import StickerCharacterAction
-from core.dtos.sticker import MinimalStickerCollectionWithCharactersDTO
 from core.enums.jetton import CurrencyCategory
 from core.enums.nft import ASSET_TO_CATEGORY_TYPE_MAPPING
 from core.exceptions.external import ExternalResourceNotFound
@@ -84,9 +84,16 @@ async def get_currency_categories() -> list[CategoriesFDO]:
 @admin_resource_router.get("/stickers")
 async def get_stickers_collections(
     db_session: Session = Depends(get_db_session),
-) -> list[MinimalStickerCollectionWithCharactersDTO]:
+) -> list[MinimalStickerCollectionWithCharactersFDO]:
     action = StickerCharacterAction(db_session)
-    return await action.get_all_grouped()
+    collections = await action.get_all_grouped()
+
+    return [
+        MinimalStickerCollectionWithCharactersFDO.model_validate(
+            collection.model_dump()
+        )
+        for collection in collections
+    ]
 
 
 @admin_resource_router.get("/jettons", deprecated=True)

backend/community_manager/actions/__init__.py

@@ -0,0 +1,144 @@
+import logging
+
+from sqlalchemy.orm import Session
+
+from community_manager.dtos.chat import TargetChatMembersDTO
+from community_manager.entrypoint import init_client
+from community_manager.settings import community_manager_settings
+from core.actions.authorization import AuthorizationAction
+from core.constants import UPDATED_WALLETS_SET_NAME, UPDATED_STICKERS_USER_IDS
+from core.services.chat.rule.sticker import TelegramChatStickerCollectionService
+from core.services.chat.user import TelegramChatUserService
+from core.services.superredis import RedisService
+
+
+logger = logging.getLogger(__name__)
+
+
+class CommunityManagerChatAction:
+    def __init__(self, db_session: Session):
+        self.db_session = db_session
+        self.telegram_chat_user_service = TelegramChatUserService(db_session)
+        self.telegram_chat_sticker_collection_service = (
+            TelegramChatStickerCollectionService(db_session)
+        )
+        self.redis_service = RedisService()
+
+    def get_updated_chat_members(self) -> TargetChatMembersDTO:
+        """
+        Fetches and updates the target chat members based on specific criteria including
+        linked wallets and sticker owners. The method retrieves updated wallet addresses
+        and sticker owner IDs from the Redis service, identifies relevant chat members
+        from the Telegram chat services, and compiles the processed data into a
+        `TargetChatMembersDTO` object.
+
+        :raises ValueError: If unexpected data types are retrieved or processed in the logic.
+
+        :return: A `TargetChatMembersDTO` object containing the updated wallet addresses,
+            sticker owner IDs, and the compiled target chat members.
+        """
+        wallets = self.redis_service.pop_from_set(
+            name=UPDATED_WALLETS_SET_NAME,
+            count=community_manager_settings.items_per_task,
+        )
+        if isinstance(wallets, str):
+            wallets = [wallets]
+
+        sticker_owners_ids = (
+            self.redis_service.pop_from_set(
+                name=UPDATED_STICKERS_USER_IDS,
+                count=community_manager_settings.items_per_task,
+            )
+            or []
+        )
+        if isinstance(sticker_owners_ids, str):
+            sticker_owners_ids = [sticker_owners_ids]
+            sticker_owners_ids = set(map(int, sticker_owners_ids))
+
+        target_chat_members: set[tuple[int, int]] = set()
+
+        if wallets:
+            chat_members = self.telegram_chat_user_service.get_all_by_linked_wallet(
+                addresses=wallets
+            )
+            target_chat_members.update(
+                {(cm.chat_id, cm.user_id) for cm in chat_members}
+            )
+
+        if sticker_owners_ids:
+            rules = self.telegram_chat_sticker_collection_service.get_all(
+                enabled_only=True
+            )
+            unique_chat_ids = {r.chat_id for r in rules}
+            target_chat_members.update(
+                {
+                    (chat_id, user_id)
+                    for chat_id in unique_chat_ids
+                    for user_id in sticker_owners_ids
+                }
+            )
+
+        return TargetChatMembersDTO(
+            wallets=wallets,
+            sticker_owners_ids=sticker_owners_ids,
+            target_chat_members=target_chat_members,
+        )
+
+    async def sanity_chat_checks(self) -> None:
+        """
+        Performs sanity checks on chat members and validates their eligibility. If there are
+        any chat members to validate, it initiates the validation process with the help of
+        a Telegram service client. Ineligible members are removed based on the validation
+        logic. If an error occurs during validation, a fallback mechanism is triggered
+        to add wallets and users back to the redis database to try again later.
+
+        The method logs the progress at various stages and handles exceptions to ensure
+        fallback processes are executed if needed.
+
+        :raises Exception: If validation of chat members fails during execution.
+        """
+        dto = self.get_updated_chat_members()
+        if target_chat_members := dto.target_chat_members:
+            try:
+                logger.info(f"Validating chat members for {target_chat_members}")
+                chat_members = self.telegram_chat_user_service.get_all_pairs(
+                    chat_member_pairs=target_chat_members
+                )
+
+                if not chat_members:
+                    logger.info("No chats to validate. Skipping")
+                    return
+                else:
+                    logger.info(f"Found {len(chat_members)} chat members to validate")
+
+                telethon_service = init_client()
+                authorization_action = AuthorizationAction(
+                    self.db_session, telethon_client=telethon_service.client
+                )
+                await authorization_action.kick_ineligible_chat_members(
+                    chat_members=chat_members
+                )
+                logger.info(
+                    f"Successfully validated {len(chat_members)} chat members. "
+                )
+            except Exception as exc:
+                logger.error(f"Failed to validate chat members: {exc}", exc_info=True)
+                self.fallback_update_chat_members(dto=dto)
+                raise exc
+        else:
+            logger.info("No users to validate. Skipping")
+
+    def fallback_update_chat_members(self, dto: TargetChatMembersDTO) -> None:
+        """
+        Activates a fallback mechanism to update chat members by storing provided wallets
+        and sticker owner IDs in Redis sets. This ensures that the required updates are
+        persisted and managed separately if the primary update mechanism fails.
+
+        :param dto: A data transfer object containing the wallets and sticker owner IDs
+            to be updated in Redis sets.
+        """
+        logger.warning("Activating fallback method for chat members.")
+        self.redis_service.add_to_set(UPDATED_WALLETS_SET_NAME, *dto.wallets)
+        self.redis_service.add_to_set(
+            UPDATED_STICKERS_USER_IDS, *map(str, dto.sticker_owners_ids)
+        )

backend/community_manager/actions/chat.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+
+
+class TargetChatMembersDTO(BaseModel):
+    wallets: list[str]
+    sticker_owners_ids: list[int]
+    target_chat_members: set[tuple[int, int]]

backend/community_manager/dtos/__init__.py

@@ -17,14 +17,14 @@
 
 
 def init_client():
-    # Make session persistent for community manager
+    # Make session persistent for the community manager
     session_path = Path(__file__).parent.parent / "data"
     if not session_path.exists():
         raise FileNotFoundError(
             f"Session path {session_path} does not exist. Please create it."
         )
 
-    # This session is not thread-safe, and cannot be used by multiple clients at the same time.
+    # This session is not thread-safe and cannot be used by multiple clients at the same time.
     session = SQLiteSession(str(session_path / "gateway.session"))
     client = TelegramClient(
         session,

backend/community_manager/dtos/chat.py

@@ -2,6 +2,7 @@
 
 from celery.utils.log import get_task_logger
 
+from community_manager.actions.chat import CommunityManagerChatAction
 from community_manager.celery_app import app
 from community_manager.entrypoint import init_client
 from community_manager.settings import community_manager_settings
@@ -12,25 +13,26 @@
 )
 from core.constants import (
     CELERY_SYSTEM_QUEUE_NAME,
-    UPDATED_WALLETS_SET_NAME,
 )
 from core.services.chat.user import TelegramChatUserService
 from core.services.db import DBService
-from core.services.superredis import RedisService
 
 logger = get_task_logger(__name__)
 
 
-async def sanity_chat_checks(wallets: list[str] | None) -> None:
-    logger.info(f"Validating chat members for {wallets}")
+async def sanity_chat_checks(target_chat_members: set[tuple[int, int]]) -> None:
+    logger.info(f"Validating chat members for {target_chat_members}")
     with DBService().db_session() as db_session:
         telegram_chat_user_service = TelegramChatUserService(db_session)
-        chat_members = telegram_chat_user_service.get_all_by_linked_wallet(
-            addresses=wallets
+        chat_members = telegram_chat_user_service.get_all_pairs(
+            chat_member_pairs=target_chat_members
         )
 
         if not chat_members:
             logger.info("No chats to validate. Skipping")
+            return
+        else:
+            logger.info(f"Found {len(chat_members)} chat members to validate")
 
         telethon_service = init_client()
         authorization_action = AuthorizationAction(
@@ -50,23 +52,9 @@ def check_chat_members() -> None:
         logger.warning("Community manager is disabled.")
         return
 
-    redis_service = RedisService()
-    wallets = redis_service.pop_from_set(
-        name=UPDATED_WALLETS_SET_NAME,
-        count=community_manager_settings.items_per_task,
-    )
-    if isinstance(wallets, str):
-        wallets = [wallets]
-
-    if wallets:
-        try:
-            asyncio.run(sanity_chat_checks(wallets=wallets))
-        except Exception as exc:
-            # Add wallets back to the set to retry later
-            logger.error(f"Failed to validate chat members: {exc}", exc_info=True)
-            redis_service.add_to_set(UPDATED_WALLETS_SET_NAME, *wallets)
-    else:
-        logger.info("No users to validate. Skipping")
+    with DBService().db_session() as db_session:
+        action = CommunityManagerChatAction(db_session=db_session)
+        asyncio.run(action.sanity_chat_checks())
 
 
 @app.task(

backend/community_manager/entrypoint.py

@@ -79,7 +79,7 @@ async def refresh_external_source(
                 raise
             return
         except TelegramChatInvalidExternalSourceError as e:
-            logger.exception(f"Invalid external source {source.url!r}: {e}")
+            logger.warning(f"Invalid external source {source.url!r}: {e}")
             if raise_for_error:
                 raise
             return