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