leo/Stan-Ansible
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactoring: step 1

Browse Source 
- пакет app переименован в src
- файл БД теперь должен лежать в data
- удалена папка коммандс, все команды вынесены в модуль коммандс или админ_коммандс
- нет импортов в ините
- убраны все импорты *
- теперь в .env можно указать use_reminder="FALSE", тогда не будет стартовать ремайндер
- можно указать урл для Правил
- в конфиге теперь только конфиг -то, что из енва получаем
- константы теперь лежат отдельно
- добавлены явные return там, где не было
- бот и вебхук разнесены отдельно
- при старте проверяем, что Правила доступны
- добавлено еще логирование
This commit is contained in:
kotolex
2023-12-22 19:16:57 +05:00
parent 1cd96220a5
commit 9103669cac
19 changed files with 268 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/__init__.py
-1
View File
@@ -1 +0,0 @@
from .commands import service
app/commands/get.py
-27
View File
@@ -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/helpers.py
-39
View File
@@ -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
requirements.txt
BIN
View File
Binary file not shown.
src/__init__.py
+1
View File
@@ -0,0 +1 @@
app/commands/service.py → src/admin_commands.py
+27 -44
View File
@@ -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, is_admin
from .helpers import me, my_ip
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"])
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,10 +90,7 @@ 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, "Отказ. Этой группы нет в БД.")
@@ -130,21 +117,17 @@ 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, 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"])
@@ -172,8 +155,8 @@ ID группы: {group.chat_id}
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")
app/bot.py → src/commands.py
+31 -44
View File
@@ -1,11 +1,23 @@
import logging
import logging.handlers
import shelve
from . import stan
from . import rules
from telebot import types
from .constants import DATA, LOG_COMM, FAQ, LIB, RULES, RUS, RUS_ENG_TABLE, ENG_RUS_TABLE, PYTHONCHATRU, ZEN
from .filters import in_spam_list, in_caption_spam_list, in_delete_list, is_admin
from .helpers import represent_as_get, detect_args, update_stats
from .filters import *
from .config import *
from .admin_commands import bot
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)
""" [ ANTISPAM ] """
@@ -16,8 +28,8 @@ 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
with shelve.open(f"{DATA}{message.chat.id}") as shelve_db:
shelve_db["Banned"] += 1
@bot.message_handler(func=in_caption_spam_list, content_types=["video"], chat_types=["supergroup", "group"])
@@ -25,8 +37,8 @@ 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
with shelve.open(f"{DATA}{message.chat.id}") as shelve_db:
shelve_db["Banned"] += 1
@bot.edited_message_handler(func=in_delete_list, chat_types=["supergroup", "group"])
@@ -34,8 +46,8 @@ 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
with shelve.open(f"{DATA}{message.chat.id}") as shelve_db:
shelve_db["Deleted"] += 1
""" [ COMMANDS ] """
@@ -58,7 +70,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:
@@ -164,7 +176,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 +192,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 +209,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 +235,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 +283,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 +360,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
+26
View File
@@ -0,0 +1,26 @@
import logging
import os
from urllib.error import URLError
from urllib.request import urlopen
from dotenv import load_dotenv
from telebot import logger
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")
try:
assert urlopen(RULES_URL).status == 200
except (URLError, AssertionError):
raise LookupError(f"LUTZPYBOT rules url({RULES_URL}) is not reachable")
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
logger.setLevel(logging.DEBUG)
app/config.py → src/constants.py
+6 -37
View File
@@ -1,40 +1,15 @@
import os
import re
import logging
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,13 @@ 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 = """ёйцукенгшщзхъфывапролджэячсмитьбю.Ё!"№;%:?ЙЦУКЕНГШЩЗХЪ/ФЫВАПРОЛДЖЭЯЧСМИТЬБЮ,"""
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т',
@@ -88,6 +56,7 @@ SPAM = {"me.sv/", "tg.sv/", "goo.by/", "go.sv/", "intim.video/", "uclck.ru/", '
'P2P', 'Идет набор', 'в лс', 'нaбop', 'комaндy', 'kpuпт', 'бupж', '💰', 'пишитe', '💸',
'$ в неделю', 'обучаю с нуля', 'доходы для всех', 'по трейдингу', 'фальшивые рубли', '‼️',
'зароботку', 'Набuраю', 'цuфр', 'Обучу вас', 'прuб', 'остались курсы', 'Crypt', 'kр', 'рu'}
NON_GRATA = {"дудар", "хауди", "dudar"}
BAN_WORDS = {"GREEN"}
app/filters.py → src/filters.py
+17 -23
View File
@@ -1,20 +1,23 @@
import logging.handlers
from .models import session
from telebot import types
from .config import WHITEIDS
from .constants import BAN_WORDS, LOG_COMM, SPAM, ALLOWED_WORDS, URL_RX, NON_GRATA, ADMIN_ID
from .models import Chat
from .config import *
from .models import session
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:
from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name
if from_user_id not in WHITEIDS and antispam_is_enabled:
for text in SPAM:
if text.casefold() in message.text.casefold():
logging.info(
"!BAN!" + LOG_COMM % (
message.chat.title, message.from_user.id, message.from_user.first_name, 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:
@@ -24,6 +27,7 @@ def in_caption_spam_list(message: types.Message) -> bool:
logging.info("!BAN!" + LOG_COMM % (
message.chat.title, message.from_user.id, message.from_user.first_name, message.video.file_name))
return True
return False
def in_not_allowed(word_list, msg):
@@ -36,21 +40,17 @@ def in_not_allowed(word_list, msg):
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}"
)
from_user_id, title, from_user_name = message.from_user.id, message.chat.title, message.from_user.first_name
if from_user_id not in WHITEIDS and antispam_is_enabled:
if URL_RX.search(message.text) and in_not_allowed(ALLOWED_WORDS, message.text):
logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: {message.text}")
return True
if message.entities:
for entity in message.entities:
if entity.url and in_not_allowed(ALLOWED_WORDS, entity.url):
logging.info(
f"[DELETE] [{message.chat.title}] [{message.from_user.id}] {message.from_user.first_name}: [entity] {message.text}"
)
logging.info(f"[DELETE] [{title}] [{from_user_id}] {from_user_name}: [entity] {message.text}")
return True
return False
def is_nongrata(type_message: types.Message) -> bool:
@@ -58,6 +58,7 @@ 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:
@@ -66,10 +67,3 @@ def is_admin(message: types.Message) -> bool:
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)
src/helpers.py
+62
View File
@@ -0,0 +1,62 @@
""" Send User's info. """
import logging
import shelve
from pathlib import Path
from urllib import parse
from urllib.request import urlopen
from telebot import types
from .constants import DATA
from .report import reset_report_stats
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:])
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
app/models.py → src/models.py
+8 -6
View File
@@ -1,15 +1,17 @@
import os
from pathlib import Path
from sqlalchemy import Column, String, Integer, Boolean, ForeignKey
from sqlalchemy import create_engine
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'))
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()
app/reminder.py → src/reminder.py
+17 -8
View File
@@ -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
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
if USE_REMINDER:
Thread(target=scheduler).start()
app/report.py → src/report.py
+11 -9
View File
@@ -1,8 +1,10 @@
import html
import os
import random
import shelve
import os
from .config import DATA, ROLLBACK
from .config import ROLLBACK
from .constants import DATA
def create_report_text(chat_id):
@@ -52,12 +54,12 @@ 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']=}"""
src/robot.py
+14
View File
@@ -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,
)
app/rules.py → src/rules.py
+5 -2
View File
@@ -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):
app/stan.py → src/stan.py
+8 -8
View File
@@ -4,11 +4,14 @@ 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 .models import session
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):
@@ -20,7 +23,7 @@ def speak(chance_of, group_id):
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))
@@ -39,8 +42,7 @@ async def add_stan_quote(message: types.Message):
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')
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')
@@ -49,13 +51,11 @@ async def add_stan_quote(message: types.Message):
@bot.message_handler(func=is_white, 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()
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 +65,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
+20
View File
@@ -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
wsgi.py
+6 -1
View File
@@ -1 +1,6 @@
from app.bot import app
import asyncio
from src.webhook import app, bot
if __name__ == '__main__':
asyncio.run(bot.infinity_polling())