"""Утилита для запуска утреннего дайджеста.""" import asyncio import logging import os from dataclasses import dataclass from datetime import datetime, timedelta from typing import Optional import discord from discord.ext import commands from discord.ext import tasks from utils.pogoda import ( API_URL_WEATHER, fetch_weather, format_weather_for_embed, ) from utils.news import fetch_rss, format_articles, RSS_URL_ARTICLES, RSS_URL_POSTS from utils.cat import fetch_cat logger = logging.getLogger(__name__) @dataclass class MorningData: """Собранные данные для утреннего дайджеста.""" weather: Optional[dict] articles: Optional[list] posts: Optional[list] cat_url: Optional[str] async def gather_morning() -> MorningData: """Собрать все данные для утреннего дайджеста параллельно.""" weather_data, articles, posts, cat_url = await asyncio.gather( fetch_weather(API_URL_WEATHER), fetch_rss(RSS_URL_ARTICLES), fetch_rss(RSS_URL_POSTS), fetch_cat(), ) return MorningData( weather=weather_data, articles=articles, posts=posts, cat_url=cat_url, ) async def run_morning(bot: "commands.Bot", channel: discord.TextChannel): """Выполнить утренний дайджест и отправить в канал.""" try: data = await gather_morning() # --- Формируем embed --- embed = discord.Embed(title="🌅 Утренний дайджест!", color=0xF4A460) # Котик как thumbnail if data.cat_url: embed.set_thumbnail(url=data.cat_url) description_lines = [] has_real_data = False # --- Погода --- weather_text = format_weather_for_embed(data.weather) if weather_text: has_real_data = True description_lines.append(weather_text) else: description_lines.append("Не удалось получить данные о погоде.") description_lines.append("") # --- Новости: статьи --- if data.articles is not None: if data.articles: has_real_data = True lines = format_articles(data.articles, "Лучшие статьи за сутки / Искусственный интеллект / Хабr", "https://habr.com/ru/hubs/artificial_intelligence/articles/top/daily/") description_lines.append("\n".join(lines)) else: description_lines.append("Новостей пока нет.") else: description_lines.append("Не удалось получить новости.") description_lines.append("") # --- Новости: посты --- if data.posts is not None: if data.posts: has_real_data = True lines = format_articles(data.posts, "Лучшие новости за сутки / Искусственный интеллект / Хабr", "https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/") description_lines.append("\n".join(lines)) else: description_lines.append("Новостей пока нет.") else: description_lines.append("Не удалось получить новости.") # Fallback для пустых данных if not has_real_data: description_lines = [ "Не удалось получить данные из внешних источников.", "Проверьте доступность API и повторите попытку позже." ] embed.description = "\n".join(description_lines) await channel.send(embed=embed) logger.info("✅ Утренний дайджест отправлен в #%s", channel.name) except Exception as e: logger.error("Ошибка при выполнении утреннего дайджеста: %s", e, exc_info=True) try: await channel.send("❌ Не удалось выполнить утренний дайджест.") except Exception: pass class Scheduler: """Планировщик ежедневных задач.""" def __init__(self, bot: commands.Bot, morning_time: str = "07:00"): self.bot = bot self.morning_time = morning_time self._last_run_date = None # Канал для утреннего дайджеста (по умолчанию None — первый попавшийся) self._target_channel_id: int | None = None channel_id_str = os.getenv("MORNING_CHANNEL_ID") if channel_id_str: try: self._target_channel_id = int(channel_id_str) except ValueError: logger.warning("Неверное значение MORNING_CHANNEL_ID: %s", channel_id_str) self.morning_loop = tasks.loop(seconds=1.0)(self._check_and_run_morning) self._start_scheduler() def _start_scheduler(self): try: self.morning_loop.start() logger.info("Планировщик запущен (время: %s)", self.morning_time) except RuntimeError: logger.warning("Планировщик уже запущен") def _stop_scheduler(self): try: self.morning_loop.stop() logger.info("Планировщик остановлен") except RuntimeError: logger.warning("Планировщик уже остановлен") def _calculate_next_run(self) -> datetime: now = datetime.now() hour, minute = map(int, self.morning_time.split(":")) today_run = now.replace(hour=hour, minute=minute, second=0, microsecond=0) if now >= today_run: return today_run + timedelta(days=1) return today_run async def _check_and_run_morning(self): now = datetime.now() target = self._calculate_next_run() if now >= target and now.day != self._last_run_date: self._last_run_date = now.day await self._run_morning() async def _run_morning(self): logger.info(f"Выполняю morning в {self.morning_time}") # Если задан конкретный канал — отправляем туда if self._target_channel_id: channel = self.bot.get_channel(self._target_channel_id) if isinstance(channel, discord.TextChannel): if channel.permissions_for(channel.guild.me).send_messages: try: await channel.send("🌅 Утренний дайджест!") await run_morning(self.bot, channel) return except Exception as e: logger.error("Ошибка отправки в канал %s: %s", self._target_channel_id, e) return else: logger.warning("Канал с ID %s не найден или не текстовый — fallback", self._target_channel_id) # Fallback: первый канал с правами send_messages for channel in self.bot.get_all_channels(): if isinstance(channel, discord.TextChannel): if channel.permissions_for(channel.guild.me).send_messages: try: await channel.send("🌅 Утренний дайджест!") await run_morning(self.bot, channel) return except Exception as e: logger.error("Ошибка отправки в #%s: %s", channel.name, e) continue def start(self): self._start_scheduler() def stop(self): self._stop_scheduler()