From a31bc7f1c05f9ce6e956959679c4570762493eb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Apr 2025 00:07:36 +0200 Subject: [PATCH] feat: MORE telegram numbers categories --- backend/api/routes/admin/resource.py | 6 +- backend/core/actions/chat/__init__.py | 1 - backend/core/enums/nft.py | 13 ++++ backend/core/services/supertelethon.py | 1 + backend/core/utils/custom_rules/constants.py | 8 ++ backend/core/utils/custom_rules/mapping.py | 49 ++++++++++++ .../utils/custom_rules/telegram_numbers.py | 74 +++++++++++++++---- 7 files changed, 136 insertions(+), 16 deletions(-) create mode 100644 backend/core/utils/custom_rules/constants.py diff --git a/backend/api/routes/admin/resource.py b/backend/api/routes/admin/resource.py index 961fdce9..28ed2e59 100644 --- a/backend/api/routes/admin/resource.py +++ b/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() ] diff --git a/backend/core/actions/chat/__init__.py b/backend/core/actions/chat/__init__.py index 4c137bed..6369fc90 100644 --- a/backend/core/actions/chat/__init__.py +++ b/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) diff --git a/backend/core/enums/nft.py b/backend/core/enums/nft.py index 5e1bf402..9da66c5b 100644 --- a/backend/core/enums/nft.py +++ b/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): diff --git a/backend/core/services/supertelethon.py b/backend/core/services/supertelethon.py index 40482c23..dc6a1ef9 100644 --- a/backend/core/services/supertelethon.py +++ b/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: diff --git a/backend/core/utils/custom_rules/constants.py b/backend/core/utils/custom_rules/constants.py new file mode 100644 index 00000000..5c36a3de --- /dev/null +++ b/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]+$") diff --git a/backend/core/utils/custom_rules/mapping.py b/backend/core/utils/custom_rules/mapping.py index 5ecb6253..8b1550ba 100644 --- a/backend/core/utils/custom_rules/mapping.py +++ b/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 diff --git a/backend/core/utils/custom_rules/telegram_numbers.py b/backend/core/utils/custom_rules/telegram_numbers.py index 2fc674a7..1ff3beaa 100644 --- a/backend/core/utils/custom_rules/telegram_numbers.py +++ b/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