1.2 Вроде все готово

This commit is contained in:
Verum
2025-02-24 00:54:00 +07:00
parent 410b8eff59
commit 61241a2cf4
17 changed files with 385 additions and 171 deletions

View File

@@ -2,15 +2,22 @@
# Определение типа сообщения
from aiogram.types import ContentType
from aiogram.types import Message
# Настройка экспорта из модуля
__all__ = ("types_message",)
# Функция определения типа сообщения
def types_message(message):
def types_message(message: Message) -> str:
"""
Функция для определения типа сообщения на основе его содержимого.
:param message: Сообщение от пользователя.
:return: Описание типа сообщения.
"""
# Словарь для соответствия типов сообщений
content_types = {
content_types: dict = {
ContentType.TEXT: "Текст",
ContentType.PHOTO: "Фото",
ContentType.STICKER: "Стикер",
@@ -61,7 +68,7 @@ def types_message(message):
# Проверка для обычных сообщений
for content_type, description in content_types.items():
if getattr(message, str(content_type.value)):
if getattr(message, str(content_type.value)): # Если тип содержимого найден
return description
# Если сообщение не соответствует ни одному из типов

View File

@@ -4,4 +4,5 @@
# Экспортирование модулей во внешние слои проекта
from .logs import *
from .msg_logger import *
from .custom_loggers import *
from .start_info_out import *

View File

@@ -0,0 +1,69 @@
# BotLibrary/loggers/custom_loggers.py
# Кастомные логгеры для проекта, с более стандартизированным использованием
from loguru import logger
from ..validators import username
from aiogram.types import Message
# Настройка экспорта из модуля
__all__ = ("Logs",)
class Logs:
"""Класс для логирования с разными уровнями через loguru."""
@staticmethod
def debug(text: str = "Логирование!", log_type: str = "Logs", user: str = "Console", message: Message = None) -> None:
"""
Логирует сообщение на уровне DEBUG.
:param text: Сообщение для логирования.
:param log_type: Тип лога (например, "Logs").
:param user: Имя пользователя или источник вызова лога.
:param message: Сообщение от пользователя, если необходимо извлечь имя.
"""
if message:
user = username(message)
logger.bind(log_type=log_type, user=user).debug(text)
@staticmethod
def info(text: str = "Логирование!", log_type: str = "Logs", user: str = "Console", message: Message = None) -> None:
"""
Логирует сообщение на уровне INFO.
:param text: Сообщение для логирования.
:param log_type: Тип лога (например, "Logs").
:param user: Имя пользователя или источник вызова лога.
:param message: Сообщение от пользователя, если необходимо извлечь имя.
"""
if message:
user = username(message)
logger.bind(log_type=log_type, user=user).info(text)
@staticmethod
def warning(text: str = "Логирование!", log_type: str = "Logs", user: str = "Console", message: Message = None) -> None:
"""
Логирует сообщение на уровне WARNING.
:param text: Сообщение для логирования.
:param log_type: Тип лога (например, "Logs").
:param user: Имя пользователя или источник вызова лога.
:param message: Сообщение от пользователя, если необходимо извлечь имя.
"""
if message:
user = username(message)
logger.bind(log_type=log_type, user=user).warning(text)
@staticmethod
def error(text: str = "Логирование!", log_type: str = "Logs", user: str = "Console", message: Message = None) -> None:
"""
Логирует сообщение на уровне ERROR.
:param text: Сообщение для логирования.
:param log_type: Тип лога (например, "Logs").
:param user: Имя пользователя или источник вызова лога.
:param message: Сообщение от пользователя, если необходимо извлечь имя.
"""
if message:
user = username(message)
logger.bind(log_type=log_type, user=user).error(text)

View File

@@ -5,34 +5,55 @@ import sys
from loguru import logger
from ProjectsFiles import BotLogs
# Создание обычного логгера + логгер в файл
async def setup_logger():
async def setup_logger() -> None:
"""
Настройка логгеров для проекта, выводящих логи в консоль.
Логгеры конфигурируются в зависимости от настроек в BotLogs.
Если разрешено логирование, добавляются логи для уровней DEBUG, INFO, WARNING, ERROR.
"""
logger.remove() # Удаляем все логгеры
# Настройка логирования в консоль для каждого уровня
if BotLogs.permission:
logger.add(sys.stderr,
colorize=True,
format=BotLogs.debug_text,
level="DEBUG",
filter=lambda record: record["level"].name == "DEBUG")
logger.add(sys.stderr,
colorize=True,
format=BotLogs.info_text,
level="INFO",
filter=lambda record: record["level"].name == "INFO")
logger.add(sys.stderr,
colorize=True,
format=BotLogs.warning_text,
level="WARNING",
filter=lambda record: record["level"].name == "WARNING")
logger.add(sys.stderr,
colorize=True,
format=BotLogs.error_text,
level="ERROR",
filter=lambda record: record["level"].name == "ERROR")
if BotLogs.permission:
"""logger.add(ProjectPath.log_file,
rotation=BotLogs.max_size,
format=BotLogs.info_text,
backtrace=True,
diagnose=True,
level="INFO",
filter=lambda record: record["level"].name == "INFO")
logger.add(ProjectPath.log_error_file,
rotation=BotLogs.max_size,
format=BotLogs.error_text,
backtrace=True,
diagnose=True,
level="ERROR",
filter=lambda record: record["level"].name == "ERROR")"""
# Добавление логгера для записи в файл (закомментированное, по необходимости активируется)
if BotLogs.permission_to_file:
# Uncomment and adjust if file logging is required
# logger.add(ProjectPath.log_file,
# rotation=BotLogs.max_size,
# format=BotLogs.info_text,
# backtrace=True,
# diagnose=True,
# level="INFO",
# filter=lambda record: record["level"].name == "INFO")
# logger.add(ProjectPath.log_error_file,
# rotation=BotLogs.max_size,
# format=BotLogs.error_text,
# backtrace=True,
# diagnose=True,
# level="ERROR",
# filter=lambda record: record["level"].name == "ERROR")
return

View File

@@ -1,19 +1,29 @@
# BotLibrary/loggers/msg_logger.py
# Логгер для всех не обработанных сообщений
from .logs import logger
from ProjectsFiles import BotLogs
from .custom_loggers import Logs
from ..analytics.type_msg import types_message
from aiogram.types import Message
# Настройка экспорта из модуля
__all__ = ("logger_msg",)
# Создание функции логирования на обычные сообщения
async def logger_msg(message, log_type : str = "Message"):
user = f"@{message.from_user.username or message.from_user.id}"
async def logger_msg(message: Message, log_type: str = "Message") -> None:
"""
Логирует сообщение, если оно не обработано.
:param message: Сообщение от пользователя.
:param log_type: Тип лога (по умолчанию "Message").
"""
# Получаем username или id пользователя
user: str = f"@{message.from_user.username or message.from_user.id}"
# Логирование только если разрешено
if BotLogs.permission:
# Проверка на наличие текста и его типа
if message.text is None:
logger.bind(log_type=log_type, user=user).info(f"Получено сообщение из ({message.chat.id}) : {types_message(message)}")
Logs.info(log_type=log_type, user=user, text=f"Получено сообщение из ({message.chat.id}) : {types_message(message)}")
else:
logger.bind(log_type=log_type, user=user).info(f"Получено сообщение из ({message.chat.id}) : {message.text}")
Logs.info(log_type=log_type, user=user, text=f"Получено сообщение из ({message.chat.id}) : {message.text}")

View File

@@ -3,33 +3,37 @@
from time import sleep
from colorama import Fore
from ProjectsFiles import Permissions
from .logs import logger
from .custom_loggers import Logs
from ..system import BotInfo
# Функция для получения информации о боте и выводе ее в консоль и файл
def bot_info_out():
def bot_info_out() -> str:
"""
Собирает информацию о боте и выводит её в консоль, а также возвращает как строку.
:return: Информация о боте в виде строки.
"""
try:
bot_name = f"Основное имя: {BotInfo.first_name}\n"
bot_post_name = f" Доп. имя: {BotInfo.last_name}\n"
bot_username = f" Юзернейм: @{BotInfo.username}\n"
bot_id = f" ID: {BotInfo.id}\n"
bot_language = f" Языковой код: {BotInfo.language_code}\n"
bot_can_join_groups = f" Может ли вступать в группы: {BotInfo.can_join_groups}\n"
bot_can_read_all_group_messages = f" Чтение всех сообщений: {BotInfo.can_read_all_group_messages}\n"
bot_is_premium = f" Является премиум-ботом: {BotInfo.is_premium}\n"
bot_added_to_attachment_menu = f" Добавлен в меню вложений: {BotInfo.added_to_attachment_menu}\n"
bot_supports_inline_queries = f" Поддерживает инлайн-запросы: {BotInfo.supports_inline_queries}\n"
bot_can_connect_to_business = f" Подключение к бизнес-аккаунтам: {BotInfo.can_connect_to_business}\n"
bot_has_main_web_app = f" Основное веб-приложение: {BotInfo.has_main_web_app}\n"
# Собираем данные о боте
bot_name: str = f"Основное имя: {BotInfo.first_name}\n"
bot_post_name: str = f"Доп. имя: {BotInfo.last_name}\n"
bot_username: str = f"Юзернейм: @{BotInfo.username}\n"
bot_id: str = f"ID: {BotInfo.id}\n"
bot_language: str = f"Языковой код: {BotInfo.language_code}\n"
bot_can_join_groups: str = f"Может ли вступать в группы: {BotInfo.can_join_groups}\n"
bot_can_read_all_group_messages: str = f"Чтение всех сообщений: {BotInfo.can_read_all_group_messages}\n"
bot_is_premium: str = f"Является премиум-ботом: {BotInfo.is_premium}\n"
bot_added_to_attachment_menu: str = f"Добавлен в меню вложений: {BotInfo.added_to_attachment_menu}\n"
bot_supports_inline_queries: str = f"Поддерживает инлайн-запросы: {BotInfo.supports_inline_queries}\n"
bot_can_connect_to_business: str = f"Подключение к бизнес-аккаунтам: {BotInfo.can_connect_to_business}\n"
bot_has_main_web_app: str = f"Основное веб-приложение: {BotInfo.has_main_web_app}\n"
# Формируем полный текст с выводом информации о боте
bot_all_info = (f"{bot_name} {bot_post_name} {bot_username} {bot_id} {bot_language} "
f"{bot_can_join_groups} {bot_can_read_all_group_messages} {bot_is_premium} "
f"{bot_added_to_attachment_menu} {bot_supports_inline_queries} {bot_can_connect_to_business} "
f"{bot_has_main_web_app}")
bot_all_info: str = (f"{bot_name} {bot_post_name} {bot_username} {bot_id} {bot_language} "
f"{bot_can_join_groups} {bot_can_read_all_group_messages} {bot_is_premium} "
f"{bot_added_to_attachment_menu} {bot_supports_inline_queries} {bot_can_connect_to_business} "
f"{bot_has_main_web_app}")
# Печатаем все данные в консоль с задержкой в 1 секунду
sleep(1)
@@ -38,6 +42,5 @@ def bot_info_out():
return bot_all_info
# Проверка на ошибку и ее логирование
except Exception as e:
logger.bind(log_type="INFO", user="Start_INFO").error(f"Ошибка при получении ID пользователя: {e}")
Logs.error(log_type="INFO", user="Start_INFO", text=f"Ошибка при получении ID пользователя: {e}")

View File

@@ -4,13 +4,18 @@
from aiogram.types import ChatAdministratorRights
from ProjectsFiles import BotEdit
from .bots import bot
from ..loggers import logger
from ..loggers import Logs
# Настройка логирования
log_type = "Edit"
# Функция для выполнения всех настроек, если они не совпадают
async def set_all():
async def set_all() -> None:
"""
Выполняет все необходимые настройки бота, если они не совпадают с текущими значениями.
:return: None
"""
await set_adm_rights()
await set_bot_name()
await set_bot_description()
@@ -18,8 +23,12 @@ async def set_all():
# Функция установки прав администратора
async def set_adm_rights():
# Применить права администратора для бота
async def set_adm_rights() -> None:
"""
Устанавливает права администратора для бота, если они отличаются от текущих.
:return: None
"""
rights = ChatAdministratorRights(
is_anonymous=BotEdit.is_anonymous,
can_manage_chat=BotEdit.manage_chat,
@@ -45,15 +54,18 @@ async def set_adm_rights():
# Функция установки имени бота с проверкой на ограничения
async def set_bot_name():
async def set_bot_name() -> None:
"""
Устанавливает имя бота, если оно отличается от текущего и соответствует ограничениям.
:return: None
"""
# Получаем текущее имя бота
current_name = (await bot.get_me()).first_name
# Проверка длины имени
if len(BotEdit.name) < 1 or len(BotEdit.name) > 32:
# Логируем ошибку, если имя не соответствует ограничению
(logger.bind(log_type=log_type, user="NAME_BOT")
.error("Имя бота должно быть от 1 до 32 символов."))
Logs.error(log_type=log_type, user="NAME_BOT", text="Имя бота должно быть от 1 до 32 символов.")
# Проверяем, совпадает ли текущее имя с тем, которое мы хотим установить
if current_name != BotEdit.name:
@@ -61,14 +73,18 @@ async def set_bot_name():
# Функция установки описания бота с проверкой на ограничения
async def set_bot_description():
async def set_bot_description() -> None:
"""
Устанавливает описание бота, если оно отличается от текущего и соответствует ограничениям.
:return: None
"""
# Получаем текущее описание бота
current_description = await bot.get_my_description()
# Проверка длины описания
if len(BotEdit.description) > 255:
(logger.bind(log_type=log_type, user="DISCRIPT")
.error("Короткое описание бота не может превышать 255 символов."))
Logs.error(log_type=log_type, user="DISCRIPT", text="Короткое описание бота не может превышать 255 символов.")
# Проверяем, совпадает ли текущее описание с тем, которое мы хотим установить
if current_description != BotEdit.description:
@@ -76,14 +92,18 @@ async def set_bot_description():
# Функция установки короткого описания бота с проверкой на ограничения
async def set_bot_short_description():
async def set_bot_short_description() -> None:
"""
Устанавливает короткое описание бота, если оно отличается от текущего и соответствует ограничениям.
:return: None
"""
# Получаем текущее короткое описание бота
current_short_description = await bot.get_my_short_description()
# Проверка длины короткого описания
if len(BotEdit.short_description) > 512:
(logger.bind(log_type=log_type, user="SHORT_DISCRIPT")
.error("Описание виджета не может превышать 512 символов."))
Logs.error(log_type=log_type, user="SHORT_DISCRIPT", text="Описание виджета не может превышать 512 символов.")
# Проверяем, совпадает ли текущее короткое описание с тем, которое мы хотим установить
if current_short_description != BotEdit.short_description:

View File

@@ -5,17 +5,18 @@ from aiogram import Dispatcher, Bot, F
from aiogram.enums import ParseMode
from aiogram.client.default import DefaultBotProperties
from aiogram.utils.keyboard import InlineKeyboardBuilder, ReplyKeyboardBuilder
from aiogram.types import Bot as BotType
from ..timer import *
from ProjectsFiles import *
from ..timer import get_host_time, get_moscow_time
from ProjectsFiles import bot_token, BotVar
# Создание экземпляра диспатчера, строителей кнопок
dp = Dispatcher()
# Создание строителей кнопок
rkb = ReplyKeyboardBuilder()
ikb = InlineKeyboardBuilder()
# Настройка параметров диспатчера
dp = Dispatcher()
dp["started_at"] = get_host_time()
dp["started_at_msk"] = get_moscow_time()
dp["is_active"] = True # Флаг активности бота
@@ -32,8 +33,7 @@ dp["handlers"] = {"on_message": [], "on_error": []}
dp["storage"] = {}
dp["database"] = None
# Объявление экземпляров и переменных
# Настройки для бота
bot_properties = DefaultBotProperties(
parse_mode=ParseMode.HTML, # Устанавливаем формат HTML для всех сообщений
disable_notification=True, # Отключаем уведомления при отправке сообщений
@@ -42,39 +42,49 @@ bot_properties = DefaultBotProperties(
link_preview_is_disabled=True, # Отключаем предварительный просмотр ссылок
show_caption_above_media=False, # Показываем подпись выше медиа
)
bot = Bot(token=bot_token, default=bot_properties) # Объявление бота
F_Media = F.photo | F.files | F.video | F.animation | F.voice | F.video_note # Фильтр-медиа
F_All = F.text | F.photo | F.files | F.video | F.animation | F.voice | F.video_note # Фильтр на все
# Создание экземпляра бота
bot = Bot(token=bot_token, default=bot_properties) # Объявление бота
# Класс для хранения данных о боте (некоторые переменные даны как шаблон)
# Фильтры для различных типов сообщений
F_Media = F.photo | F.files | F.video | F.animation | F.voice | F.video_note # Фильтр для медиа
F_All = F.text | F.photo | F.files | F.video | F.animation | F.voice | F.video_note # Фильтр на все
# Класс для хранения данных о боте
class BotInfo:
"""
Класс для хранения данных о боте и их обновления.
"""
# Статические переменные для хранения данных
id = None
first_name = None
last_name = None
username = None
description = None
short_description = None
can_join_groups = None
can_read_all_group_messages = None
language_code = BotVar.language
prefix = BotVar.prefix
is_premium = None
added_to_attachment_menu = None
supports_inline_queries = None
can_connect_to_business = None
has_main_web_app = None
id: int = None
first_name: str = None
last_name: str = None
username: str = None
description: str = None
short_description: str = None
can_join_groups: bool = None
can_read_all_group_messages: bool = None
language_code: str = BotVar.language
prefix: str = BotVar.prefix
is_premium: bool = None
added_to_attachment_menu: bool = None
supports_inline_queries: bool = None
can_connect_to_business: bool = None
has_main_web_app: bool = None
# Метод для обновления данных
@classmethod
def update(cls, bot_info):
def update(cls, bot_info) -> None:
"""
Обновляет данные о боте.
:param bot_info: Объект с данными о боте, полученные через API Telegram.
"""
cls.id = bot_info.id
cls.first_name = bot_info.first_name
cls.last_name = bot_info.last_name
cls.username = bot_info.username
cls.description = getattr(bot_info, 'description', '')
cls.short_description = getattr(bot_info, 'description', '')
cls.short_description = getattr(bot_info, 'short_description', '')
cls.language_code = bot_info.language_code
cls.is_premium = bot_info.is_premium
cls.added_to_attachment_menu = bot_info.added_to_attachment_menu
@@ -85,15 +95,19 @@ class BotInfo:
cls.can_read_all_group_messages = getattr(bot_info, 'can_read_all_group_messages', None)
# Функция получения данных о боте
async def bot_get_info():
# Получение информации о боте
async def bot_get_info() -> dict:
"""
Получает информацию о боте и обновляет данные в классе BotInfo.
:return: Словарь с данными о боте.
"""
# Получение информации о боте через API
bot_info_data = await bot.get_me()
# Обновляем данные о боте в BotInfo
BotInfo.update(bot_info_data)
# Возвращаем обновленные данные
# Возвращаем обновленные данные о боте
return {
'bot_info': bot_info_data,
'id': bot_info_data.id,
@@ -101,7 +115,7 @@ async def bot_get_info():
'last_name': bot_info_data.last_name,
'username': bot_info_data.username,
'description': getattr(bot_info_data, 'description', ''),
'short_description': getattr(bot_info_data, 'description', ''),
'short_description': getattr(bot_info_data, 'short_description', ''),
'language_code': bot_info_data.language_code,
'prefix': BotVar.prefix,
'is_premium': bot_info_data.is_premium,

View File

@@ -3,12 +3,19 @@
import os
from ProjectsFiles import ProjectPath, TypeDirectory
from typing import List
# Настройка экспорта из модуля
__all__ = ("create_directories", "setup_directories")
# Функция создания пустых директорий
async def create_directories(base_directory, subdirectories):
async def create_directories(base_directory: str, subdirectories: List[str]) -> None:
"""
Создает указанные поддиректории в указанной базовой директории, если они еще не существуют.
:param base_directory: Путь к базовой директории.
:param subdirectories: Список поддиректорий, которые необходимо создать.
"""
# Создание директорий и файлов в каждой из них
for subdirectory in subdirectories:
directory_path = os.path.join(base_directory, subdirectory)
@@ -16,11 +23,17 @@ async def create_directories(base_directory, subdirectories):
# Проверка, существует ли директория, если нет - создаём
if not os.path.exists(directory_path):
os.makedirs(directory_path)
print(f"Создана директория: {directory_path}")
# Начальная установка пустых директорий
async def setup_directories():
async def setup_directories() -> None:
"""
Настройка начальных пустых директорий для проекта.
"""
# Создание директорий для медиа файлов
await create_directories(ProjectPath.personal_media, TypeDirectory.media_directories)
# Раскомментируйте следующие строки, если необходимо создать другие директории
# await create_directories(ProjectPath.received_media, TypeDirectory.media_directories)
# await create_directories(ProjectPath.bot_files, TypeDirectory.avatar_directories)
# await create_directories(ProjectPath.msg, TypeDirectory.msg_directories)
# await create_directories(ProjectPath.msg, TypeDirectory.msg_directories)

View File

@@ -1,5 +1,5 @@
# BotLibrary/timer/start_time.py
# Получение времени по
# Получение времени по разным часовым поясам
import pytz
from datetime import datetime
@@ -7,18 +7,31 @@ from tzlocal import get_localzone
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from ProjectsFiles import BotVar
# Создание планировщика
# Создание планировщика для работы с задачами по времени
scheduler = AsyncIOScheduler(timezone=get_localzone().key)
# Функция получение времени по Московскому времени
def get_moscow_time():
def get_moscow_time() -> str:
"""
Получение текущего времени по московскому времени.
:return: Строка, представляющая время в формате, заданном в BotVar.time_format.
"""
# Устанавливаем временную зону для Москвы
moscow_tz = pytz.timezone('Europe/Moscow')
# Получаем текущее время по московскому времени
moscow_time = datetime.now(moscow_tz)
# Возвращаем строку с форматом времени
return moscow_time.strftime(BotVar.time_format)
# Функция получение времени хоста
def get_host_time():
def get_host_time() -> str:
"""
Получение текущего времени хоста (локального времени).
:return: Строка, представляющая локальное время в формате, заданном в BotVar.time_format.
"""
# Получаем текущее время на хосте
host_time = datetime.now()
# Возвращаем строку с форматом времени
return host_time.strftime(BotVar.time_format)

View File

@@ -2,15 +2,23 @@
# Создание валидации почты для проекта
from email_validator import validate_email, EmailNotValidError
from typing import Optional
# Настройка экспорта из этого модуля
__all__ = ("valid_email",)
# Функция проверки почты на корректность
def valid_email(text: str) -> str | None:
def valid_email(text: str) -> Optional[str]:
"""
Проверяет корректность почтового адреса.
:param text: Почтовый адрес в виде строки.
:return: Нормализованный почтовый адрес, если он валиден, иначе None.
"""
try:
# Проверка и нормализация email
email = validate_email(text)
return email.normalized
except EmailNotValidError:
# Если email невалиден, можно добавить логирование или обработку ошибок
return None
return email.normalized

View File

@@ -1,11 +1,19 @@
# BotLibrary/validators/username.py
# Получение юзера пользователя
# Получение юзера или ID пользователя
from aiogram.types import message
from aiogram.types import Message
# Настройка экспорта из модуля
__all__ = ("username",)
# Функция получения юзера или id пользователя
def username(message: message.Message):
return f"@{message.from_user.username or message.from_user.id}"
# Функция получения юзера или ID пользователя
def username(message: Message) -> str:
"""
Возвращает юзернейм пользователя из сообщения, или ID, если юзернейм не указан.
:param message: Объект сообщения из aiogram.
:return: Строка с юзернеймом пользователя или его ID.
"""
if message.from_user:
return f"@{message.from_user.username}" if message.from_user.username else str(message.from_user.id)
return "@Unknown_User" # Если from_user отсутствует