Refactoring: step 2

- независимые функции вынесены в хелперы
- добавлены тесты на хелперы и простые команды бота
- проверка урла Правил вынесена в хелпер
- уровень логирования повышен для бота
- добавлено пара кейсов и слов в спам
- при проверке на спам идет сработка если кроме русского и препинания есть еще какие-то символы
-

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':

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):

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)

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"}

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

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

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

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

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'], "<i>-тся</i> и <i>-ться</i> в глаголах")
+        self.assertTrue("https://tsya.ru/" in RESULTS[0]['reply_markup'])
+
+    async def test_nometa(self):
+        text = """Не задавай мета-вопросов вроде:\n<i>  «Можно задать вопрос?»\n  «Кто-нибудь пользовался .. ?»\n  «Привет, мне нужна помощь по .. !»</i>\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Здесь помогают по конкретным вопросам в <u>ТВОЁМ</u> коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️"""
+        await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")])
+        self.assertEqual(RESULTS[0]['text'], text)
+
+    async def test_nobot(self):
+        text = """<b>Внимание</b>:\nТелеграм бот <i>не должен</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>основы веб-технологий</code>, <code>асинхронное программирование</code> и <code>отладку</code> до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете."""
+        await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")])
+        self.assertEqual(RESULTS[0]['text'], text)
+
+    async def test_nogui(self):
+        text = """<b>Внимание</b>:\nGUI приложение <i>не должно</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>циклы событий</code> и <code>отладку</code> до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете."""
+        await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")])
+        self.assertEqual(RESULTS[0]['text'], text)
+
+
+if __name__ == '__main__':
+    main()

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()

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