FEATURE: bam for url in bio
- добавил бан за наличие ссылки в био - добавил тест на такой бан
This commit is contained in:
@@ -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
|
||||
|
||||
+7
-3
@@ -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
|
||||
|
||||
+31
-20
@@ -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')
|
||||
|
||||
|
||||
+15
-2
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user