From 52347c19f780713efb784d6aae7f9e61d8522164 Mon Sep 17 00:00:00 2001 From: kotolex Date: Fri, 22 Dec 2023 22:17:07 +0500 Subject: [PATCH] Refactoring: step 2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - независимые функции вынесены в хелперы - добавлены тесты на хелперы и простые команды бота - проверка урла Правил вынесена в хелпер - уровень логирования повышен для бота - добавлено пара кейсов и слов в спам - при проверке на спам идет сработка если кроме русского и препинания есть еще какие-то символы - --- src/admin_commands.py | 14 ++--- src/commands.py | 5 +- src/config.py | 12 ++-- src/constants.py | 3 +- src/filters.py | 47 +++++----------- src/helpers.py | 63 +++++++++++++++------ src/report.py | 17 ++++++ src/stan.py | 10 ++-- tests/test_bot.py | 127 ++++++++++++++++++++++++++++++++++++++++++ tests/test_helpers.py | 73 ++++++++++++++++++++++++ wsgi.py | 7 +-- 11 files changed, 297 insertions(+), 81 deletions(-) create mode 100644 tests/test_bot.py create mode 100644 tests/test_helpers.py diff --git a/src/admin_commands.py b/src/admin_commands.py index 1d86d7b..13daa6c 100644 --- a/src/admin_commands.py +++ b/src/admin_commands.py @@ -4,8 +4,8 @@ from datetime import datetime as dt from telebot import types, logger from .constants import ADMIN_ID, LOGGING_LEVEL_DEBUG, LOGGING_LEVEL_INFO -from .filters import is_white, is_admin -from .helpers import me, my_ip +from .filters import is_white_id +from .helpers import me, my_ip, is_admin from .models import Chat, Quote, session from .reminder import remind, print_get_jobs, create_report_text from .report import reset_report_stats @@ -67,7 +67,7 @@ async def send_stats(message: types.Message): await bot.send_message(message.chat.id, reset_report_stats(message.chat.id)) -@bot.message_handler(func=is_white, commands=["enable_stan"], chat_types=["supergroup", "group"]) +@bot.message_handler(func=is_white_id, commands=["enable_stan"], chat_types=["supergroup", "group"]) async def enable_stan(message: types.Message): """Add group to database.""" logging.info(f"[{message.chat.title}] [{message.from_user.id}] {message.from_user.username}: {message.text}") @@ -96,7 +96,7 @@ async def disable_stan(message: types.Message): @bot.message_handler( - func=is_white, + func=is_white_id, commands=["set_antispam_report_reminder", "set_rules"], chat_types=["supergroup", "group"], ) @@ -117,7 +117,7 @@ async def set_antispam_report_reminder(message: types.Message): await bot.send_message(message.chat.id, f"""Настройки обновлены. Проверить: /get_group_info""") -@bot.message_handler(func=is_white, commands=["get_quotes"], chat_types=["supergroup", "group"]) +@bot.message_handler(func=is_white_id, commands=["get_quotes"], chat_types=["supergroup", "group"]) async def get_quotes(message: types.Message): logging.info(f"[{message.chat.id}] {message.from_user.first_name} {message.text}.") quotes = session.query(Quote.text).filter(Quote.chat_id == message.chat.id).all() @@ -130,7 +130,7 @@ async def get_quotes(message: types.Message): await bot.send_message(message.chat.id, f"Цитаты отсутствуют. Подробнее: /get_group_info") -@bot.message_handler(func=is_white, commands=["get_group_info"], chat_types=["supergroup", "group"]) +@bot.message_handler(func=is_white_id, commands=["get_group_info"], chat_types=["supergroup", "group"]) async def get_group_info(message: types.Message): group = session.query(Chat).filter(Chat.chat_id == message.chat.id).first() if group: @@ -151,7 +151,7 @@ ID группы: {group.chat_id} await bot.send_message(message.chat.id, f"Группа не включена. Включить: /enable_stan") -@bot.message_handler(func=is_white, commands=["set_logging_level"], chat_types=["supergroup", "group"]) +@bot.message_handler(func=is_white_id, commands=["set_logging_level"], chat_types=["supergroup", "group"]) async def set_logging_level(message: types.Message): args = message.text.split() if args[-1] == '10': diff --git a/src/commands.py b/src/commands.py index b192469..83ce911 100644 --- a/src/commands.py +++ b/src/commands.py @@ -5,11 +5,12 @@ import shelve from telebot import types from .constants import DATA, LOG_COMM, FAQ, LIB, RULES, RUS, RUS_ENG_TABLE, ENG_RUS_TABLE, PYTHONCHATRU, ZEN -from .filters import in_spam_list, in_caption_spam_list, in_delete_list, is_admin -from .helpers import represent_as_get, detect_args, update_stats +from .filters import in_spam_list, in_caption_spam_list, in_delete_list +from .helpers import represent_as_get, detect_args, is_admin from .admin_commands import bot from .rules import fetch_rule from .stan import act, speak +from .report import update_stats async def send_or_reply(message: types.Message, answer, **kwargs): diff --git a/src/config.py b/src/config.py index 0e34a9e..3fe0bea 100644 --- a/src/config.py +++ b/src/config.py @@ -1,8 +1,6 @@ import logging import os -from urllib.error import URLError -from urllib.request import urlopen - +from .helpers import is_url_reachable from dotenv import load_dotenv from telebot import logger @@ -17,10 +15,8 @@ ROLLBACK = {int(i) for i in os.environ.get("rollback").split(",")} USE_REMINDER = os.environ.get("use_reminder", "TRUE") == "TRUE" RULES_URL = os.environ.get("rules_url", "https://telegra.ph/pythonchatru-07-07") -try: - assert urlopen(RULES_URL).status == 200 -except (URLError, AssertionError): +if not is_url_reachable(RULES_URL): raise LookupError(f"LUTZPYBOT rules url({RULES_URL}) is not reachable") -logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") -logger.setLevel(logging.DEBUG) +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") +logger.setLevel(logging.INFO) diff --git a/src/constants.py b/src/constants.py index 271b4b3..7d1c9e5 100644 --- a/src/constants.py +++ b/src/constants.py @@ -55,7 +55,8 @@ SPAM = {"me.sv/", "tg.sv/", "goo.by/", "go.sv/", "intim.video/", "uclck.ru/", ' 'заработка в крипте', 'Веду набор', 'Идёт набор', 'мне в ЛС', 'Доход от', 'Trading', 'P2P', 'Идет набор', 'в лс', 'нaбop', 'комaндy', 'kpuпт', 'бupж', '💰', 'пишитe', '💸', '$ в неделю', 'обучаю с нуля', 'доходы для всех', 'по трейдингу', 'фальшивые рубли', '‼️', - 'зароботку', 'Набuраю', 'цuфр', 'Обучу вас', 'прuб', 'остались курсы', 'Crypt', 'kр', 'рu'} + 'зароботку', 'Набuраю', 'цuфр', 'Обучу вас', 'прuб', 'остались курсы', 'Crypt', 'kр', 'рu', + 'в лc'} NON_GRATA = {"дудар", "хауди", "dudar"} BAN_WORDS = {"GREEN"} diff --git a/src/filters.py b/src/filters.py index 6a803a9..c140498 100644 --- a/src/filters.py +++ b/src/filters.py @@ -3,9 +3,9 @@ import logging.handlers from telebot import types from .config import WHITEIDS -from .constants import BAN_WORDS, LOG_COMM, SPAM, ALLOWED_WORDS, URL_RX, NON_GRATA, ADMIN_ID -from .models import Chat -from .models import session +from .constants import LOG_COMM, ALLOWED_WORDS, URL_RX +from .helpers import is_spam, is_in_not_allowed, is_ban_words_in_caption +from .models import Chat, session def in_spam_list(message: types.Message) -> bool: @@ -13,57 +13,36 @@ def in_spam_list(message: types.Message) -> bool: antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name if from_user_id not in WHITEIDS and antispam_is_enabled: - for text in SPAM: - if text.casefold() in message.text.casefold(): - logging.info("!BAN!" + LOG_COMM % (title, from_user_id, from_user_name, message.text)) - return True + if is_spam(message.text): + logging.info("!BAN!" + LOG_COMM % (title, from_user_id, from_user_name, message.text)) + return True return False def in_caption_spam_list(message: types.Message) -> bool: """Check for mentioning unwanted words in caption.""" - for phrase in BAN_WORDS: - if message.caption and phrase in message.caption: - logging.info("!BAN!" + LOG_COMM % ( - message.chat.title, message.from_user.id, message.from_user.first_name, message.video.file_name)) - return True + if message.caption and is_ban_words_in_caption(message.caption): + from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name + logging.info("!BAN!" + LOG_COMM % (title, from_user_id, from_user_name, message.video.file_name)) + return True return False -def in_not_allowed(word_list, msg): - for word in word_list: - if word in msg.casefold(): - return False - return True - - def in_delete_list(message: types.Message) -> bool: """Check for URLs in message and delete.""" antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name if from_user_id not in WHITEIDS and antispam_is_enabled: - if URL_RX.search(message.text) and in_not_allowed(ALLOWED_WORDS, message.text): + if URL_RX.search(message.text) and is_in_not_allowed(ALLOWED_WORDS, message.text): logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: {message.text}") return True if message.entities: for entity in message.entities: - if entity.url and in_not_allowed(ALLOWED_WORDS, entity.url): + if entity.url and is_in_not_allowed(ALLOWED_WORDS, entity.url): logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: [entity] {message.text}") return True return False -def is_nongrata(type_message: types.Message) -> bool: - """Check for non grata persons.""" - for phrase in NON_GRATA: - if phrase in type_message.text.casefold(): - return True - return False - - -def is_admin(message: types.Message) -> bool: - return message.from_user.id == ADMIN_ID - - -def is_white(message: types.Message) -> bool: +def is_white_id(message: types.Message) -> bool: return message.from_user.id in WHITEIDS diff --git a/src/helpers.py b/src/helpers.py index 7a5127b..5e2739a 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -1,14 +1,54 @@ """ Send User's info. """ -import logging -import shelve -from pathlib import Path + from urllib import parse +from urllib.error import URLError from urllib.request import urlopen from telebot import types -from .constants import DATA -from .report import reset_report_stats +from .constants import SPAM, RUS, BAN_WORDS, NON_GRATA, ADMIN_ID + + +def is_url_reachable(url: str) -> bool: + try: + return urlopen(url).status == 200 + except URLError: + return False + + +def is_spam(message_text: str) -> bool: + if is_mixed(message_text): + return True + return any(text.casefold() in message_text.casefold() for text in SPAM) + + +def is_mixed(text: str) -> bool: + for word in text.split(): + if not all(e in RUS for e in word.strip().lower()): + return True + return False + + +def is_in_not_allowed(word_list: list, msg: str) -> bool: + for word in word_list: + if word in msg.casefold(): + return False + return True + + +def is_ban_words_in_caption(caption: str) -> bool: + return any(ban_word in caption for ban_word in BAN_WORDS) + + +def is_nongrata(type_message: types.Message) -> bool: + for phrase in NON_GRATA: + if phrase in type_message.text.casefold(): + return True + return False + + +def is_admin(message: types.Message) -> bool: + return message.from_user.id == ADMIN_ID def my_ip(): @@ -47,16 +87,3 @@ def detect_args(message: types.Message): else: if len(message.text.split()) > 1: return " ".join(message.text.split()[1:]) - - -def update_stats(message: types.Message): - if not Path(f"{DATA}{message.chat.id}").exists(): - reset_report_stats(message.chat.id) - with shelve.open(f"{DATA}{message.chat.id}", writeback=True) as shelve_db: - if "Messages" not in shelve_db: - reset_report_stats(message.chat.id) - if message.from_user.id not in shelve_db["Messages"]: - shelve_db["Messages"][message.from_user.id] = {"User": message.from_user, "Count": 1} - logging.info(f"[{message.chat.title[:10]}] [{message.from_user.id}] {message.from_user.first_name}") - else: - shelve_db["Messages"][message.from_user.id]["Count"] += 1 diff --git a/src/report.py b/src/report.py index df17683..66a2c59 100644 --- a/src/report.py +++ b/src/report.py @@ -1,7 +1,11 @@ import html +import logging import os import random import shelve +from pathlib import Path + +from telebot import types from .config import ROLLBACK from .constants import DATA @@ -63,3 +67,16 @@ def reset_report_stats(chat_id): {shelve_db['Messages']=} {shelve_db['Banned']=} {shelve_db['Deleted']=}""" + + +def update_stats(message: types.Message): + if not Path(f"{DATA}{message.chat.id}").exists(): + reset_report_stats(message.chat.id) + with shelve.open(f"{DATA}{message.chat.id}", writeback=True) as shelve_db: + if "Messages" not in shelve_db: + reset_report_stats(message.chat.id) + if message.from_user.id not in shelve_db["Messages"]: + shelve_db["Messages"][message.from_user.id] = {"User": message.from_user, "Count": 1} + logging.info(f"[{message.chat.title[:10]}] [{message.from_user.id}] {message.from_user.first_name}") + else: + shelve_db["Messages"][message.from_user.id]["Count"] += 1 diff --git a/src/stan.py b/src/stan.py index a184cac..1414337 100644 --- a/src/stan.py +++ b/src/stan.py @@ -6,9 +6,9 @@ import random from sqlalchemy import delete from telebot import types -from .filters import is_white, is_nongrata -from .models import Quote -from .models import session +from .filters import is_white_id +from .helpers import is_nongrata +from .models import Quote, session from .robot import bot TYPING_TIMEOUT = 0.13 / 4 # Reading time is quarter of the same text writing time @@ -35,7 +35,7 @@ async def act(message: types.Message): await send_quote(len(quote) * 0.13, message, quote) -@bot.message_handler(func=is_white, commands=["add"]) +@bot.message_handler(func=is_white_id, commands=["add"]) async def add_stan_quote(message: types.Message): if message.reply_to_message and message.reply_to_message.text: quote = message.reply_to_message.text @@ -48,7 +48,7 @@ async def add_stan_quote(message: types.Message): await bot.send_message(message.chat.id, f"⛔️ Не добавил, есть токое\n └ {quote}", parse_mode='Markdown') -@bot.message_handler(func=is_white, commands=["remove"]) +@bot.message_handler(func=is_white_id, commands=["remove"]) async def remove_stan_quote(message: types.Message): if message.reply_to_message and message.reply_to_message.text: quote = message.reply_to_message.text diff --git a/tests/test_bot.py b/tests/test_bot.py new file mode 100644 index 0000000..8d8183f --- /dev/null +++ b/tests/test_bot.py @@ -0,0 +1,127 @@ +import logging +from unittest import IsolatedAsyncioTestCase, main + +from telebot import asyncio_helper, types, logger + +from src import helpers + +# нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата +helpers.is_url_reachable = lambda a: True +from src.commands import bot + +# отключаем логи бота, в тестах они не нужны +logger.setLevel(logging.ERROR) + + +def get_update(text, reply_to=None): + params = {'text': text} + chat = types.Chat(id=11, type='private') + user = types.User(id=10, is_bot=False, first_name='Some User') + if reply_to: + params["reply_to_message"] = types.Message(message_id=1, from_user=user, date=None, chat=chat, + content_type='text', options={'text': reply_to}, json_string="") + mess = types.Message(message_id=1, from_user=user, date=None, chat=chat, content_type='text', options=params, + json_string="") + edited_message = None + channel_post = None + edited_channel_post = None + inline_query = None + chosen_inline_result = None + callback_query = None + shipping_query = None + pre_checkout_query = None + poll = None + poll_answer = None + my_chat_member = None + chat_member = None + chat_join_request = None + return types.Update(1001234038283, mess, edited_message, channel_post, edited_channel_post, inline_query, + chosen_inline_result, callback_query, shipping_query, pre_checkout_query, poll, poll_answer, + my_chat_member, chat_member, chat_join_request) + +# сюда будут складываться результаты, перед началом теста очищается +RESULTS = [] + + +async def custom_sender(token, url, method='get', params=None, files=None, **kwargs): + RESULTS.append(params) + result = {"message_id": 1, "date": 1, "chat": {"id": 1, "type": "private"}} + return result + +# Чтобы перехватить запросы к телеге +asyncio_helper._process_request = custom_sender + + +class TestBot(IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.bot = bot + RESULTS.clear() + + async def test_start(self): + await self.bot.process_new_updates([get_update('/start')]) + self.assertEqual(RESULTS[0]['text'], "Начни с прочтения") + + async def test_non_grata(self): + await self.bot.process_new_updates([get_update('как вам дударь?')]) + self.assertEqual(RESULTS[0]['text'], "у нас тут таких не любят") + + async def test_non_grata2(self): + await self.bot.process_new_updates([get_update('смотрю хауди')]) + self.assertEqual(RESULTS[0]['text'], "у нас тут таких не любят") + + async def test_rules(self): + await self.bot.process_new_updates([get_update('/rules')]) + self.assertEqual(RESULTS[0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0]['reply_markup']) + + async def test_faq(self): + await self.bot.process_new_updates([get_update('/faq')]) + self.assertEqual(RESULTS[0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0]['reply_markup']) + + async def test_lib(self): + await self.bot.process_new_updates([get_update('/lib')]) + self.assertEqual(RESULTS[0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0]['reply_markup']) + + async def test_tr(self): + await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) + self.assertEqual(RESULTS[0]['text'], "привет") + + async def test_tr2(self): + await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) + self.assertEqual(RESULTS[0]['text'], "hello") + + async def test_tsya(self): + await self.bot.process_new_updates([get_update('/tsya')]) + self.assertEqual(RESULTS[0]['text'], "-тся и -ться в глаголах") + self.assertTrue("https://tsya.ru/" in RESULTS[0]['reply_markup']) + + async def test_nometa(self): + text = """Не задавай мета-вопросов вроде:\n «Можно задать вопрос?»\n «Кто-нибудь пользовался .. ?»\n «Привет, мне нужна помощь по .. !»\n\nПросто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь.""" + await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) + self.assertEqual(RESULTS[0]['text'], text) + + async def test_neprivet(self): + await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")]) + self.assertEqual(RESULTS[0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") + self.assertTrue("https://neprivet.com/" in RESULTS[0]['reply_markup']) + + async def test_nojob(self): + text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в ТВОЁМ коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️""" + await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) + self.assertEqual(RESULTS[0]['text'], text) + + async def test_nobot(self): + text = """Внимание:\nТелеграм бот не должен быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, основы веб-технологий, асинхронное программирование и отладку до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете.""" + await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) + self.assertEqual(RESULTS[0]['text'], text) + + async def test_nogui(self): + text = """Внимание:\nGUI приложение не должно быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, циклы событий и отладку до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете.""" + await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")]) + self.assertEqual(RESULTS[0]['text'], text) + + +if __name__ == '__main__': + main() diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..60da9e9 --- /dev/null +++ b/tests/test_helpers.py @@ -0,0 +1,73 @@ +from unittest import TestCase, main +from unittest.mock import MagicMock +from src.constants import SPAM, BAN_WORDS, ADMIN_ID +from src.helpers import is_spam, is_mixed, is_ban_words_in_caption, is_in_not_allowed, is_nongrata, is_admin + +CASES = ( + "Оформим резидeнтствo ОАЭ без предoплаты за 5000 дирхам. Пoдробнoсти в лc", + "Всем привет Я в поисках людей в kpиптo кoмaнду У нас вы пройдете 6ecплaтноe обученue 1-3 дня Можно совмещать с основной paбoтoй" +) + + +class TestSpam(TestCase): + def test_simple(self): + for spam_word in SPAM: + with self.subTest(f"check spam word({spam_word})"): + self.assertTrue(is_spam(spam_word)) + + def test_case_first(self): + for text in CASES: + with self.subTest(f"check case({text})"): + self.assertTrue(is_spam(text)) + + def test_one(self): + self.assertTrue(is_spam("привeт")) # e английская + + def test_is_mixed(self): + self.assertTrue(is_mixed("привeт")) # e английская + self.assertFalse(is_mixed("привет ребята!")) + + +class TestOthers(TestCase): + def test_ban(self): + for word in BAN_WORDS: + with self.subTest(f"check ban word({word})"): + self.assertTrue(is_ban_words_in_caption(word)) + + def test_allowed(self): + params = ( + (True, [], 'any'), + (True, ['one'], 'two'), + (False, ['one', 'two'], 'two'), + (False, ['one', 'two'], 'Two'), + (False, ['one', 'two'], 'TWO'), + ) + for expected, words, mess in params: + with self.subTest(f"check allowed({words, mess})"): + self.assertEqual(is_in_not_allowed(words, mess), expected) + + def test_non_grata(self): + params = ( + (False, 'any'), + (True, 'дудар'), + (True, 'хауди'), + (True, 'dudar'), + ) + for expected, text in params: + with self.subTest(f"check non-grata({text})"): + m = MagicMock(text=text) + self.assertEqual(is_nongrata(m), expected) + + def test_is_admin(self): + params = ( + (False, 'any'), + (True, ADMIN_ID), + ) + for expected, text in params: + with self.subTest(f"check non-grata({text})"): + m = MagicMock(from_user=MagicMock(id=text)) + self.assertEqual(is_admin(m), expected) + + +if __name__ == '__main__': + main() diff --git a/wsgi.py b/wsgi.py index c8a4b7f..8fc75a2 100644 --- a/wsgi.py +++ b/wsgi.py @@ -1,6 +1 @@ -import asyncio - -from src.webhook import app, bot - -if __name__ == '__main__': - asyncio.run(bot.infinity_polling()) +from src.webhook import app