409 lines
17 KiB
Python
409 lines
17 KiB
Python
"""
|
||
Ядро PrimoGuard Bot: Инициализация, Управление и Информация
|
||
"""
|
||
from datetime import datetime
|
||
|
||
from aiogram import Bot, Dispatcher
|
||
from aiogram.client.default import DefaultBotProperties
|
||
from aiogram.fsm.storage.memory import MemoryStorage
|
||
from aiogram.types import User, ChatAdministratorRights, BotDescription, BotShortDescription
|
||
from aiogram.utils.i18n import I18n, SimpleI18nMiddleware
|
||
from pymorphy3 import MorphAnalyzer
|
||
|
||
from configs import settings
|
||
from middleware.loggers import logger
|
||
|
||
__all__ = ('bot', 'dp', 'storage', 'i18n', 'morph', 'BotInfo')
|
||
|
||
|
||
# ================= STORAGE И DISPATCHER =================
|
||
|
||
storage = MemoryStorage()
|
||
dp = Dispatcher(storage=storage)
|
||
dp["is_active"] = True
|
||
|
||
|
||
# ================= ИНТЕРНАЦИОНАЛИЗАЦИЯ =================
|
||
|
||
i18n = I18n(path="locales", default_locale="ru", domain="bot")
|
||
i18n_middleware = SimpleI18nMiddleware(i18n=i18n)
|
||
i18n_middleware.setup(dp)
|
||
|
||
|
||
# ================= БОТ =================
|
||
|
||
bot = Bot(
|
||
token=settings.active_bot_token,
|
||
default=DefaultBotProperties(
|
||
parse_mode=settings.PARSE_MODE,
|
||
disable_notification=settings.DISABLE_NOTIFICATION,
|
||
protect_content=settings.PROTECT_CONTENT,
|
||
allow_sending_without_reply=settings.ALLOW_SENDING_WITHOUT_REPLY,
|
||
link_preview_is_disabled=settings.LINK_PREVIEW_IS_DISABLED,
|
||
link_preview_prefer_small_media=settings.LINK_PREVIEW_PREFER_SMALL_MEDIA,
|
||
link_preview_prefer_large_media=settings.LINK_PREVIEW_PREFER_LARGE_MEDIA,
|
||
link_preview_show_above_text=settings.LINK_PREVIEW_SHOW_ABOVE_TEXT,
|
||
show_caption_above_media=settings.SHOW_CAPTION_ABOVE_MEDIA
|
||
)
|
||
)
|
||
|
||
|
||
# ================= МОРФОАНАЛИЗАТОР =================
|
||
|
||
morph = MorphAnalyzer()
|
||
|
||
|
||
# ================= КЛАСС УПРАВЛЕНИЯ БОТОМ =================
|
||
|
||
class BotInfo:
|
||
"""Класс для хранения данных и управления ботом"""
|
||
|
||
# Основные данные бота
|
||
id: int = None
|
||
url: str = None
|
||
first_name: str = None
|
||
last_name: str = None
|
||
username: str = None
|
||
description: str = None
|
||
short_description: str = None
|
||
is_premium: bool = False
|
||
|
||
# Возможности бота
|
||
can_join_groups: bool = False
|
||
can_read_all_group_messages: bool = False
|
||
supports_inline_queries: bool = False
|
||
can_connect_to_business: bool = False
|
||
has_main_web_app: bool = False
|
||
added_to_attachment_menu: bool = False
|
||
|
||
# Данные из конфига
|
||
prefix: str = settings.PREFIX
|
||
started_at: datetime = None
|
||
|
||
@classmethod
|
||
def mention(cls) -> str:
|
||
"""Упоминание бота"""
|
||
return f'@{cls.username}' if cls.username else f'id{cls.id}'
|
||
|
||
@classmethod
|
||
async def webhook(cls, bots: Bot = bot) -> None:
|
||
"""
|
||
Настраивает webhook для бота.
|
||
|
||
Args:
|
||
bots: Объект бота для управления
|
||
"""
|
||
# Только если включен режим webhook
|
||
if not settings.WEBHOOK:
|
||
logger.debug("Режим Webhook отключен (WEBHOOK=False)", log_type='WEBHOOK')
|
||
return
|
||
|
||
# Проверяем наличие URL
|
||
if not settings.WEBHOOK_URL:
|
||
logger.warning(
|
||
"⚠️ WEBHOOK_URL не указан в настройках",
|
||
log_type='WEBHOOK'
|
||
)
|
||
return
|
||
|
||
try:
|
||
logger.info("Настройка вебхука бота", log_type='BOT')
|
||
|
||
# Проверяем текущий webhook
|
||
current_info = await bots.get_webhook_info()
|
||
|
||
# Если уже установлен нужный URL, пропускаем
|
||
if current_info.url == settings.WEBHOOK_URL:
|
||
logger.info(
|
||
f"✓ Вебхук уже установлен: {settings.WEBHOOK_URL}",
|
||
log_type='BOT'
|
||
)
|
||
return
|
||
|
||
# Устанавливаем webhook
|
||
await bots.set_webhook(
|
||
url=settings.WEBHOOK_URL,
|
||
allowed_updates=[
|
||
"message",
|
||
"edited_message",
|
||
"channel_post", # ← ВОТ ЭТО ДОБАВЬ!
|
||
"edited_channel_post", # ← И ЭТО
|
||
"callback_query",
|
||
"inline_query",
|
||
"my_chat_member",
|
||
"chat_member"
|
||
],
|
||
secret_token=settings.SECRET_TOKEN,
|
||
drop_pending_updates=True,
|
||
)
|
||
|
||
logger.success(
|
||
f"✓ Вебхук установлен: {settings.WEBHOOK_URL}",
|
||
log_type='BOT'
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(
|
||
f"❌ Ошибка установки вебхука: {e}",
|
||
log_type='BOT'
|
||
)
|
||
|
||
@classmethod
|
||
async def info(cls, bots: Bot = bot) -> dict:
|
||
"""
|
||
Получает и сохраняет информацию о боте.
|
||
|
||
:param bots: Объект бота для управления
|
||
:return: Словарь с данными о боте
|
||
"""
|
||
logger.info("Получение информации о боте", log_type='BOT')
|
||
|
||
bot_info: User = await bots.get_me()
|
||
|
||
cls.id = bot_info.id
|
||
cls.url = f'tg://user?id={cls.id}'
|
||
cls.first_name = bot_info.first_name
|
||
cls.last_name = bot_info.last_name
|
||
cls.username = bot_info.username
|
||
cls.can_join_groups = getattr(bot_info, 'can_join_groups', False)
|
||
cls.can_read_all_group_messages = getattr(bot_info, 'can_read_all_group_messages', False)
|
||
cls.supports_inline_queries = bot_info.supports_inline_queries or False
|
||
cls.can_connect_to_business = bot_info.can_connect_to_business or False
|
||
cls.has_main_web_app = bot_info.has_main_web_app or False
|
||
cls.added_to_attachment_menu = bot_info.added_to_attachment_menu or False
|
||
cls.started_at = datetime.now()
|
||
|
||
logger.success(f"Информация о боте @{cls.username} получена", log_type='BOT')
|
||
|
||
return {
|
||
'id': cls.id,
|
||
'url': cls.url,
|
||
'first_name': cls.first_name,
|
||
'last_name': cls.last_name,
|
||
'username': cls.username,
|
||
'prefix': cls.prefix,
|
||
'is_premium': cls.is_premium,
|
||
'can_join_groups': cls.can_join_groups,
|
||
'can_read_all_group_messages': cls.can_read_all_group_messages,
|
||
'supports_inline_queries': cls.supports_inline_queries,
|
||
'can_connect_to_business': cls.can_connect_to_business,
|
||
'has_main_web_app': cls.has_main_web_app,
|
||
'added_to_attachment_menu': cls.added_to_attachment_menu,
|
||
}
|
||
|
||
@staticmethod
|
||
async def set_name(bots: Bot = bot, new_name: str = None) -> bool:
|
||
"""Устанавливает имя бота"""
|
||
new_name = new_name or settings.BOT_NAME
|
||
|
||
if not (1 <= len(new_name) <= 64):
|
||
logger.error(f"Имя бота должно быть от 1 до 64 символов (текущее: {len(new_name)})", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
try:
|
||
current_name = (await bots.get_me()).first_name
|
||
|
||
if current_name == new_name:
|
||
logger.debug(f"Имя бота уже установлено: '{current_name}'", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
await bots.set_my_name(new_name)
|
||
logger.success(f"Имя бота изменено: '{current_name}' → '{new_name}'", log_type='BOT_SETUP')
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка установки имени бота: {e}", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
@staticmethod
|
||
async def set_description(bots: Bot = bot, new_description: str = None) -> bool:
|
||
"""Устанавливает полное описание бота"""
|
||
new_description = new_description or settings.BOT_DESCRIPTION
|
||
|
||
if not (0 < len(new_description) <= 512):
|
||
logger.error(f"Описание должно быть от 1 до 512 символов (текущее: {len(new_description)})", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
try:
|
||
current_description: BotDescription = await bots.get_my_description()
|
||
current_text = current_description.description if current_description else ""
|
||
|
||
if current_text == new_description:
|
||
logger.debug("Описание бота уже установлено", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
await bots.set_my_description(description=new_description)
|
||
logger.success("Описание бота обновлено", log_type='BOT_SETUP')
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка установки описания бота: {e}", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
@staticmethod
|
||
async def set_short_description(bots: Bot = bot, new_short: str = None) -> bool:
|
||
"""Устанавливает короткое описание бота"""
|
||
new_short = new_short or settings.BOT_SHORT_DESCRIPTION
|
||
|
||
if not (0 < len(new_short) <= 120):
|
||
logger.error(f"Короткое описание должно быть от 1 до 120 символов (текущее: {len(new_short)})", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
try:
|
||
current_short: BotShortDescription = await bots.get_my_short_description()
|
||
current_text = current_short.short_description if current_short else ""
|
||
|
||
if current_text == new_short:
|
||
logger.debug("Короткое описание бота уже установлено", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
await bots.set_my_short_description(short_description=new_short)
|
||
logger.success("Короткое описание бота обновлено", log_type='BOT_SETUP')
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка установки короткого описания: {e}", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
@staticmethod
|
||
async def set_administrator_rights(bots: Bot = bot, rights: ChatAdministratorRights = None) -> bool:
|
||
"""Устанавливает права администратора по умолчанию"""
|
||
rights = rights or settings.rights
|
||
|
||
try:
|
||
current_rights = await bots.get_my_default_administrator_rights()
|
||
|
||
if current_rights == rights:
|
||
logger.debug("Права администратора уже установлены", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
await bots.set_my_default_administrator_rights(rights)
|
||
logger.success("Права администратора обновлены", log_type='BOT_SETUP')
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка установки прав администратора: {e}", log_type='BOT_SETUP')
|
||
return False
|
||
|
||
@classmethod
|
||
def print(cls, to_console: bool = True, to_file: bool = True) -> str:
|
||
"""
|
||
Красиво форматирует и выводит информацию о боте.
|
||
|
||
:param to_console: Вывести в консоль
|
||
:param to_file: Записать в файлы
|
||
:return: Отформатированная строка
|
||
"""
|
||
# Формирование блоков информации
|
||
header = f"╔═══════════════════════════════════════════════════════════╗"
|
||
title = f"║ 🤖 PRIMOGUARD BOT - ИНФОРМАЦИЯ О ЗАПУСКЕ ║"
|
||
separator = f"╠═══════════════════════════════════════════════════════════╣"
|
||
footer = f"╚═══════════════════════════════════════════════════════════╝"
|
||
|
||
lines = [
|
||
header,
|
||
title,
|
||
separator,
|
||
f"║ ⏰ Время запуска: {cls.started_at.strftime('%d.%m.%Y %H:%M:%S')}",
|
||
f"║",
|
||
f"║ 📋 ОСНОВНАЯ ИНФОРМАЦИЯ:",
|
||
f"║ • Имя: {cls.first_name} {cls.last_name or ''}".ljust(60) + "║",
|
||
f"║ • Username: @{cls.username}".ljust(60) + "║",
|
||
f"║ • ID: {cls.id}".ljust(60) + "║",
|
||
f"║",
|
||
f"║ ⚙️ ВОЗМОЖНОСТИ БОТА:",
|
||
f"║ • Вступать в группы: {'✅' if cls.can_join_groups else '❌'}".ljust(60) + "║",
|
||
f"║ • Читать все сообщения: {'✅' if cls.can_read_all_group_messages else '❌'}".ljust(60) + "║",
|
||
f"║ • Инлайн-запросы: {'✅' if cls.supports_inline_queries else '❌'}".ljust(60) + "║",
|
||
f"║ • Бизнес-аккаунты: {'✅' if cls.can_connect_to_business else '❌'}".ljust(60) + "║",
|
||
f"║ • Веб-приложение: {'✅' if cls.has_main_web_app else '❌'}".ljust(60) + "║",
|
||
f"║ • Меню вложений: {'✅' if cls.added_to_attachment_menu else '❌'}".ljust(60) + "║",
|
||
f"║",
|
||
f"║ 🔧 НАСТРОЙКИ:",
|
||
f"║ • Префикс команд: {cls.prefix}".ljust(60) + "║",
|
||
f"║ • Режим: {'Webhook' if settings.WEBHOOK else 'Polling'}".ljust(60) + "║",
|
||
footer
|
||
]
|
||
|
||
output = '\n'.join(lines)
|
||
|
||
# Вывод в консоль с цветом
|
||
if to_console and settings.START_INFO_CONSOLE:
|
||
colored_output = f"\033[96m{output}\033[0m" # Cyan цвет
|
||
print(colored_output)
|
||
|
||
# Запись в файлы
|
||
if to_file and settings.START_INFO_TO_FILE:
|
||
try:
|
||
settings.LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Полная информация в bot_info.log
|
||
info_file = settings.LOG_DIR / 'bot_info.log'
|
||
with open(info_file, 'w', encoding='utf-8') as f:
|
||
f.write(output)
|
||
|
||
# Краткая запись в историю запусков
|
||
start_file = settings.LOG_DIR / 'bot_starts.log'
|
||
with open(start_file, 'a', encoding='utf-8') as f:
|
||
start_entry = f"{cls.started_at.strftime('%d.%m.%Y %H:%M:%S')} | @{cls.username} | Mode: {'Webhook' if settings.WEBHOOK else 'Polling'}\n"
|
||
f.write(start_entry)
|
||
|
||
logger.debug(f"Информация о боте записана в {info_file}", log_type='BOT_INFO')
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка записи информации в файл: {e}", log_type='BOT_INFO')
|
||
|
||
return output
|
||
|
||
@classmethod
|
||
async def setup(
|
||
cls,
|
||
bots: Bot = bot,
|
||
perm: bool = None,
|
||
setup_webhook: bool = True
|
||
) -> None:
|
||
"""
|
||
Выполняет полную настройку бота.
|
||
|
||
Args:
|
||
bots: Объект бота для управления
|
||
perm: Разрешение на изменения (если None, берется из настроек)
|
||
setup_webhook: Устанавливать ли webhook (по умолчанию True)
|
||
"""
|
||
perm = perm if perm is not None else settings.BOT_EDIT
|
||
|
||
logger.info("🚀 Процесс запуска бота!", log_type='START')
|
||
|
||
# Настройка вебхука (только если разрешено)
|
||
if setup_webhook:
|
||
await cls.webhook(bots=bots)
|
||
|
||
# Получение информации
|
||
await cls.info(bots=bots)
|
||
|
||
# Обновление профиля (если разрешено)
|
||
if perm:
|
||
logger.info("Начало настройки профиля бота...", log_type='BOT_SETUP')
|
||
|
||
results = {
|
||
'name': await cls.set_name(bots=bots),
|
||
'description': await cls.set_description(bots=bots),
|
||
'short_description': await cls.set_short_description(bots=bots),
|
||
'admin_rights': await cls.set_administrator_rights(bots=bots)
|
||
}
|
||
|
||
changed_count = sum(results.values())
|
||
logger.info(
|
||
f"Настройка завершена. Изменено параметров: {changed_count}/4",
|
||
log_type='BOT_SETUP'
|
||
)
|
||
else:
|
||
logger.warning(
|
||
"⚠️ Изменение настроек бота отключено (BOT_EDIT=False)",
|
||
log_type='BOT_SETUP'
|
||
)
|
||
|
||
# Вывод красивой информации
|
||
cls.print()
|
||
|