""" Ядро 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()