diff --git a/src/helpers.py b/src/helpers.py
index 252a0c7..e2c84f1 100644
--- a/src/helpers.py
+++ b/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
diff --git a/src/stan.py b/src/stan.py
index a194c4c..129756f 100644
--- a/src/stan.py
+++ b/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,11 +73,15 @@ 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
- logging.info(f"[JOINED] {info['full_name']}")
+ 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
info = short_user_data(from_user)
diff --git a/tests/test_bot.py b/tests/test_bot.py
index 32752d7..13e4e6a 100644
--- a/tests/test_bot.py
+++ b/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'], 'None, правило 6\nИмя должно быть читаемым и понятным.')
+ self.assertEqual(RESULTS[0][0]['text'],
+ 'None, правило 6\nИмя должно быть читаемым и понятным.')
self.assertEqual(RESULTS[0][1], 'sendMessage')
self.assertEqual(RESULTS[1][1], 'deleteMessage')
diff --git a/tests/test_helpers.py b/tests/test_helpers.py
index abb35ac..cf45605 100644
--- a/tests/test_helpers.py
+++ b/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()