Merge pull request #12 from kotolex/master

FEATURE: ban for url in bio

src/helpers.py

@@ -126,3 +126,7 @@ def fetch_rule(index: int) -> str:
         return RULES_TEXT[index - 1]
     else:
         return "Пока не придумали"
+
+
+def has_links(body: str) -> bool:
+    return "https://" in body or "http://" in body

src/stan.py

@@ -4,10 +4,10 @@ import html
 import logging
 import random
 
-from telebot import types, util
+from telebot import types
 
 from .filters import is_white_id
-from .helpers import is_nongrata, short_user_data
+from .helpers import is_nongrata, short_user_data, has_links
 from .models import all_chat_quotes, add_quote, is_quote_in_chat, delete_quote_in_chat
 from .robot import bot
 
@@ -73,10 +73,14 @@ async def check_new_members(message: types.Message):
         info = short_user_data(from_user)
         await bot.delete_message(message.chat.id, message.id)
         # мы не знаем как быстро получим инфу о био, потому сначала удаляем сообщение, потом запрашиваем био/логируем
-        more_data = await bot.get_chat(from_user.id)
+        more_data = await bot.get_chat(info['id'])
         info['bio'] = more_data.bio
         if more_data.photo:
             info['photo'] = more_data.photo.big_file_id
+        if has_links(info['bio']):
+            await bot.ban_chat_member(message.chat.id, info['id'])
+            logging.info(f"[BANNED] User {info['full_name']} has url in bio: {info['bio']}")
+        else:
             logging.info(f"[JOINED] {info['full_name']}")
     if message.content_type == "left_chat_member":
         from_user = message.from_user

tests/test_bot.py

@@ -25,6 +25,25 @@ logger.setLevel(logging.ERROR)
 
 # сюда будут складываться результаты, перед началом теста очищается
 RESULTS = []
+USER_ID = 555
+SPAMMER_ID = 666
+
+
+async def custom_sender(_token, url, _method='get', params=None, _files=None, **_kwargs):
+    """Замена для тестов, чтобы не слать запросы в телегу"""
+    RESULTS.append((params, url))
+    bio = "bio" if params["chat_id"] == USER_ID else "My bot here https://t.me/stupid.bot"
+    chat = {"id": 1000, "type": "group", "bio": bio,
+            "photo": {"big_file_id": "file_id", "small_file_id": "file_id", "small_file_unique_id": "file_id",
+                      "big_file_unique_id": "file_id"}}
+    result = {"message_id": 1000, "date": 1, "chat": chat}
+    if url == "getChat":
+        result = chat
+    return result
+
+
+# Чтобы перехватить запросы к телеге
+asyncio_helper._process_request = custom_sender
 
 
 def _all_nones():
@@ -58,33 +77,18 @@ def get_update(text, reply_to=None, user_id=10, first_name='Some User'):
     return types.Update(1001234038283, mess, *_all_nones())
 
 
-def member(leave: bool = False):
+def member(leave: bool = False, url_in_bio: bool = False):
     """Хелпер, генерирующий событие входа и выхода юзера"""
     type_ = 'left_chat_member' if leave else 'new_chat_members'
     chat = types.Chat(id=11, type='group')
     params = {'content_type': type_}
-    user = types.User(id=555, is_bot=False, first_name='Some User')
+    id_ = SPAMMER_ID if url_in_bio else USER_ID
+    user = types.User(id=id_, is_bot=False, first_name='Some User')
     mess = types.Message(message_id=1, from_user=user, date=None, chat=chat, content_type=type_,
                          options=params, json_string='')
     return types.Update(100100, mess, *_all_nones())
 
 
-async def custom_sender(token, url, method='get', params=None, files=None, **kwargs):
-    """Замена для тестов, чтобы не слать запросы в телегу"""
-    RESULTS.append((params, url))
-    chat = {"id": 1000, "type": "group", "bio": "bio",
-            "photo": {"big_file_id": "file_id", "small_file_id": "file_id", "small_file_unique_id": "file_id",
-                      "big_file_unique_id": "file_id"}}
-    result = {"message_id": 1000, "date": 1, "chat": chat}
-    if url == "getChat":
-        result = chat
-    return result
-
-
-# Чтобы перехватить запросы к телеге
-asyncio_helper._process_request = custom_sender
-
-
 class TestOther(TestCase):
     def test_is_white(self):
         params = (
@@ -345,7 +349,13 @@ class TestBot(IsolatedAsyncioTestCase):
     async def test_new_user_in_chat(self):
         await self.bot.process_new_updates([member()])
         self.assertEqual(RESULTS[0][1], 'deleteMessage')
-        self.assertEqual(RESULTS[1], ({'chat_id': 555}, 'getChat'))
+        self.assertEqual(RESULTS[1], ({'chat_id': USER_ID}, 'getChat'))
+
+    async def test_new_user_in_chat_ban_for_bio(self):
+        await self.bot.process_new_updates([member(url_in_bio=True)])
+        self.assertEqual(RESULTS[0][1], 'deleteMessage')
+        self.assertEqual(RESULTS[1], ({'chat_id': SPAMMER_ID}, 'getChat'))
+        self.assertEqual(RESULTS[2][1], 'banChatMember')
 
     async def test_user_leave_chat(self):
         await self.bot.process_new_updates([member(leave=True)])
@@ -496,7 +506,8 @@ class TestBot(IsolatedAsyncioTestCase):
     async def test_invalid_name_user(self):
         await self.bot.process_new_updates([get_update('новый текст', first_name=None)])
         self.assertEqual(RESULTS[0][0]['chat_id'], '11')
-        self.assertEqual(RESULTS[0][0]['text'], '<b><a href="tg:user?id=10">None</a>, правило 6</b>\n<i>Имя должно быть читаемым и понятным.</i>')
+        self.assertEqual(RESULTS[0][0]['text'],
+                         '<b><a href="tg:user?id=10">None</a>, правило 6</b>\n<i>Имя должно быть читаемым и понятным.</i>')
         self.assertEqual(RESULTS[0][1], 'sendMessage')
         self.assertEqual(RESULTS[1][1], 'deleteMessage')
 

tests/test_helpers.py

@@ -1,8 +1,10 @@
 from unittest import TestCase, main
 from unittest.mock import MagicMock
 from src.constants import SPAM, BAN_WORDS, ADMIN_ID, RULES_TEXT
-from src.helpers import is_spam, is_mixed, is_ban_words_in_caption, is_in_not_allowed, is_nongrata, is_admin, \
-    fetch_rule, cleaned_text, remove_spaces, has_no_letters
+from src.helpers import (
+    is_spam, is_mixed, is_ban_words_in_caption, is_in_not_allowed, is_nongrata, is_admin, fetch_rule, cleaned_text,
+    remove_spaces, has_no_letters, has_links
+)
 
 REAL_CASES = (
     "Оформим резидeнтствo ОАЭ без предoплаты за 5000 дирхам. Пoдробнoсти в лc",
@@ -180,6 +182,17 @@ class TestOthers(TestCase):
             with self.subTest(f"has_no_letters {text}"):
                 self.assertEqual(expected, has_no_letters(text))
 
+    def test_has_links(self):
+        params = (
+            (True, 'My bot https://t.me/bot.some'),
+            (True, 'My bot http://t.me/bot.some'),
+            (False, 'My bot here'),
+            (False, 'I can use http'),
+        )
+        for expected, text in params:
+            with self.subTest(f"has_links {text}"):
+                self.assertEqual(expected, has_links(text))
+
 
 if __name__ == '__main__':
     main()