import asyncio import inspect import logging import os import sys import threading from typing import TYPE_CHECKING import discord from discord.ext import commands from discord.ext.commands import CommandNotFound from dotenv import load_dotenv from commands import ALL_COMMANDS from console_commands import ALL_CONSOLE_COMMANDS from utils.morning_runner import Scheduler if TYPE_CHECKING: from utils.morning_runner import Scheduler as SchedulerType logger = logging.getLogger(__name__) load_dotenv() intents = discord.Intents.default() intents.message_content = True class BotRunner: """Управляет жизненным циклом бота.""" def __init__(self) -> None: import time self.bot = commands.Bot(command_prefix="!", intents=intents) self.bot._start_time = time.time() self.stop_event = threading.Event() self.bot_ready = threading.Event() self.scheduler: SchedulerType | None = None self._setup_events() def _setup_events(self) -> None: """Настроить обработчики событий бота.""" @self.bot.event async def on_ready() -> None: logger.info("Бот вошёл как %s", self.bot.user) for cog_class in ALL_COMMANDS: cog = cog_class() await self.bot.add_cog(cog) for cog in self.bot.cogs: logger.info(" Загружен: %s", cog) self.bot_ready.set() @self.bot.event async def on_guild_available(guild: discord.Guild) -> None: """Запуск планировщика после загрузки кэша сервера. on_ready срабатывает до полной загрузки guild-кэша, из-за чего get_channel() возвращает None. on_guild_available гарантирует, что данные сервера в кэше. """ # Запускаем планировщик только один раз if self.scheduler is not None: return morning_time = os.getenv("MORNING_TIME", "07:00") self.scheduler = Scheduler(self.bot, morning_time) self.bot._scheduler = self.scheduler logger.info("Планировщик запущен (время: %s, сервер: %s)", morning_time, guild.name) @self.bot.event async def on_command_error(ctx: commands.Context, error: Exception) -> None: if isinstance(error, CommandNotFound): return # Терминал — детали для разработчика cmd_name = ctx.command.name if ctx and ctx.command else "?" logger.error( f"Ошибка команды {cmd_name}: {error}", exc_info=True, ) # Discord — только если команда не ответила сама # ctx.interaction есть только у slash-команд (AutoshardedInteractionContext) # Для текстовых команд (!prefix) атрибута нет — используем hasattr if ( ctx and hasattr(ctx, "interaction") and ctx.interaction and ctx.interaction.response.is_done() ): return try: await ctx.send("Не удалось выполнить команду. Попробуйте позже.") except (discord.NotFound, discord.Forbidden): pass # Бот не может писать в канал — игнорируем @self.bot.command(name="msg") async def msg(ctx: commands.Context, *, text: str) -> None: """Повторяет текст после !msg""" await ctx.send(text) def _print_commands(self) -> None: """Вывести список доступных консольных команд.""" available = {k: v for k, v in ALL_CONSOLE_COMMANDS.items() if k != "stop"} print("\nДоступные команды:") for idx, (name, func) in enumerate(available.items(), 1): print(f" {idx}. {name}") print(" 0. stop") def console_input(self) -> None: """Обработка ввода команд из консоли.""" logger.info("Консольный режим ввода запущен") self.bot_ready.wait() self._print_commands() while not self.stop_event.is_set(): try: choice = input("\nВыберите команду (номер): ").strip() if choice == "0": logger.info("Пользователь выбрал команду stop через консоль") print("\nОстановка бота...") self.stop_event.set() asyncio.run_coroutine_threadsafe( self.bot.close(), self.bot.loop ).result(timeout=5) break try: available = { k: v for k, v in ALL_CONSOLE_COMMANDS.items() if k != "stop" } idx = int(choice) if 0 < idx <= len(available): cmd_name = list(available.keys())[idx - 1] cmd_func = ALL_CONSOLE_COMMANDS[cmd_name] logger.info("Выполняется консольная команда: %s", cmd_name) if inspect.iscoroutinefunction(cmd_func): asyncio.run_coroutine_threadsafe( cmd_func(self.stop_event, self.bot), self.bot.loop ).result() else: cmd_func(self.stop_event, self.bot) else: logger.warning("Неизвестная консольная команда: %s", choice) print(f"Неизвестная команда: {choice}") except (ValueError, IndexError): logger.warning("Неверный формат ввода консоли: %s", choice) print(f"Неверный формат: {choice}") self._print_commands() except (EOFError, KeyboardInterrupt): logger.info("Консольный ввод завершен (EOF/KeyboardInterrupt)") self.stop_event.set() try: asyncio.run_coroutine_threadsafe( self.bot.close(), self.bot.loop ).result(timeout=5) except Exception as e: logger.error("Ошибка при остановке бота: %s", e) break def run(self, token: str) -> None: """Запустить бота.""" logger.info("Запуск бота...") try: self.bot.run(token) except discord.LoginFailure as e: logger.critical("Ошибка авторизации бота: %s", e, exc_info=True) logger.error("Токен неверный или бот отключён. Код ошибки: %s", e) sys.exit(1) except discord.HTTPException as e: logger.critical( "HTTP ошибка при подключении к Discord: %s", e, exc_info=True ) logger.error( "Сбой соединения с Discord API. Проверьте доступность сервиса." ) sys.exit(1) except Exception as e: logger.critical( "Непредвиденная ошибка при запуске бота: %s", e, exc_info=True ) logger.error("Критическая ошибка при запуске. Код ошибки: %s", type(e).__name__) sys.exit(1) except KeyboardInterrupt: logger.info("Получен сигнал KeyboardInterrupt") self.stop_event.set() if self.scheduler: self.scheduler.stop() asyncio.run_coroutine_threadsafe( self.bot.close(), self.bot.loop ).result() sys.exit(0) def _validate_config() -> None: """Проверить конфигурацию при запуске.""" logger.info("Проверка конфигурации...") token = os.getenv("DISCORD_TOKEN") if not token: logger.error("Токен Discord не найден в .env") sys.exit(1) morning_time = os.getenv("MORNING_TIME", "07:00") try: hour, minute = map(int, morning_time.split(":")) if not (0 <= hour <= 23 and 0 <= minute <= 59): raise ValueError except (ValueError, AttributeError): logger.error( "Неверный формат MORNING_TIME: %s (ожидается ЧЧ:ММ)", morning_time ) sys.exit(1) channel_id = os.getenv("MORNING_CHANNEL_ID") if channel_id: try: int(channel_id) except ValueError: logger.error( "Неверное значение MORNING_CHANNEL_ID: %s (ожидается число)", channel_id, ) sys.exit(1) logger.info("Конфигурация проверена успешно") if __name__ == "__main__": from utils.logger import setup_logging logger.info("=== Запуск Discord бота ===") setup_logging() _validate_config() runner = BotRunner() # Консольный ввод работает только в интерактивном терминале # В Docker stdin недоступен — пропускаем консольный режим if sys.stdin.isatty(): logger.info("Введите 'stop' для остановки бота") thread = threading.Thread(target=runner.console_input, daemon=True) thread.start() else: logger.info("Консольный режим отключен (stdin не интерактивный)") token = os.getenv("DISCORD_TOKEN") runner.run(token)