diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml new file mode 100644 index 0000000..4e36e6d --- /dev/null +++ b/.github/workflows/python-package.yml @@ -0,0 +1,32 @@ +# This workflow will install Python dependencies, run tests and lint with a single version of Python +# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python + +name: Python application + +on: + push: + branches: [ "master" ] + pull_request: + branches: [ "master" ] + +permissions: + contents: read + +jobs: + build: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v3 + - name: Set up Python 3.10 + uses: actions/setup-python@v3 + with: + python-version: "3.10" + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + - name: Test with unittest + run: | + python -m unittest discover -s tests diff --git a/app/__init__.py b/app/__init__.py deleted file mode 100644 index 2b3c21e..0000000 --- a/app/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from .commands import service diff --git a/app/commands/get.py b/app/commands/get.py deleted file mode 100644 index 415fa1c..0000000 --- a/app/commands/get.py +++ /dev/null @@ -1,27 +0,0 @@ -from urllib.request import urlopen - - -def my_ip(): - ip_service = "https://icanhazip.com" - ip = urlopen( - ip_service, - ) - return ip.read().decode("utf-8") - - -def me(message): - msg = f"Telegram ID: {message.from_user.id}\n\ - ├ Is bot: {message.from_user.is_bot}\n\ - ├ First name: {message.from_user.first_name}\n" - if message.from_user.last_name: - msg += f" ├ Last name: {message.from_user.last_name}\n" - if message.from_user.username: - msg += f" ├ Username: {message.from_user.username}\n" - msg += f" ├ Language code: {message.from_user.language_code}\n" - if message.from_user.is_premium is True: - msg += f" ├ Is Premium: ⭐️\n" - else: - msg += f" ├ Is Premium: No\n" - msg += f" ├ Chat ID: {message.chat.id}" - - return msg diff --git a/app/filters.py b/app/filters.py deleted file mode 100644 index 432d0e1..0000000 --- a/app/filters.py +++ /dev/null @@ -1,75 +0,0 @@ -import logging.handlers - -from .models import session -from .models import Chat -from .config import * - - -def in_spam_list(message: types.Message) -> bool: - """Check for mentioning unwanted persons in text.""" - antispam_is_enabled = session.query(Chat.antispam).filter_by(chat_id=message.chat.id).first()[0] - if message.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 % ( - message.chat.title, message.from_user.id, message.from_user.first_name, message.text)) - return True - - -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 - - -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] - if message.from_user.id not in WHITEIDS and antispam_is_enabled: - if URL_RX.search(message.text) and in_not_allowed( - ALLOWED_WORDS, message.text - ): - logging.info( - f"[DELETE] [{message.chat.title}] [{message.from_user.id}] {message.from_user.first_name}: {message.text}" - ) - return True - if message.entities: - for entity in message.entities: - if entity.url and in_not_allowed(ALLOWED_WORDS, entity.url): - logging.info( - f"[DELETE] [{message.chat.title}] [{message.from_user.id}] {message.from_user.first_name}: [entity] {message.text}" - ) - return True - - -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 - - -def is_admin(message: types.Message) -> bool: - return message.from_user.id == ADMIN_ID - - -def is_white(message: types.Message) -> bool: - return message.from_user.id in WHITEIDS - - -async def send_or_reply(m: types.Message, answer, **kwargs): - if m.reply_to_message: - await bot.reply_to(m.reply_to_message, answer, **kwargs) - else: - await bot.send_message(m.chat.id, answer, **kwargs) diff --git a/app/helpers.py b/app/helpers.py deleted file mode 100644 index 2204944..0000000 --- a/app/helpers.py +++ /dev/null @@ -1,39 +0,0 @@ -""" Send User's info. """ -import shelve -import logging -from . import report -from .config import DATA -from urllib import parse -from telebot import types - - -def represent_as_get(message: types.Message): - return parse.quote_plus(detect_args(message)) - - -def detect_args(message: types.Message): - if message.reply_to_message and message.reply_to_message.text: - if len(message.text.split()) == 1: - return message.reply_to_message.text - else: - return " ".join(message.text.split()[1:]) - else: - if len(message.text.split()) > 1: - return " ".join(message.text.split()[1:]) - - -def update_stats(message: types.Message): - with shelve.open(f"{DATA}{message.chat.id}", writeback=True) as s: - if "Messages" not in s: - report.reset_report_stats(message.chat.id) - - if message.from_user.id not in s["Messages"]: - s["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: - s["Messages"][message.from_user.id]["Count"] += 1 diff --git a/app/models.py b/app/models.py deleted file mode 100644 index 51c6c96..0000000 --- a/app/models.py +++ /dev/null @@ -1,49 +0,0 @@ -import os -from sqlalchemy import Column, String, Integer, Boolean, ForeignKey -from sqlalchemy.orm import relationship, declarative_base, sessionmaker - -from sqlalchemy import create_engine - -Base = declarative_base() -basedir = os.path.abspath(os.path.dirname(__file__)) - -engine = create_engine('sqlite:///' + os.path.join(basedir, 'db/db.sqlite')) -Base.metadata.create_all(engine) - -Session = sessionmaker(bind=engine) -session = Session() - - -class Permission: - USE_COMMANDS = 1 # Пользоваться обычными командами. - SEND_LINKS = 2 # Может отправлять ссылки без удаления. - MODIFY_QUOTES = 4 # Может добавлять и удалять цитаты. - MODERATE = 8 - ADMIN = 16 # Может банить юзеров и удалять сообщения. - - -class Quote(Base): - __tablename__ = "quotes" - id = Column(Integer, primary_key=True) - chat_id = Column(Integer, ForeignKey('chats.chat_id')) - text = Column(String) - - def __repr__(self): - return f"Quote({self.text=}, {self.chat_id=})" - - -class Chat(Base): - __tablename__ = "chats" - id = Column(Integer, primary_key=True) - chat_id = Column(Integer, unique=True) - title = Column(String) - - # Settings - quotes = relationship('Quote', backref='chat', cascade='all, delete') - antispam = Column(Boolean) - report = Column(Boolean) - reminder = Column(Boolean) - hidden = Column(Boolean) - - def __repr__(self): - return f'Chat({self.chat_id=}, {self.title=})' diff --git a/requirements.txt b/requirements.txt index 951a96f..de8b4d4 100644 Binary files a/requirements.txt and b/requirements.txt differ diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ + diff --git a/app/commands/service.py b/src/admin_commands.py similarity index 70% rename from app/commands/service.py rename to src/admin_commands.py index 97f7231..13daa6c 100644 --- a/app/commands/service.py +++ b/src/admin_commands.py @@ -1,18 +1,19 @@ -""" Service commands for Admin purposes. """ - import logging -from telebot import logger from datetime import datetime as dt -from ..models import Chat, Quote, session -from ..config import bot, ADMIN_ID, types -from .. import reminder -from .. import report -from ..filters import is_admin, is_white -from .get import me, my_ip + +from telebot import types, logger + +from .constants import ADMIN_ID, LOGGING_LEVEL_DEBUG, LOGGING_LEVEL_INFO +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 +from .robot import bot @bot.message_handler(commands=["me"]) -async def command_me(message): +async def send_me(message): """Send info about user and chat id [Service].""" await bot.send_message(message.chat.id, me(message)) @@ -32,7 +33,7 @@ async def remind_manually(message): except ValueError as ve: await bot.send_message(message.chat.id, f"Не удалось разобрать дату!\n{ve}") else: - await reminder.remind(message.chat.id, today) + await remind(message.chat.id, today) else: await bot.send_message( message.chat.id, @@ -44,45 +45,34 @@ async def remind_manually(message): @bot.message_handler(func=is_admin, commands=["jobs"]) -async def list_jobs(message): +async def list_jobs(_message): """List all the jobs in schedule.""" - await bot.send_message(ADMIN_ID, reminder.print_get_jobs()) + await bot.send_message(ADMIN_ID, print_get_jobs()) @bot.message_handler(func=is_admin, commands=["stats"]) async def send_stats(message: types.Message): if len(message.text.split()) == 1: - rep = report.create_report_text(message.chat.id) + rep = create_report_text(message.chat.id) if rep: await bot.send_message(message.chat.id, rep) else: - await bot.send_message(message.chat.id, report.create_report_text(message.text.split()[-1])) + await bot.send_message(message.chat.id, create_report_text(message.text.split()[-1])) @bot.message_handler(func=is_admin, commands=["reset_stats"], chat_types=["supergroup", "group"]) async def send_stats(message: types.Message): logging.warning("reset_stats") - report.reset_report_stats(message.chat.id) - await bot.send_message(message.chat.id, report.reset_report_stats(message.chat.id)) + reset_report_stats(message.chat.id) + 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}" - ) + logging.info(f"[{message.chat.title}] [{message.from_user.id}] {message.from_user.username}: {message.text}") if not session.query(Chat.id).filter(Chat.chat_id == message.chat.id).first(): - session.add( - Chat( - chat_id=message.chat.id, - title=message.chat.title, - antispam=1, - report=0, - reminder=1, - ) - ) + session.add(Chat(chat_id=message.chat.id, title=message.chat.title, antispam=1, report=0, reminder=1)) session.commit() await bot.send_message( message.chat.id, @@ -100,16 +90,13 @@ async def disable_stan(message: types.Message): chat = session.query(Chat).filter_by(chat_id=message.chat.id).first() session.delete(chat) session.commit() - await bot.send_message( - message.chat.id, - f"Группа {message.chat.title} и все связанные с ней цитаты удалёны!", - ) + await bot.send_message(message.chat.id, f"Группа {message.chat.title} и все связанные с ней цитаты удалёны!") else: await bot.send_message(message.chat.id, "Отказ. Этой группы нет в БД.") @bot.message_handler( - func=is_white, + func=is_white_id, commands=["set_antispam_report_reminder", "set_rules"], chat_types=["supergroup", "group"], ) @@ -130,24 +117,20 @@ 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() if quotes: - text = f"Количество цитат: {len(session.query(Quote).filter(Quote.chat_id == message.chat.id).all())}\n\ -Последние добавленные\n\n" + length = len(session.query(Quote).filter(Quote.chat_id == message.chat.id).all()) + text = f"Количество цитат: {length}\n\Последние добавленные\n\n" text += "\n".join(f"· {quote[0]}" for quote in quotes[-10:]) await bot.send_message(message.chat.id, f"{text}") else: - await bot.send_message( - message.chat.id, f"Цитаты отсутствуют. Подробнее: /get_group_info" - ) + 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: @@ -168,12 +151,12 @@ 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': - logger.setLevel(10) # DEBUG + logger.setLevel(LOGGING_LEVEL_DEBUG) logging.info("[LEVEL] Установлен уровень DEBUG") else: - logger.setLevel(20) # INFO + logger.setLevel(LOGGING_LEVEL_INFO) logging.info("[LEVEL] Установлен уровень INFO") diff --git a/app/bot.py b/src/commands.py similarity index 86% rename from app/bot.py rename to src/commands.py index 4572899..bdcc949 100644 --- a/app/bot.py +++ b/src/commands.py @@ -1,11 +1,24 @@ -import shelve +import logging +import logging.handlers -from . import stan -from . import rules +from telebot import types + +from .admin_commands import bot +from .constants import (LOG_COMM, FAQ, LIB, RULES, RUS, RUS_ENG_TABLE, ENG_RUS_TABLE, PYTHONCHATRU, ZEN, LUTZ_ID, + BDMTSS_ID) +from .filters import in_spam_list, in_caption_spam_list, in_delete_list +from .helpers import represent_as_get, detect_args, is_admin +from .report import update_stats, increment +from .rules import fetch_rule +from .stan import act, speak + + +async def send_or_reply(message: types.Message, answer, **kwargs): + if message.reply_to_message: + await bot.reply_to(message.reply_to_message, answer, **kwargs) + else: + await bot.send_message(message.chat.id, answer, **kwargs) -from .helpers import represent_as_get, detect_args, update_stats -from .filters import * -from .config import * """ [ ANTISPAM ] """ @@ -16,8 +29,7 @@ async def moderate_messages(message: types.Message): """Ban user and delete their message.""" await bot.delete_message(message.chat.id, message.id) await bot.ban_chat_member(message.chat.id, message.from_user.id) - with shelve.open(f"{DATA}{message.chat.id}") as s: - s["Banned"] += 1 + increment(message.chat.id, banned=True) @bot.message_handler(func=in_caption_spam_list, content_types=["video"], chat_types=["supergroup", "group"]) @@ -25,8 +37,7 @@ async def catch_videos(message: types.Message): """Catch offensive videos""" await bot.delete_message(message.chat.id, message.id) await bot.ban_chat_member(message.chat.id, message.from_user.id) - with shelve.open(f"{DATA}{message.chat.id}") as s: - s["Banned"] += 1 + increment(message.chat.id, banned=True) @bot.edited_message_handler(func=in_delete_list, chat_types=["supergroup", "group"]) @@ -34,8 +45,7 @@ async def catch_videos(message: types.Message): async def delete_message(message: types.Message): """Delete unwanted message.""" await bot.delete_message(message.chat.id, message.id) - with shelve.open(f"{DATA}{message.chat.id}") as s: - s["Deleted"] += 1 + increment(message.chat.id, banned=False) """ [ COMMANDS ] """ @@ -58,7 +68,7 @@ async def send_rules(message: types.Message): if len(args) > 1 and args[-1].isdigit() and 0 < int(args[-1]): await send_or_reply( message, - f"Правило {args[-1]}\n{rules.fetch_rule(args[-1])}", + f"Правило {args[-1]}\n{fetch_rule(args[-1])}", reply_markup=markup, ) else: @@ -109,11 +119,7 @@ async def send_lutz(message: types.Message): message.text, ) ) - await bot.send_document( - message.chat.id, - document="BQACAgQAAxkBAAPBYsWJG9Ml0fPrnbU9UyzTQiQSuHkAAjkDAAIstCxSkuRbXAlcqeQpBA", - caption="вот, не позорься", - ) + await bot.send_document(message.chat.id, document=LUTZ_ID, caption="вот, не позорься") await bot.delete_message(message.chat.id, message.id) @@ -128,10 +134,7 @@ async def send_bdmtss_audio(message: types.Message): message.text, ) ) - await bot.send_voice( - message.chat.id, - "AwACAgIAAxkBAAIJrWOg2WUvLwrf7ahyJxQHB8_nqllwAAL5JQAC2_IJSbhfQIO5YnVmLAQ", - ) + await bot.send_voice(message.chat.id, BDMTSS_ID) await bot.delete_message(message.chat.id, message.id) @@ -164,7 +167,7 @@ async def stan_speak(message: types.Message): message.text, ) ) - await bot.send_message(message.chat.id, stan.speak(0, message.chat.id), parse_mode='Markdown') + await bot.send_message(message.chat.id, speak(0, message.chat.id), parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) @@ -180,10 +183,7 @@ async def send_tsya(message: types.Message): ) ) markup = types.InlineKeyboardMarkup() - markup.add( - types.InlineKeyboardButton("🧑🏼‍🎓 Читать правило", url="https://tsya.ru/"), - row_width=1, - ) + markup.add(types.InlineKeyboardButton("🧑🏼‍🎓 Читать правило", url="https://tsya.ru/"), row_width=1) await send_or_reply(message, "-тся и -ться в глаголах", reply_markup=markup) await bot.delete_message(message.chat.id, message.id) @@ -200,10 +200,7 @@ async def send_nometa(message: types.Message): ) ) markup = types.InlineKeyboardMarkup() - markup.add( - types.InlineKeyboardButton("❓ nometa.xyz", url="https://nometa.xyz/ru.html"), - row_width=1, - ) + markup.add(types.InlineKeyboardButton("❓ nometa.xyz", url="https://nometa.xyz/ru.html"), row_width=1) await send_or_reply( message, """Не задавай мета-вопросов вроде: @@ -229,13 +226,8 @@ async def send_neprivet(message: types.Message): ) ) markup = types.InlineKeyboardMarkup() - markup.add( - types.InlineKeyboardButton("👋 Непривет", url="https://neprivet.com/"), - row_width=1, - ) - await send_or_reply( - message, "Пожалуйста, не пишите просто «Привет» в чате.", reply_markup=markup - ) + markup.add(types.InlineKeyboardButton("👋 Непривет", url="https://neprivet.com/"), row_width=1) + await send_or_reply(message, "Пожалуйста, не пишите просто «Привет» в чате.", reply_markup=markup) await bot.delete_message(message.chat.id, message.id) @@ -282,8 +274,7 @@ async def google_it(message: types.Message): ) ) query = f"{detect_args(message)}" - get_query = "https://www.google.com/search?q=" + represent_as_get(message) - + get_query = f"https://www.google.com/search?q={represent_as_get(message)}" markup = types.InlineKeyboardMarkup() markup.add(types.InlineKeyboardButton("🔍 Google Поиск", url=get_query), row_width=1) await send_or_reply(message, f"Ищем «{query}»...", reply_markup=markup) @@ -360,22 +351,9 @@ async def default_query(inline_query): """ [ COUNTER ] """ -@bot.message_handler(content_types=["text", "sticker", "photo", "animation", "video", "audio", "document", ], +@bot.message_handler(content_types=["text", "sticker", "photo", "animation", "video", "audio", "document"], chat_types=["supergroup", "group"]) async def handle_msg(message: types.Message): """Count messages, Stan.""" update_stats(message) - await stan.act(message) - - -""" [ WEBHOOK ] """ - - -@app.post(f"/bot{TOKEN}/") -async def webhook(update: dict): - """Parse POST requests from Telegram.""" - if update: - update = telebot.types.Update.de_json(update) - await bot.process_new_updates([update]) - else: - return + await act(message) diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..1ca5646 --- /dev/null +++ b/src/config.py @@ -0,0 +1,24 @@ +import logging +import os + +from dotenv import load_dotenv +from telebot import logger + +from .helpers import is_url_reachable + +load_dotenv() + +TOKEN = os.environ.get("LUTZPYBOT") +if not TOKEN: + raise LookupError("LUTZPYBOT (token) has not been found in .env") + +WHITEIDS = {int(i) for i in os.environ.get("whiteids").split(",")} +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") +if not is_url_reachable(RULES_URL): + raise LookupError(f"LUTZPYBOT rules url({RULES_URL}) is not reachable") + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") +logger.setLevel(logging.INFO) diff --git a/app/config.py b/src/constants.py similarity index 76% rename from app/config.py rename to src/constants.py index 75976c4..15c30dc 100644 --- a/app/config.py +++ b/src/constants.py @@ -1,40 +1,15 @@ -import os import re -import logging +import string -import telebot.types -from fastapi import FastAPI -from telebot.async_telebot import AsyncTeleBot -from telebot import logger, types -from dotenv import load_dotenv +from telebot import types -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") -logger.setLevel(logging.ERROR) - -logging.warning("[START] Stan") - -load_dotenv() - -TOKEN = os.environ.get("LUTZPYBOT") -if not TOKEN: - raise LookupError("LUTZPYBOT (token) has not been found in .env") - -bot = AsyncTeleBot( - TOKEN, - "HTML", - disable_web_page_preview=True, - allow_sending_without_reply=True, - colorful_logs=True, -) LOG_TEXT = "[%s] %s: %s" LOG_COMM = "[%s] [%s] %s: %s" -app = FastAPI(docs=None, redoc_url=None) - +LOGGING_LEVEL_DEBUG = 10 +LOGGING_LEVEL_INFO = 20 DATA = "data/chat" - ADMIN_ID = 280887861 # Rykov7 PYTHONCHATRU = -1001338616632 # pythonchatru - URL_RX = re.compile(r"\w+\.\w+/(\+)?\w+") ALLOWED_WORDS = [ "paste", @@ -61,20 +36,15 @@ ALLOWED_WORDS = [ "jetbrains", ] -WHITEIDS = {int(i) for i in os.environ.get("whiteids").split(",")} -ROLLBACK = {int(i) for i in os.environ.get("rollback").split(",")} - RUS = """ёйцукенгшщзхъфывапролджэячсмитьбю.Ё!"№;%:?ЙЦУКЕНГШЩЗХЪ/ФЫВАПРОЛДЖЭЯЧСМИТЬБЮ,""" +ONLY_RUS_LETTERS = "ёйцукенгшщзхъфывапролджэячсмитьбю" +RUS_LETTERS_AND_PUNCTUATION = ONLY_RUS_LETTERS + string.punctuation ENG = """`qwertyuiop[]asdfghjkl;'zxcvbnm,./~!@#$%^&QWERTYUIOP{}|ASDFGHJKL:"ZXCVBNM<>?""" RUS_ENG_TABLE = str.maketrans(RUS, ENG) ENG_RUS_TABLE = str.maketrans(ENG, RUS) -RULES = types.InlineKeyboardButton( - "🟡 Правила чата", url="https://telegra.ph/pythonchatru-07-07" -) -FAQ = types.InlineKeyboardButton( - "🔵 Частые вопросы", url="https://telegra.ph/faq-10-07-4" -) +RULES = types.InlineKeyboardButton("🟡 Правила чата", url="https://telegra.ph/pythonchatru-07-07") +FAQ = types.InlineKeyboardButton("🔵 Частые вопросы", url="https://telegra.ph/faq-10-07-4") LIB = types.InlineKeyboardButton("📚 Книги", url="https://telegra.ph/what-to-read-10-06") SPAM = {"me.sv/", "tg.sv/", "goo.by/", "go.sv/", "intim.video/", "uclck.ru/", 'раб0т', @@ -87,7 +57,9 @@ 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"} @@ -108,4 +80,5 @@ ZEN = [ "If the implementation is easy to explain, it may be a good idea.", "Namespaces are one honking great idea — let's do more of those!", ] - +LUTZ_ID = "BQACAgQAAxkBAAPBYsWJG9Ml0fPrnbU9UyzTQiQSuHkAAjkDAAIstCxSkuRbXAlcqeQpBA" +BDMTSS_ID = "AwACAgIAAxkBAAIJrWOg2WUvLwrf7ahyJxQHB8_nqllwAAL5JQAC2_IJSbhfQIO5YnVmLAQ" diff --git a/src/filters.py b/src/filters.py new file mode 100644 index 0000000..87f68bb --- /dev/null +++ b/src/filters.py @@ -0,0 +1,47 @@ +import logging.handlers + +from telebot import types + +from .config import WHITEIDS +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 is_antispam_enabled + + +def in_spam_list(message: types.Message) -> bool: + antispam_is_enabled = is_antispam_enabled(message.chat.id) + 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 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.""" + 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_delete_list(message: types.Message) -> bool: + """Check for URLs in message and delete.""" + antispam_is_enabled = is_antispam_enabled(message.chat.id) + 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 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 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_white_id(message: types.Message) -> bool: + return message.from_user.id in WHITEIDS diff --git a/src/helpers.py b/src/helpers.py new file mode 100644 index 0000000..b8f57e2 --- /dev/null +++ b/src/helpers.py @@ -0,0 +1,97 @@ +""" Send User's info. """ + +from urllib import parse +from urllib.error import URLError +from urllib.request import urlopen + +from telebot import types + +from .constants import SPAM, BAN_WORDS, NON_GRATA, ADMIN_ID, ONLY_RUS_LETTERS, RUS_LETTERS_AND_PUNCTUATION + + +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(): + word = word.strip().lower() + if any(e in ONLY_RUS_LETTERS for e in word) and not all(e in RUS_LETTERS_AND_PUNCTUATION for e in word): + 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(): + ip_service = "https://icanhazip.com" + ip = urlopen(ip_service) + return ip.read().decode("utf-8") + + +def me(message: types.Message): + msg = f"Telegram ID: {message.from_user.id}\n\ + ├ Is bot: {message.from_user.is_bot}\n\ + ├ First name: {message.from_user.first_name}\n" + if message.from_user.last_name: + msg += f" ├ Last name: {message.from_user.last_name}\n" + if message.from_user.username: + msg += f" ├ Username: {message.from_user.username}\n" + msg += f" ├ Language code: {message.from_user.language_code}\n" + if message.from_user.is_premium is True: + msg += f" ├ Is Premium: ⭐️\n" + else: + msg += f" ├ Is Premium: No\n" + msg += f" ├ Chat ID: {message.chat.id}" + return msg + + +def represent_as_get(message: types.Message): + return parse.quote_plus(detect_args(message)) + + +def detect_args(message: types.Message): + if message.reply_to_message and message.reply_to_message.text: + if len(message.text.split()) == 1: + return message.reply_to_message.text + else: + return " ".join(message.text.split()[1:]) + else: + if len(message.text.split()) > 1: + return " ".join(message.text.split()[1:]) diff --git a/src/models.py b/src/models.py new file mode 100644 index 0000000..50acc92 --- /dev/null +++ b/src/models.py @@ -0,0 +1,78 @@ +from pathlib import Path + +from sqlalchemy import Column, String, Integer, Boolean, ForeignKey +from sqlalchemy import create_engine +from sqlalchemy import delete +from sqlalchemy.orm import relationship, declarative_base, sessionmaker + +Base = declarative_base() + +root = Path(__file__).parent.parent +data_folder = root / "data" +db_path = data_folder / "db.sqlite" +engine = create_engine(f'sqlite:///{db_path}') + +Base.metadata.create_all(engine) +Session = sessionmaker(bind=engine) +session = Session() + + +class Permission: + USE_COMMANDS = 1 # Пользоваться обычными командами. + SEND_LINKS = 2 # Может отправлять ссылки без удаления. + MODIFY_QUOTES = 4 # Может добавлять и удалять цитаты. + MODERATE = 8 + ADMIN = 16 # Может банить юзеров и удалять сообщения. + + +class Quote(Base): + __tablename__ = "quotes" + id = Column(Integer, primary_key=True) + chat_id = Column(Integer, ForeignKey('chats.chat_id')) + text = Column(String) + + def __repr__(self): + return f"Quote({self.text=}, {self.chat_id=})" + + +class Chat(Base): + __tablename__ = "chats" + id = Column(Integer, primary_key=True) + chat_id = Column(Integer, unique=True) + title = Column(String) + + # Settings + quotes = relationship('Quote', backref='chat', cascade='all, delete') + antispam = Column(Boolean) + report = Column(Boolean) + reminder = Column(Boolean) + hidden = Column(Boolean) + + def __repr__(self): + return f'Chat({self.chat_id=}, {self.title=})' + + +def is_antispam_enabled(chat_id: int) -> bool: + try: + return session.query(Chat.antispam).filter_by(chat_id=chat_id).first()[0] + except TypeError: + return False + + +def is_quote_in_chat(text: str, chat_id: int) -> bool: + result = session.query(Quote.text).filter_by(text=text, chat_id=chat_id).first() + return bool(result) + + +def delete_quote_in_chat(text: str, chat_id: int): + session.execute(delete(Quote).filter_by(text=text, chat_id=chat_id)) + session.commit() + + +def all_chat_quotes(chat_id: int) -> list: + return session.query(Quote.text).filter(Quote.chat_id == chat_id).all() + + +def add_quote(chat_id: int, text: str): + session.add(Quote(chat_id=chat_id, text=text)) + session.commit() diff --git a/app/reminder.py b/src/reminder.py similarity index 75% rename from app/reminder.py rename to src/reminder.py index e09b987..3591abe 100644 --- a/app/reminder.py +++ b/src/reminder.py @@ -1,15 +1,23 @@ """ Schedule holidays. """ -from schedule import repeat, every, get_jobs, run_pending -import time -from threading import Thread -from datetime import datetime as dt import csv -from .config import PYTHONCHATRU, TOKEN -from . import report +import logging +import time +from datetime import datetime as dt +from threading import Thread + +from schedule import repeat, every, get_jobs, run_pending from telebot import TeleBot -# We need synchronous instance of bot here as we use synchronous library 'schedule' in a separate thread. -bot = TeleBot(TOKEN, "HTML", disable_web_page_preview=True, allow_sending_without_reply=True) +from .config import TOKEN, USE_REMINDER +from .constants import PYTHONCHATRU +from .report import create_report_text, reset_report_stats + +if USE_REMINDER: + logging.debug("Starts reminder!!!") + # We need synchronous instance of bot here as we use synchronous library 'schedule' in a separate thread. + bot = TeleBot(TOKEN, "HTML", disable_web_page_preview=True, allow_sending_without_reply=True) +else: + bot = None def scheduler(): @@ -42,10 +50,10 @@ def remind(chat_to_repeat, today): @repeat(every().day.at("06:00"), PYTHONCHATRU) def stat_report(chat_to_repeat): """Everyday group statistic info.""" - rep = report.create_report_text(PYTHONCHATRU) + rep = create_report_text(PYTHONCHATRU) if rep: bot.send_message(chat_to_repeat, rep) - report.reset_report_stats(PYTHONCHATRU) + reset_report_stats(PYTHONCHATRU) def print_get_jobs(): @@ -58,4 +66,5 @@ def print_get_jobs(): return text -Thread(target=scheduler).start() +if USE_REMINDER: + Thread(target=scheduler).start() diff --git a/app/report.py b/src/report.py similarity index 58% rename from app/report.py rename to src/report.py index 9953e10..e09c571 100644 --- a/app/report.py +++ b/src/report.py @@ -1,8 +1,21 @@ import html +import logging +import os import random import shelve -import os -from .config import DATA, ROLLBACK +from pathlib import Path + +from .config import ROLLBACK +from .constants import DATA, types + + +def increment(chat_id: int, banned: bool = True): + """ + Увеличивает счетчик банов или удаленных + """ + category = "Banned" if banned else "Deleted" + with shelve.open(f"{DATA}{chat_id}") as shelve_db: + shelve_db[category] += 1 def create_report_text(chat_id): @@ -44,7 +57,7 @@ def create_report_text(chat_id): if s["Banned"]: report += f""" Криптоидов поймано: {s['Banned']} - └ [ {s['Banned']*current_spammer_face} ] + └ [ {s['Banned'] * current_spammer_face} ] """ return report else: @@ -52,12 +65,25 @@ def create_report_text(chat_id): def reset_report_stats(chat_id): - with shelve.open(f"{DATA}{chat_id}") as s: - s["Messages"] = {} - s["Banned"] = 0 - s["Deleted"] = 0 + with shelve.open(f"{DATA}{chat_id}") as shelve_db: + shelve_db["Messages"] = {} + shelve_db["Banned"] = 0 + shelve_db["Deleted"] = 0 return f"""Chat ID: {chat_id} -{s['Messages']=} -{s['Banned']=} -{s['Deleted']=}""" +{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/robot.py b/src/robot.py new file mode 100644 index 0000000..4ba4d95 --- /dev/null +++ b/src/robot.py @@ -0,0 +1,14 @@ +import logging + +from telebot.async_telebot import AsyncTeleBot + +from .config import TOKEN + +logging.warning("[START] Stan") +bot = AsyncTeleBot( + TOKEN, + "HTML", + disable_web_page_preview=True, + allow_sending_without_reply=True, + colorful_logs=True, +) diff --git a/app/rules.py b/src/rules.py similarity index 75% rename from app/rules.py rename to src/rules.py index 6a34b94..8af710b 100644 --- a/app/rules.py +++ b/src/rules.py @@ -1,7 +1,10 @@ -from bs4 import BeautifulSoup from urllib.request import urlopen -html = BeautifulSoup(urlopen("https://telegra.ph/pythonchatru-07-07"), "lxml") +from bs4 import BeautifulSoup + +from .config import RULES_URL + +html = BeautifulSoup(urlopen(RULES_URL), "lxml") def fetch_rule(li): diff --git a/app/stan.py b/src/stan.py similarity index 56% rename from app/stan.py rename to src/stan.py index 87160a4..42b7e56 100644 --- a/app/stan.py +++ b/src/stan.py @@ -3,24 +3,27 @@ import asyncio import html import random -from sqlalchemy import delete +from telebot import types -from .config import bot, types -from .filters import is_white, is_nongrata -from .models import session -from .models import Quote +from .filters import is_white_id +from .helpers import is_nongrata +from .models import all_chat_quotes, add_quote, is_quote_in_chat, delete_quote_in_chat +from .robot import bot + +TYPING_TIMEOUT = 0.13 / 4 # Reading time is quarter of the same text writing time -def speak(chance_of, group_id): +def speak(chance_of: int, chat_id: int) -> None | str: number = random.randint(0, chance_of) if number == 0: - return random.choice([i[0] for i in session.query(Quote.text).filter(Quote.chat_id == group_id).all()]) + return random.choice([i[0] for i in all_chat_quotes(chat_id)]) + return None async def send_quote(after_sec, message, quote): """Pretend Reading, pretend Typing, send.""" if message.text: - await asyncio.sleep(len(message.text) * 0.13 / 4) # Reading time is quarter of the same text writing time + await asyncio.sleep(len(message.text) * TYPING_TIMEOUT) await bot.send_chat_action(message.chat.id, action="typing") await asyncio.sleep(after_sec) # Typing time await bot.send_message(message.chat.id, html.escape(quote)) @@ -29,33 +32,27 @@ async def send_quote(after_sec, message, quote): async def act(message: types.Message): quote = speak(50, message.chat.id) if quote: - await send_quote(len(quote) * 0.13, message, quote ) + 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 - if quote not in [i[0] for i in session.query(Quote.text).filter(Quote.chat_id == message.chat.id).all()]: - session.add(Quote(chat_id=message.chat.id, text=quote.replace("\n", " "))) - session.commit() - await bot.send_message(message.chat.id, "➕\n └ " + quote.replace("\n", " "), - parse_mode='Markdown') + if quote not in {i[0] for i in all_chat_quotes(message.chat.id)}: + add_quote(message.chat.id, quote.replace("\n", " ")) + await bot.send_message(message.chat.id, "➕\n └ " + quote.replace("\n", " "), parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) else: 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 - already_exist = session.query(Quote.text).filter_by(text=quote, chat_id=message.chat.id).first() - if already_exist: - session.execute(delete(Quote).filter_by(text=quote, chat_id=message.chat.id)) - session.commit() - + if is_quote_in_chat(quote, message.chat.id): + delete_quote_in_chat(quote, message.chat.id) await bot.send_message(message.chat.id, f"➖ \n └ {quote}", parse_mode='Markdown') await bot.delete_message(message.chat.id, message.id) else: @@ -65,4 +62,4 @@ async def remove_stan_quote(message: types.Message): @bot.message_handler(func=is_nongrata) async def tease_nongrata(message: types.Message): """Reply to non grata mentions.""" - await bot.reply_to(message, f"у нас тут таких не любят") + await bot.reply_to(message, "у нас тут таких не любят") diff --git a/src/webhook.py b/src/webhook.py new file mode 100644 index 0000000..1a7929e --- /dev/null +++ b/src/webhook.py @@ -0,0 +1,20 @@ +import logging + +import telebot +from fastapi import FastAPI + +from .commands import bot +from .config import TOKEN + +logging.warning("[START] Webhook server") +app = FastAPI(docs=None, redoc_url=None) + + +@app.post(f"/bot{TOKEN}/") +async def webhook(update: dict): + """Parse POST requests from Telegram.""" + if update: + update = telebot.types.Update.de_json(update) + await bot.process_new_updates([update]) + else: + return None diff --git a/tests/test_bot.py b/tests/test_bot.py new file mode 100644 index 0000000..c6da398 --- /dev/null +++ b/tests/test_bot.py @@ -0,0 +1,348 @@ +import logging +from unittest import IsolatedAsyncioTestCase, main, TestCase +import os +from unittest.mock import patch, MagicMock + +from telebot import asyncio_helper, types, logger + +# Надо сделать до импорта бота, иначе упадет, так как нет енва в тестах +os.environ["LUTZPYBOT"] = "00000:AAAAAAAAAAAA" +os.environ["whiteids"] = "100,200,300" +os.environ["rollback"] = "1,2,3" +os.environ["use_reminder"] = "FALSE" + +from src import helpers + +# нужно не дать делать реальные запросы в тестах, конкретно тут -за Правилами чата +helpers.is_url_reachable = lambda a: True +from src.commands import bot +from src.constants import BDMTSS_ID, LUTZ_ID, ADMIN_ID +from src.filters import is_white_id + +# отключаем логи бота, в тестах они не нужны +logger.setLevel(logging.ERROR) + +# сюда будут складываться результаты, перед началом теста очищается +RESULTS = [] + + +def get_update(text, reply_to=None, user_id=10): + """Хелпер, генерирующий событие обновления для бота""" + params = {'text': text} + chat = types.Chat(id=11, type='group') + user = types.User(id=user_id, is_bot=False, first_name='Some User') + if reply_to: + params["reply_to_message"] = types.Message(message_id=2, 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) + + +async def custom_sender(token, url, method='get', params=None, files=None, **kwargs): + """Замена для тестов, чтобы не слать запросы в телегу""" + RESULTS.append((params, url)) + result = {"message_id": 1000, "date": 1, "chat": {"id": 1000, "type": "group"}} + return result + + +# Чтобы перехватить запросы к телеге +asyncio_helper._process_request = custom_sender + + +class TestOther(TestCase): + def test_is_white(self): + params = ( + (False, 1), + (False, 10), + (True, 100), + (True, 200), + (True, 300), + ) + for expected, text in params: + with self.subTest(f"check non-is_white_id({text})"): + m = MagicMock(from_user=MagicMock(id=text)) + self.assertEqual(is_white_id(m), expected) + + +class TestBot(IsolatedAsyncioTestCase): + + def setUp(self) -> None: + self.bot = bot + RESULTS.clear() + + async def test_start(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/start')]) + self.assertEqual(RESULTS[0][0]['text'], "Начни с прочтения") + + async def test_non_grata(self): + await self.bot.process_new_updates([get_update('как вам дударь?')]) + self.assertEqual(RESULTS[0][0]['text'], "у нас тут таких не любят") + + async def test_non_grata2(self): + await self.bot.process_new_updates([get_update('смотрю хауди')]) + self.assertEqual(RESULTS[0][0]['text'], "у нас тут таких не любят") + + async def test_rules(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/rules')]) + self.assertEqual(RESULTS[0][0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/pythonchatru-07-07" in RESULTS[0][0]['reply_markup']) + + async def test_faq(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/faq')]) + self.assertEqual(RESULTS[0][0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/faq-10-07-4" in RESULTS[0][0]['reply_markup']) + + async def test_lib(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/lib')]) + self.assertEqual(RESULTS[0][0]['text'], "Читай...") + self.assertTrue("https://telegra.ph/what-to-read-10-06" in RESULTS[0][0]['reply_markup']) + + async def test_tr(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/tr', reply_to="ghbdtn")]) + self.assertEqual(RESULTS[0][0]['text'], "привет") + + async def test_tr2(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/tr', reply_to="руддщ")]) + self.assertEqual(RESULTS[0][0]['text'], "hello") + + async def test_tsya(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/tsya')]) + self.assertEqual(RESULTS[0][0]['text'], "-тся и -ться в глаголах") + self.assertTrue("https://tsya.ru/" in RESULTS[0][0]['reply_markup']) + + async def test_nometa(self): + text = """Не задавай мета-вопросов вроде:\n «Можно задать вопрос?»\n «Кто-нибудь пользовался .. ?»\n «Привет, мне нужна помощь по .. !»\n\nПросто спроси сразу! И чем лучше объяснишь проблему, тем вероятнее получишь помощь.""" + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/nometa', reply_to="кто тут питонист?")]) + self.assertEqual(RESULTS[0][0]['text'], text) + + async def test_neprivet(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/neprivet', reply_to="привет всем")]) + self.assertEqual(RESULTS[0][0]['text'], "Пожалуйста, не пишите просто «Привет» в чате.") + self.assertTrue("https://neprivet.com/" in RESULTS[0][0]['reply_markup']) + + async def test_lutz(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/lutz', reply_to="привет всем")]) + self.assertEqual(RESULTS[0][0]['caption'], "вот, не позорься") + self.assertEqual(RESULTS[0][0]['document'], LUTZ_ID) + self.assertEqual(RESULTS[0][1], "sendDocument") + self.assertEqual(RESULTS[1][1], "deleteMessage") + + async def test_bdmtss(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/bdmtss', reply_to="привет всем")]) + self.assertEqual(RESULTS[0][0]['voice'], BDMTSS_ID) + self.assertEqual(RESULTS[0][1], "sendVoice") + self.assertEqual(RESULTS[1][1], "deleteMessage") + + async def test_google(self): + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/g', reply_to="python")]) + self.assertEqual(RESULTS[0][0]['text'], "Ищем «python»...") + self.assertTrue("https://www.google.com/search?q=python" in RESULTS[0][0]['reply_markup']) + + async def test_nojob(self): + text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в ТВОЁМ коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️""" + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/nojob', reply_to="есть работа")]) + self.assertEqual(RESULTS[0][0]['text'], text) + + async def test_nobot(self): + text = """Внимание:\nТелеграм бот не должен быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, основы веб-технологий, асинхронное программирование и отладку до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете.""" + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/nobot', reply_to="как бота написать?")]) + self.assertEqual(RESULTS[0][0]['text'], text) + + async def test_nogui(self): + text = """Внимание:\nGUI приложение не должно быть твоим первым проектом на Python. Пожалуйста, изучи основы Python, работу с модулями, циклы событий и отладку до начала работы с какими-либо GUI-фреймворками. Существует много ресурсов для этого в интернете.""" + # Нужно пропустить проверку на спам + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/nogui', reply_to="как gui написать?")]) + self.assertEqual(RESULTS[0][0]['text'], text) + + async def test_spam_text(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = True + await self.bot.process_new_updates([get_update('пишите мне в ЛС, Писать в ЛС')]) + self.assertEqual(RESULTS[0][1], "deleteMessage") + self.assertEqual(RESULTS[1][1], "banChatMember") + self.assertEqual(RESULTS[1][0], {'chat_id': 11, 'user_id': 10, 'until_date': None}) + + async def test_wrong_url(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = True + await self.bot.process_new_updates([get_update('тут подробнее https://zvuk.com/artist/61375')]) + self.assertEqual(RESULTS[0][1], "deleteMessage") + self.assertEqual(RESULTS[0][0], {'chat_id': 11, 'message_id': 1}) + + async def test_update_stats(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.commands.update_stats") as update_stats: + mocked.return_value = True + update_stats.side_effect = lambda x: RESULTS.append(x) + await self.bot.process_new_updates([get_update('просто текст')]) + self.assertEqual(RESULTS[0].text, "просто текст") + + async def test_admin_can_delete(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + mocked.return_value = True + await self.bot.process_new_updates([get_update('/ddel', reply_to="shit text", user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0], ({'chat_id': 11, 'message_id': 1}, 'deleteMessage')) + self.assertEqual(RESULTS[1], ({'chat_id': 11, 'message_id': 2}, 'deleteMessage')) + + async def test_user_cant(self): + for command in ('/ddel', '/bban', '/unban_id', '/unban_id 1', '/add', '/remove'): + with self.subTest(f"user cant use {command}"): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.commands.update_stats") as update_stats: + # Подменяем speak, чтобы Стен не начал отвечать + with patch("src.stan.speak") as speak: + speak.return_value = None + mocked.return_value = True + update_stats.side_effect = lambda x: None + await self.bot.process_new_updates([get_update(command, reply_to="shit text")]) + self.assertEqual(RESULTS, []) + + async def test_admin_can_ban(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.commands.update_stats") as update_stats: + mocked.return_value = True + update_stats.side_effect = lambda x: None + await self.bot.process_new_updates([get_update('/bban', reply_to="shit text", user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0], ({'chat_id': 11, 'message_id': 1}, 'deleteMessage')) + self.assertEqual(RESULTS[1], ({'chat_id': 11, 'message_id': 2}, 'deleteMessage')) + self.assertEqual(RESULTS[2][1], 'banChatMember') + + async def test_admin_can_unban(self): + # Нужно подставить антиспам вкл. + with patch("src.filters.is_antispam_enabled") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.commands.update_stats") as update_stats: + mocked.return_value = True + update_stats.side_effect = lambda x: None + await self.bot.process_new_updates( + [get_update('/unban_id 1', reply_to="shit text", user_id=ADMIN_ID)]) + self.assertEqual(RESULTS[0][1], 'unbanChatMember') + + async def test_white_id_can_add_quote(self): + # Нужно подставить список цитат + with patch("src.stan.all_chat_quotes") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем обновление статистики, нам важно что оно просто было вызвано + with patch("src.stan.add_quote") as aq: + mocked.return_value = [] + aq.side_effect = lambda *a: None + await self.bot.process_new_updates([get_update('/add', reply_to="good text", user_id=100)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], '➕\n └ good text') + self.assertEqual(RESULTS[1][1], 'deleteMessage') + + async def test_white_id_cant_add_existing_quote(self): + # Нужно подставить список цитат + with patch("src.stan.all_chat_quotes") as mocked: + # Нужно подменить инкремент, нельзя реально в шелве работать в тестах + with patch("src.commands.increment") as _mocked_shelve: + # Подменяем реальное добавление цитаты, нам важно что оно просто было вызвано + with patch("src.stan.add_quote") as add_qoute: + mocked.return_value = [["good text"]] + add_qoute.side_effect = lambda x, y: None + await self.bot.process_new_updates([get_update('/add', reply_to="good text", user_id=100)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], '⛔️ Не добавил, есть токое\n └ good text') + + async def test_white_id_cant_remove_non_existing_quote(self): + # Нужно вернуть запрос о наличии цитаты + with patch("src.stan.is_quote_in_chat") as mocked: + mocked.return_value = False + await self.bot.process_new_updates([get_update('/remove', reply_to="good text", user_id=100)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[0][0]['text'], '⛔️ Нет такого\n └ good text') + + async def test_white_id_can_remove_existing_quote(self): + # Нужно вернуть запрос о наличии цитаты + with patch("src.stan.is_quote_in_chat") as mocked: + # Подменяем удаление цитаты, нам важно что оно просто было вызвано + with patch("src.stan.delete_quote_in_chat") as delete_qoute: + mocked.return_value = [["good text"]] + delete_qoute.side_effect = lambda x, y: None + mocked.return_value = True + await self.bot.process_new_updates([get_update('/remove', reply_to="good text", user_id=100)]) + self.assertEqual(RESULTS[0][1], 'sendMessage') + self.assertEqual(RESULTS[1][1], 'deleteMessage') + self.assertEqual(RESULTS[0][0]['text'], '➖ \n └ good text') + + +if __name__ == '__main__': + main() diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..9ad709b --- /dev/null +++ b/tests/test_helpers.py @@ -0,0 +1,78 @@ +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("привет ребята!")) + self.assertFalse(is_mixed("hi guys")) + self.assertFalse(is_mixed("hi guys! #%^&%^$ @Вася привет g! п@")) + self.assertFalse(is_mixed("тут подробнее https://hack.comp.com")) + self.assertTrue(is_mixed("hi guys! #%^&%^$ ghb привет g! пQ")) + + + +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 b334fe5..8fc75a2 100644 --- a/wsgi.py +++ b/wsgi.py @@ -1 +1 @@ -from app.bot import app \ No newline at end of file +from src.webhook import app