Merge pull request #1 from kotolex/master

.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

app/__init__.py

@@ -1 +0,0 @@
-from .commands import service

app/commands/get.py

@@ -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"<b>Telegram ID:</b> <code>{message.from_user.id}</code>\n\
-    ├ <b>Is bot:</b> {message.from_user.is_bot}\n\
-    ├ <b>First name:</b> {message.from_user.first_name}\n"
-    if message.from_user.last_name:
-        msg += f"    ├ <b>Last name:</b> {message.from_user.last_name}\n"
-    if message.from_user.username:
-        msg += f"    ├ <b>Username:</b> {message.from_user.username}\n"
-    msg += f"    ├ <b>Language code:</b> {message.from_user.language_code}\n"
-    if message.from_user.is_premium is True:
-        msg += f"    ├ <b>Is Premium</b>: ⭐️\n"
-    else:
-        msg += f"    ├ <b>Is Premium</b>: No\n"
-    msg += f"    ├ <b>Chat ID:</b> {message.chat.id}"
-
-    return msg

app/filters.py

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

app/helpers.py

@@ -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

app/models.py

@@ -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=})'

src/__init__.py

@@ -0,0 +1 @@
+

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

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"<b>Правило {args[-1]}</b>\n<i>{rules.fetch_rule(args[-1])}</i>",
+            f"<b>Правило {args[-1]}</b>\n<i>{fetch_rule(args[-1])}</i>",
             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, "<i>-тся</i> и <i>-ться</i> в глаголах", 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"<i>{detect_args(message)}</i>"
-    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"<i>Ищем «{query}»...</i>", 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)

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)

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"

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

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"<b>Telegram ID:</b> <code>{message.from_user.id}</code>\n\
+    ├ <b>Is bot:</b> {message.from_user.is_bot}\n\
+    ├ <b>First name:</b> {message.from_user.first_name}\n"
+    if message.from_user.last_name:
+        msg += f"    ├ <b>Last name:</b> {message.from_user.last_name}\n"
+    if message.from_user.username:
+        msg += f"    ├ <b>Username:</b> {message.from_user.username}\n"
+    msg += f"    ├ <b>Language code:</b> {message.from_user.language_code}\n"
+    if message.from_user.is_premium is True:
+        msg += f"    ├ <b>Is Premium</b>: ⭐️\n"
+    else:
+        msg += f"    ├ <b>Is Premium</b>: No\n"
+    msg += f"    ├ <b>Chat ID:</b> {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:])

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

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

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"""
 <b> Криптоидов поймано:</b> {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

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,
+)

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

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, "у нас тут таких не любят")

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

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'], "<i>-тся</i> и <i>-ться</i> в глаголах")
+            self.assertTrue("https://tsya.ru/" in RESULTS[0][0]['reply_markup'])
+
+    async def test_nometa(self):
+        text = """Не задавай мета-вопросов вроде:\n<i>  «Можно задать вопрос?»\n  «Кто-нибудь пользовался .. ?»\n  «Привет, мне нужна помощь по .. !»</i>\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'], "<i>Ищем «<i>python</i>»...</i>")
+            self.assertTrue("https://www.google.com/search?q=python" in RESULTS[0][0]['reply_markup'])
+
+    async def test_nojob(self):
+        text = """Мы здесь не для того, чтобы за тебя решать задачи.\n\nЗдесь помогают по конкретным вопросам в <u>ТВОЁМ</u> коде, поэтому тебе нужно показать код, который ты написал сам и объяснить где и почему застрял... всё просто. 🤷🏼️"""
+        # Нужно пропустить проверку на спам
+        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 = """<b>Внимание</b>:\nТелеграм бот <i>не должен</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>основы веб-технологий</code>, <code>асинхронное программирование</code> и <code>отладку</code> до начала работы с Телеграм ботами. Существует много ресурсов для этого в интернете."""
+        # Нужно пропустить проверку на спам
+        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 = """<b>Внимание</b>:\nGUI приложение <i>не должно</i> быть твоим первым проектом на Python. Пожалуйста, изучи <code>основы Python</code>, <code>работу с модулями</code>, <code>циклы событий</code> и <code>отладку</code> до начала работы с какими-либо 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()

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

wsgi.py

@@ -1 +1 @@
-from app.bot import app
+from src.webhook import app