Update aiogram 3 template

This commit is contained in:
dj
2026-05-29 09:44:39 +03:00
parent e144cb1424
commit 2eced5dc4f
43 changed files with 1877 additions and 1252 deletions
+114 -169
View File
@@ -1,26 +1,35 @@
# - *- coding: utf- 8 - *-
import random
import html
import secrets
import string
import textwrap
import time
import uuid
from datetime import datetime
from typing import Union
from typing import List, Optional, Union
import pytz
from aiogram import Bot
from aiogram.types import (InlineKeyboardButton, KeyboardButton, WebAppInfo, Message, InlineKeyboardMarkup,
ReplyKeyboardMarkup)
from pytz import timezone
from tgbot.data.config import get_admins, BOT_TIMEZONE
from tgbot.utils.misc.bot_logging import bot_logger
#################################### AIOGRAM ###################################
# Generate replay button
# Быстрая сборка реплай-кнопки
def rkb(text: str) -> KeyboardButton:
return KeyboardButton(text=text)
# Generate inline button
def ikb(text: str, data: str = None, url: str = None, switch: str = None, web: str = None) -> InlineKeyboardButton:
# Быстрая сборка инлайн-кнопки
def ikb(
text: str,
data: Optional[str] = None,
url: Optional[str] = None,
switch: Optional[str] = None,
web: Optional[str] = None,
) -> InlineKeyboardButton:
if data is not None:
return InlineKeyboardButton(text=text, callback_data=data)
elif url is not None:
@@ -30,24 +39,24 @@ def ikb(text: str, data: str = None, url: str = None, switch: str = None, web: s
elif web is not None:
return InlineKeyboardButton(text=text, web_app=WebAppInfo(url=web))
else:
raise "Unknown data"
raise ValueError("Не указано действие для инлайн-кнопки")
# Deleting a message with error handling from Telegram
# Удаление сообщения без падения на ошибках Telegram
async def del_message(message: Message):
try:
await message.delete()
except:
...
except Exception:
bot_logger.debug("Не удалось удалить сообщение", exc_info=True)
# Smart messaging (automatic sending of messages with or without photos)
# Отправка текста с фото, если оно передано или обычным сообщением
async def smart_message(
bot: Bot,
user_id: int,
text: str,
keyboard: Union[InlineKeyboardMarkup, ReplyKeyboardMarkup] = None,
photo: Union[str, None] = None,
keyboard: Optional[Union[InlineKeyboardMarkup, ReplyKeyboardMarkup]] = None,
photo: Optional[str] = None,
):
if photo is not None and photo.title() != "None":
await bot.send_photo(
@@ -64,7 +73,7 @@ async def smart_message(
)
# Send a message to all administrators
# Отправка сообщения всем админам
async def send_admins(bot: Bot, text: str, markup=None, not_me=0):
for admin in get_admins():
try:
@@ -75,69 +84,44 @@ async def send_admins(bot: Bot, text: str, markup=None, not_me=0):
reply_markup=markup,
disable_web_page_preview=True,
)
except:
...
except Exception:
bot_logger.warning("Не удалось отправить сообщение админу %s", admin, exc_info=True)
##################################### MISC #####################################
# Removing indents in a multi-line string ("""text""")
################################## РАЗНОЕ ######################################
# Убирает лишние отступы в многострочном тексте
def ded(get_text: str) -> str:
if get_text is not None:
split_text = get_text.split("\n")
if split_text[0] == "": split_text.pop(0)
if split_text[-1] == "": split_text.pop()
save_text = []
for text in split_text:
while text.startswith(" "):
text = text[1:].strip()
save_text.append(text)
get_text = "\n".join(save_text)
else:
get_text = ""
return get_text
return textwrap.dedent(get_text or "").strip()
# Cleaning text of HTML tags ('<b>test</b>' -> *b*test*/b*)
# Чистит HTML-символы, чтобы Telegram не сломал разметку
def clear_html(get_text: str) -> str:
if get_text is not None:
if "</" in get_text: get_text = get_text.replace("<", "*")
if "<" in get_text: get_text = get_text.replace("<", "*")
if ">" in get_text: get_text = get_text.replace(">", "*")
else:
get_text = ""
return get_text
return html.escape(get_text or "", quote=False)
# Cleaning up gaps in the list (['', 1, ' ', 2] -> [1, 2])
# Убирает пустые и мусорные элементы из списка
def clear_list(get_list: list) -> list:
while "" in get_list: get_list.remove("")
while " " in get_list: get_list.remove(" ")
while "." in get_list: get_list.remove(".")
while "," in get_list: get_list.remove(",")
while "\r" in get_list: get_list.remove("\r")
while "\n" in get_list: get_list.remove("\n")
trash = {"", " ", ".", ",", "\r", "\n"}
return get_list
return [value for value in get_list if value not in trash]
# Split the list into several parts ([1, 2, 3, 4] 2 -> [[1, 2], [3, 4]])
def split_list(get_list: list, count: int) -> list[list]:
# Делит список на части нужного размера
def split_list(get_list: list, count: int) -> List[list]:
return [get_list[i:i + count] for i in range(0, len(get_list), count)]
# Get the current date (True - date with time, False - date without time)
# Возвращает текущую дату, при full=True еще и время
def get_date(full: bool = True) -> str:
bot_timezone = timezone(BOT_TIMEZONE)
if full:
return datetime.now(pytz.timezone(BOT_TIMEZONE)).strftime("%d.%m.%Y %H:%M:%S")
return datetime.now(bot_timezone).strftime("%d.%m.%Y %H:%M:%S")
else:
return datetime.now(pytz.timezone(BOT_TIMEZONE)).strftime("%d.%m.%Y")
return datetime.now(bot_timezone).strftime("%d.%m.%Y")
# Get the current Unix time (True - time in nanoseconds, False - time in seconds)
# Возвращает Unix-время: секунды или наносекунды
def get_unix(full: bool = False) -> int:
if full:
return time.time_ns()
@@ -145,89 +129,90 @@ def get_unix(full: bool = False) -> int:
return int(time.time())
# Converting Unix to date and dates to Unix
# Конвертирует дату в Unix и обратно
def convert_date(from_time, full=True, second=True) -> Union[str, int]:
from tgbot.data.config import BOT_TIMEZONE
bot_timezone = timezone(BOT_TIMEZONE)
from_time = str(from_time).strip().replace("-", ".")
if "-" in str(from_time):
from_time = from_time.replace("-", ".")
if str(from_time).isdigit():
if from_time.isdigit():
from_timestamp = int(from_time)
if full:
to_time = datetime.fromtimestamp(from_time, pytz.timezone(BOT_TIMEZONE)).strftime("%d.%m.%Y %H:%M:%S")
to_time = datetime.fromtimestamp(from_timestamp, bot_timezone).strftime("%d.%m.%Y %H:%M:%S")
elif second:
to_time = datetime.fromtimestamp(from_time, pytz.timezone(BOT_TIMEZONE)).strftime("%d.%m.%Y %H:%M")
to_time = datetime.fromtimestamp(from_timestamp, bot_timezone).strftime("%d.%m.%Y %H:%M")
else:
to_time = datetime.fromtimestamp(from_time, pytz.timezone(BOT_TIMEZONE)).strftime("%d.%m.%Y")
to_time = datetime.fromtimestamp(from_timestamp, bot_timezone).strftime("%d.%m.%Y")
else:
if " " in str(from_time):
cache_time = from_time.split(" ")
parts = from_time.split()
if ":" in cache_time[0]:
cache_date = cache_time[1].split(".")
cache_time = cache_time[0].split(":")
else:
cache_date = cache_time[0].split(".")
cache_time = cache_time[1].split(":")
if len(cache_date[0]) == 4:
x_year, x_month, x_day = cache_date[0], cache_date[1], cache_date[2]
else:
x_year, x_month, x_day = cache_date[2], cache_date[1], cache_date[0]
x_hour, x_minute, x_second = cache_time[0], cache_time[1], cache_time[2]
from_time = f"{x_day}.{x_month}.{x_year} {x_hour}:{x_minute}:{x_second}"
if len(parts) == 2 and ":" in parts[0]:
time_part, date_part = parts
elif len(parts) == 2:
date_part, time_part = parts
else:
cache_date = from_time.split(".")
date_part, time_part = from_time, "00:00:00"
if len(cache_date[0]) == 4:
x_year, x_month, x_day = cache_date[0], cache_date[1], cache_date[2]
else:
x_year, x_month, x_day = cache_date[2], cache_date[1], cache_date[0]
date_values = date_part.split(".")
time_values = time_part.split(":")
from_time = f"{x_day}.{x_month}.{x_year}"
if len(time_values) == 2:
time_values.append("0")
if " " in str(from_time):
to_time = int(datetime.strptime(from_time, "%d.%m.%Y %H:%M:%S").timestamp())
if len(date_values[0]) == 4:
x_year, x_month, x_day = date_values[0], date_values[1], date_values[2]
else:
to_time = int(datetime.strptime(from_time, "%d.%m.%Y").timestamp())
x_day, x_month, x_year = date_values[0], date_values[1], date_values[2]
date_time = datetime(
int(x_year),
int(x_month),
int(x_day),
int(time_values[0]),
int(time_values[1]),
int(time_values[2]),
)
date_time = bot_timezone.localize(date_time)
to_time = int(date_time.timestamp())
return to_time
# Generation of a unique ID
# Генерация числового уникального ID
def gen_id(len_id: int = 16) -> int:
mac_address = uuid.getnode()
time_unix = int(str(time.time_ns())[:len_id])
random_int = int(''.join(random.choices('0123456789', k=len_id)))
if len_id <= 0:
raise ValueError("Длина ID должна быть больше нуля")
return mac_address + time_unix + random_int
first_digit = secrets.choice("123456789")
other_digits = "".join(secrets.choice(string.digits) for _ in range(len_id - 1))
return int(f"{first_digit}{other_digits}")
# Password generation | default, number, letter, onechar
# Генерация пароля под разные сценарии
def gen_password(len_password: int = 16, type_password: str = "default") -> str:
char_password = list("1234567890abcdefghigklmnopqrstuvyxwzABCDEFGHIGKLMNOPQRSTUVYXWZ")
if len_password <= 0:
raise ValueError("Длина пароля должна быть больше нуля")
if type_password == "default":
char_password = list("1234567890abcdefghigklmnopqrstuvyxwzABCDEFGHIGKLMNOPQRSTUVYXWZ")
alphabet = string.ascii_letters + string.digits
elif type_password == "letter":
char_password = list("abcdefghigklmnopqrstuvyxwzABCDEFGHIGKLMNOPQRSTUVYXWZ")
alphabet = string.ascii_letters
elif type_password == "number":
char_password = list("1234567890")
alphabet = string.digits
elif type_password == "onechar":
char_password = list("1234567890")
alphabet = string.digits
else:
raise ValueError("Неизвестный тип пароля")
random.shuffle(char_password)
random_chars = "".join([random.choice(char_password) for x in range(len_password)])
random_chars = "".join(secrets.choice(alphabet) for _ in range(len_password))
if type_password == "onechar":
random_chars = f"{random.choice('abcdefghigklmnopqrstuvyxwzABCDEFGHIGKLMNOPQRSTUVYXWZ')}{random_chars[1:]}"
random_chars = f"{secrets.choice(string.ascii_letters)}{random_chars[1:]}"
return random_chars
# Addition to the correct time (1 -> 1 day, 3 -> 3 days)
# Склоняет единицы времени под число
def convert_times(get_time: int, get_type: str = "day") -> str:
get_time = int(get_time)
if get_time < 0: get_time = 0
@@ -255,20 +240,20 @@ def convert_times(get_time: int, get_type: str = "day") -> str:
return f"{get_time} {get_list[count]}"
# Boolean type check
# Приводит строку или число к bool
def is_bool(value: Union[bool, str, int]) -> bool:
value = str(value).lower()
value = str(value).strip().lower()
if value in ('y', 'yes', 't', 'true', 'on', '1'):
return True
elif value in ('n', 'no', 'f', 'false', 'off', '0'):
return False
else:
raise ValueError(f"invalid truth value {value}")
raise ValueError(f"Некорректное bool-значение: {value}")
################################### NUMBERS ####################################
# Converting exponential numbers to a readable form (1e-06 -> 0.000001)
################################### ЧИСЛА ######################################
# Приводит число к читаемой строке без лишних нулей
def snum(amount: Union[int, float], remains: int = 2) -> str:
format_str = "{:." + str(remains) + "f}"
str_amount = format_str.format(float(amount))
@@ -288,38 +273,20 @@ def snum(amount: Union[int, float], remains: int = 2) -> str:
return str(str_amount)
# Convert any number to a real number, removing trailing zeros (remains - rounding)
# Приводит входное значение к int или float
def to_float(get_number, remains: int = 2) -> Union[int, float]:
if "," in str(get_number):
get_number = str(get_number).replace(",", ".")
value = str(get_number).strip().replace(" ", "").replace(",", ".")
number = round(float(value), remains)
if "." in str(get_number):
get_last = str(get_number).split(".")
if number.is_integer():
return int(number)
if str(get_last[1]).endswith("0"):
while True:
if str(get_number).endswith("0"):
get_number = str(get_number)[:-1]
else:
break
get_number = round(float(get_number), remains)
str_number = snum(get_number)
if "." in str_number:
if str_number.split(".")[1] == "0":
get_number = int(get_number)
else:
get_number = float(get_number)
else:
get_number = int(get_number)
return get_number
return number
# Converting a real number to an integer
# Округляет число до int
def to_int(get_number: float) -> int:
if "," in get_number:
if "," in str(get_number):
get_number = str(get_number).replace(",", ".")
get_number = int(round(float(get_number)))
@@ -327,7 +294,7 @@ def to_int(get_number: float) -> int:
return get_number
# Data validation for numbers
# Проверяет, является ли значение числом
def is_number(get_number: Union[str, int, float]) -> bool:
if str(get_number).isdigit():
return True
@@ -337,39 +304,17 @@ def is_number(get_number: Union[str, int, float]) -> bool:
try:
float(get_number)
return True
except ValueError:
except (TypeError, ValueError):
return False
# Converting a number to a readable form (123456789 -> 123,456,789)
# Форматирует число с разделением тысяч
def format_rate(amount: Union[float, int], around: int = 2) -> str:
if "," in str(amount): amount = float(str(amount).replace(",", "."))
if " " in str(amount): amount = float(str(amount).replace(" ", ""))
amount = str(round(amount, around))
value = str(amount).strip().replace(" ", "").replace(",", ".")
number = round(float(value), around)
response = f"{number:,.{around}f}".replace(",", " ")
out_amount, save_remains = [], ""
if "." in amount: save_remains = amount.split(".")[1]
save_amount = [char for char in str(int(float(amount)))]
if len(save_amount) % 3 != 0:
if (len(save_amount) - 1) % 3 == 0:
out_amount.extend([save_amount[0]])
save_amount.pop(0)
elif (len(save_amount) - 2) % 3 == 0:
out_amount.extend([save_amount[0], save_amount[1]])
save_amount.pop(1)
save_amount.pop(0)
else:
print("Error 4388326")
for x, char in enumerate(save_amount):
if x % 3 == 0: out_amount.append(" ")
out_amount.append(char)
response = "".join(out_amount).strip() + "." + save_remains
if response.endswith("."):
response = response[:-1]
if "." in response:
response = response.rstrip("0").rstrip(".")
return response
+15 -12
View File
@@ -2,29 +2,32 @@
from aiogram import Bot
from aiogram.types import BotCommand, BotCommandScopeChat, BotCommandScopeDefault
from tgbot.data.config import get_admins
from tgbot.data.config import BOT_DATABASE_EXPORT, get_admins
from tgbot.utils.misc.bot_logging import bot_logger
# Commands for users
# Команды для обычных пользователей
user_commands = [
BotCommand(command="start", description="♻️ Restart bot"),
BotCommand(command="menu", description="🌀 Get keyboards"),
BotCommand(command="start", description="♻️ Перезапуск бота"),
BotCommand(command="menu", description="🌀 Получение клавиатуры"),
]
# Commands for admins
# Команды для админов
admin_commands = [
BotCommand(command="start", description="♻️ Restart bot"),
BotCommand(command="menu", description="🌀 Get keyboards"),
BotCommand(command="log", description="🖨 Get Logs"),
BotCommand(command="db", description="📦 Get Database"),
BotCommand(command="start", description="♻️ Перезапуск бота"),
BotCommand(command="menu", description="🌀 Получение клавиатуры"),
BotCommand(command="log", description="🖨 Получить логи"),
]
if BOT_DATABASE_EXPORT:
admin_commands.append(BotCommand(command="db", description="📦 Получить БД"))
# Set commands
# Обновление списка команд в Telegram
async def set_commands(bot: Bot):
await bot.set_my_commands(user_commands, scope=BotCommandScopeDefault())
for admin in get_admins():
try:
await bot.set_my_commands(admin_commands, scope=BotCommandScopeChat(chat_id=admin))
except:
...
except Exception:
bot_logger.warning("Не удалось обновить команды для админа %s", admin, exc_info=True)
+22 -6
View File
@@ -1,14 +1,30 @@
# - *- coding: utf- 8 - *-
from typing import Union
from aiogram.filters import BaseFilter
from aiogram.types import Message
from aiogram.types import CallbackQuery, Message
from tgbot.data.config import get_admins
# Filter on admin right
# Проверка, что действия совершает админ
class IsAdmin(BaseFilter):
async def __call__(self, message: Message) -> bool:
if message.from_user.id in get_admins():
async def __call__(self, event: Union[Message, CallbackQuery]) -> bool:
user = getattr(event, "from_user", None)
return bool(user and user.id in get_admins())
# Проверка приватного чата
class IsPrivate(BaseFilter):
async def __call__(self, event: Union[Message, CallbackQuery]) -> bool:
chat = getattr(event, "chat", None)
message = getattr(event, "message", None)
if chat is None and message is not None:
chat = message.chat
if chat is None:
return True
else:
return False
return chat.type == "private"
+49 -23
View File
@@ -1,32 +1,58 @@
# - *- coding: utf- 8 - *-
import logging as bot_logger
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
import colorlog
from tgbot.data.config import PATH_LOGS
# Logging format
log_formatter_file = bot_logger.Formatter("%(levelname)s | %(asctime)s | %(filename)s:%(lineno)d | %(message)s")
log_formatter_console = colorlog.ColoredFormatter(
"%(purple)s%(levelname)s %(blue)s|%(purple)s %(asctime)s %(blue)s|%(purple)s %(filename)s:%(lineno)d %(blue)s|%(purple)s %(message)s%(red)s",
datefmt="%d-%m-%Y %H:%M:%S",
)
LOG_FILE_MAX_BYTES = 5 * 1024 * 1024
LOG_FILE_BACKUP_COUNT = 5
# Logging in file logs.log
file_handler = bot_logger.FileHandler(PATH_LOGS, "w", "utf-8")
file_handler.setFormatter(log_formatter_file)
file_handler.setLevel(bot_logger.INFO)
# Папка под логи создается сама, чтобы бот не падал на старте
log_path = Path(PATH_LOGS)
log_path.parent.mkdir(parents=True, exist_ok=True)
# Logging in console
console_handler = bot_logger.StreamHandler()
console_handler.setFormatter(log_formatter_console)
console_handler.setLevel(bot_logger.CRITICAL)
# Один общий логгер для всего шаблона
bot_logger = logging.getLogger("tgbot")
bot_logger.setLevel(logging.INFO)
bot_logger.propagate = False
# Connect logging settings
bot_logger.basicConfig(
format="%(levelname)s | %(asctime)s | %(filename)s:%(lineno)d | %(message)s",
handlers=[
file_handler,
console_handler
]
)
if not bot_logger.handlers:
# Формат для файла: без цветов
file_formatter = logging.Formatter(
"%(levelname)s | %(asctime)s | %(name)s | %(filename)s:%(lineno)d | %(message)s",
datefmt="%d-%m-%Y %H:%M:%S",
)
# Формат для консоли: коротко и с цветами
console_formatter = colorlog.ColoredFormatter(
"%(log_color)s%(levelname)s%(reset)s | %(blue)s%(asctime)s%(reset)s | "
"%(purple)s%(filename)s:%(lineno)d%(reset)s | %(message)s",
datefmt="%d-%m-%Y %H:%M:%S",
log_colors={
"DEBUG": "cyan",
"INFO": "green",
"WARNING": "yellow",
"ERROR": "red",
"CRITICAL": "bold_red",
},
)
# Не очищаем файл, а крутим по размеру
file_handler = RotatingFileHandler(
log_path,
maxBytes=LOG_FILE_MAX_BYTES,
backupCount=LOG_FILE_BACKUP_COUNT,
encoding="utf-8",
)
file_handler.setFormatter(file_formatter)
file_handler.setLevel(logging.INFO)
# В консоль выводим то, что важно видеть сразу
console_handler = logging.StreamHandler()
console_handler.setFormatter(console_formatter)
console_handler.setLevel(logging.INFO)
bot_logger.addHandler(file_handler)
bot_logger.addHandler(console_handler)
+1
View File
@@ -3,5 +3,6 @@ from aiogram.fsm.context import FSMContext
from tgbot.services.api_session import AsyncRequestSession
# Короткие алиасы для типизации обработчиков
FSM = FSMContext
ARS = AsyncRequestSession
+11 -7
View File
@@ -2,24 +2,28 @@
from aiogram import Bot
from aiogram.types import FSInputFile
from tgbot.data.config import get_admins, PATH_DATABASE, BOT_STATUS_NOTIFICATION
from tgbot.data.config import BOT_DATABASE_EXPORT, BOT_STATUS_NOTIFICATION, PATH_DATABASE, get_admins
from tgbot.utils.const_functions import get_date, send_admins
from tgbot.utils.misc.bot_logging import bot_logger
# Notification after run bot (for all admins)
# Уведомление админам после запуска
async def startup_notify(bot: Bot):
if len(get_admins()) >= 1 and BOT_STATUS_NOTIFICATION:
await send_admins(bot, "<b>✅ Bot was started</b>")
await send_admins(bot, "<b>✅ Бот запущен</b>")
# Autobackup Database
# Автобэкап базы для админов
async def autobackup_admin(bot: Bot):
if not BOT_DATABASE_EXPORT:
return
for admin in get_admins():
try:
await bot.send_document(
admin,
FSInputFile(PATH_DATABASE),
caption=f"<b>📦 #AUTOBACKUP | <code>{get_date()}</code></b>",
caption=f"<b>📦 #АВТОБЭКАП | <code>{get_date()}</code></b>",
)
except:
...
except Exception:
bot_logger.warning("Не удалось отправить автобэкап админу %s", admin, exc_info=True)