Merge pull request #39 from OpenBuilders/feat/telegram-number-categories

feat: MORE telegram numbers categories

Author: danoctua

backend/api/routes/admin/resource.py

@@ -66,8 +66,10 @@ async def get_nft_collection_categories() -> list[CategoriesFDO]:
     return [
         CategoriesFDO(
             asset=asset.value,
-            categories=list(category_definition.enum)
-            + ([None] if category_definition.is_optional else []),
+            categories=(
+                ([None] if category_definition.is_optional else [])
+                + list(category_definition.enum)
+            ),
         )
         for asset, category_definition in ASSET_TO_CATEGORY_TYPE_MAPPING.items()
     ]

backend/core/actions/chat/__init__.py

@@ -57,7 +57,6 @@ class TelegramChatAction(BaseAction):
     def __init__(
         self, db_session: Session, telethon_client: TelegramClient | None = None
     ):
-        logger.info(f"Initializing TelegramChatAction {telethon_client=}")
         super().__init__(db_session)
         self.telegram_chat_service = TelegramChatService(db_session)
         self.telegram_chat_user_service = TelegramChatUserService(db_session)

backend/core/enums/nft.py

@@ -45,6 +45,19 @@ class TelegramUsernameCategory(enum.StrEnum):
 class TelegramNumberCategory(enum.StrEnum):
     DIGITS_7 = "7 digits"
     DIGITS_11 = "11 digits"
+    CLUB_007 = "007 Club"
+    CLUB_69 = "69 Club"
+    CLUB_420 = "420 Club"
+    CLUB_666 = "666 Club"
+    CLUB_777 = "777 Club"
+    CLUB_888 = "888 Club"
+    CLUB_1337 = "1337 Club"
+    CLUB_BINARY = "Binary Club"
+    REPEAT_2 = "XX"
+    REPEAT_3 = "XXX"
+    REPEAT_4 = "XXXX"
+    REPEAT_5 = "XXXXX"
+    YEAR = "Year"
 
 
 class TonDnsCategory(enum.StrEnum):

backend/core/services/supertelethon.py

@@ -48,6 +48,7 @@ def __init__(self, client: TelegramClient | None = None) -> None:
     async def start(self) -> None:
         if self.client.is_connected():
             return
+        logger.info("Initiating Telethon connection.")
         await self.client.start(bot_token=core_settings.telegram_bot_token)
 
     def start_sync(self) -> None:

backend/core/utils/custom_rules/constants.py

@@ -0,0 +1,8 @@
+import re
+
+DIGIT_REPEATS_AT_LEAST_TWICE = re.compile(r"(\d)\1+")
+DIGIT_REPEATS_AT_LEAST_THRICE = re.compile(r"(\d)\1{2,}")
+DIGIT_REPEATS_AT_LEAST_FOURTH = re.compile(r"(\d)\1{3,}")
+DIGIT_REPEATS_AT_LEAST_FIFTH = re.compile(r"(\d)\1{4,}")
+DIGIT_IS_YEAR = re.compile(r"(19[0-9]{2}|20(0[0-9]|([12])[0-9]))")
+DIGIT_IS_BINARY = re.compile(r"^[01]+$")

backend/core/utils/custom_rules/mapping.py

@@ -1,8 +1,18 @@
 from collections.abc import Callable
 
+from core.utils.custom_rules.constants import (
+    DIGIT_REPEATS_AT_LEAST_TWICE,
+    DIGIT_REPEATS_AT_LEAST_THRICE,
+    DIGIT_REPEATS_AT_LEAST_FOURTH,
+    DIGIT_REPEATS_AT_LEAST_FIFTH,
+    DIGIT_IS_YEAR,
+    DIGIT_IS_BINARY,
+)
 from core.utils.custom_rules.telegram_gifts import handle_telegram_gifts_type_category
 from core.utils.custom_rules.telegram_numbers import (
     handle_telegram_numbers_length_category,
+    handle_telegram_numbers_substring_category,
+    handle_telegram_numbers_regex_match_category,
 )
 from core.enums.nft import (
     TelegramNumberCategory,
@@ -27,6 +37,45 @@
     TelegramNumberCategory.DIGITS_11: handle_telegram_numbers_length_category(
         target_length=8
     ),
+    TelegramNumberCategory.CLUB_007: handle_telegram_numbers_substring_category(
+        substring="007"
+    ),
+    TelegramNumberCategory.CLUB_69: handle_telegram_numbers_substring_category(
+        substring="69"
+    ),
+    TelegramNumberCategory.CLUB_420: handle_telegram_numbers_substring_category(
+        substring="420"
+    ),
+    TelegramNumberCategory.CLUB_666: handle_telegram_numbers_substring_category(
+        substring="666"
+    ),
+    TelegramNumberCategory.CLUB_777: handle_telegram_numbers_substring_category(
+        substring="777"
+    ),
+    TelegramNumberCategory.CLUB_888: handle_telegram_numbers_substring_category(
+        substring="888"
+    ),
+    TelegramNumberCategory.CLUB_1337: handle_telegram_numbers_substring_category(
+        substring="1337"
+    ),
+    TelegramNumberCategory.REPEAT_2: handle_telegram_numbers_regex_match_category(
+        regex_pattern=DIGIT_REPEATS_AT_LEAST_TWICE,
+    ),
+    TelegramNumberCategory.REPEAT_3: handle_telegram_numbers_regex_match_category(
+        regex_pattern=DIGIT_REPEATS_AT_LEAST_THRICE,
+    ),
+    TelegramNumberCategory.REPEAT_4: handle_telegram_numbers_regex_match_category(
+        regex_pattern=DIGIT_REPEATS_AT_LEAST_FOURTH,
+    ),
+    TelegramNumberCategory.REPEAT_5: handle_telegram_numbers_regex_match_category(
+        regex_pattern=DIGIT_REPEATS_AT_LEAST_FIFTH,
+    ),
+    TelegramNumberCategory.YEAR: handle_telegram_numbers_regex_match_category(
+        regex_pattern=DIGIT_IS_YEAR,
+    ),
+    TelegramNumberCategory.CLUB_BINARY: handle_telegram_numbers_regex_match_category(
+        regex_pattern=DIGIT_IS_BINARY,
+    ),
     # Telegram Usernames
     TelegramUsernameCategory.LETTERS_4: handle_telegram_username_length_category(
         target_length=4

backend/core/utils/custom_rules/telegram_numbers.py

@@ -1,3 +1,4 @@
+import re
 from collections.abc import Callable
 
 from core.enums.nft import NftCollectionAsset
@@ -16,6 +17,34 @@ def __len__(self):
         return len(self.digits)
 
 
+def _is_telegram_number(nft_item: NftItem) -> bool:
+    if (
+        nft_item.collection_address
+        != NFT_ASSET_TO_ADDRESS_MAPPING[NftCollectionAsset.TELEGRAM_NUMBER]
+    ):
+        return False
+
+    if not nft_item.blockchain_metadata.name:
+        return False
+
+    return True
+
+
+def _is_valid_length(item: NftItem, target_length: int) -> bool:
+    telegram_number = TelegramNumber(item.blockchain_metadata.name)
+    return len(telegram_number) <= target_length
+
+
+def _is_substring_in_number(item: NftItem, substring: str) -> bool:
+    telegram_number = TelegramNumber(item.blockchain_metadata.name)
+    return substring in telegram_number.digits
+
+
+def _is_regex_matched(item: NftItem, regex_pattern: re.Pattern) -> bool:
+    telegram_number = TelegramNumber(item.blockchain_metadata.name)
+    return bool(regex_pattern.match(telegram_number.digits))
+
+
 def handle_telegram_numbers_length_category(
     target_length: int,
 ) -> Callable[[list[NftItem]], list[NftItem]]:
@@ -31,23 +60,42 @@ def handle_telegram_numbers_length_category(
     """
 
     def _inner(nfts: list[NftItem]) -> list[NftItem]:
-        valid_nfts = []
+        return list(
+            filter(
+                lambda item: _is_telegram_number(item)
+                and _is_valid_length(item, target_length=target_length),
+                nfts,
+            )
+        )
+
+    return _inner
 
-        for nft in nfts:
-            if (
-                nft.collection_address
-                != NFT_ASSET_TO_ADDRESS_MAPPING[NftCollectionAsset.TELEGRAM_NUMBER]
-            ):
-                continue
 
-            if not nft.blockchain_metadata.name:
-                continue
+def handle_telegram_numbers_substring_category(
+    substring: str,
+) -> Callable[[list[NftItem]], list[NftItem]]:
+    def _inner(nfts: list[NftItem]) -> list[NftItem]:
+        return list(
+            filter(
+                lambda item: _is_telegram_number(item)
+                and _is_substring_in_number(item, substring=substring),
+                nfts,
+            )
+        )
 
-            telegram_number = TelegramNumber(nft.blockchain_metadata.name)
+    return _inner
 
-            if len(telegram_number) <= target_length:
-                valid_nfts.append(nft)
 
-        return valid_nfts
+def handle_telegram_numbers_regex_match_category(
+    regex_pattern: re.Pattern,
+) -> Callable[[list[NftItem]], list[NftItem]]:
+    def _inner(nfts: list[NftItem]) -> list[NftItem]:
+        return list(
+            filter(
+                lambda item: _is_telegram_number(item)
+                and _is_regex_matched(item, regex_pattern=regex_pattern),
+                nfts,
+            )
+        )
 
     return _inner