""" Абстракция канала уведомлений. Паттерн «Адаптер»: весь код бота работает с интерфейсом Notifier, не зная о конкретном мессенджере. Чтобы добавить новый канал (Slack, Email, Teams и т.д.) — создай класс, реализующий Notifier, и зарегистрируй его через poster.set_notifier(). Структура клавиатуры (Keyboard) тоже агностична — адаптеры конвертируют её в нативный формат своего мессенджера. """ from __future__ import annotations import logging from abc import ABC, abstractmethod from dataclasses import dataclass log = logging.getLogger(__name__) # ── Агностичное представление клавиатуры ────────────────────────────────────── @dataclass class KeyboardButton: """Одна кнопка: либо URL-ссылка, либо callback.""" label: str url: str | None = None callback_data: str | None = None # Клавиатура — список строк, каждая строка — список кнопок Keyboard = list[list[KeyboardButton]] # ── Интерфейс отправителя ────────────────────────────────────────────────────── class Notifier(ABC): """ Абстрактный адаптер канала уведомлений. Реализации: - TelegramChannelNotifier (этот файл) - SlackNotifier (создай bot/adapters/slack.py) - EmailNotifier (создай bot/adapters/email.py) - ... """ @abstractmethod def send(self, text: str, keyboard: Keyboard | None = None) -> int: """Отправить сообщение. Возвращает идентификатор сообщения.""" @abstractmethod def edit_text(self, message_id: int, text: str, keyboard: Keyboard | None = None): """Редактировать текст существующего сообщения.""" @abstractmethod def edit_keyboard(self, message_id: int, keyboard: Keyboard | None): """Заменить только клавиатуру существующего сообщения.""" @abstractmethod def delete(self, message_id: int): """Удалить сообщение (ошибки поглощаются — сообщение могло быть удалено).""" @abstractmethod def send_animation(self, url: str, disable_notification: bool = True) -> int: """Отправить анимацию/GIF. Возвращает идентификатор сообщения.""" def is_stale_error(self, exc: Exception) -> bool: """ True если ошибка означает устаревшее/уже удалённое сообщение. Адаптеры переопределяют под свою платформу. """ return False # ── Telegram-адаптер ────────────────────────────────────────────────────────── class TelegramChannelNotifier(Notifier): """Адаптер для отправки уведомлений в Telegram-канал через pytelegrambotapi.""" def __init__(self, bot, channel_id: int): self._bot = bot self._channel_id = channel_id # ── Конвертация агностичной Keyboard → telebot InlineKeyboardMarkup ──────── def _to_markup(self, keyboard: Keyboard | None): from telebot import types if not keyboard: return None markup = types.InlineKeyboardMarkup() for row in keyboard: tg_row = [] for btn in row: if btn.url: tg_row.append(types.InlineKeyboardButton(btn.label, url=btn.url)) elif btn.callback_data: tg_row.append(types.InlineKeyboardButton(btn.label, callback_data=btn.callback_data)) if tg_row: markup.row(*tg_row) return markup # ── Реализация интерфейса ────────────────────────────────────────────────── def send(self, text: str, keyboard: Keyboard | None = None) -> int: log.info("Telegram send_message → channel %s", self._channel_id) msg = self._bot.send_message( self._channel_id, text, parse_mode="Markdown", disable_web_page_preview=True, reply_markup=self._to_markup(keyboard), ) log.info("Telegram send_message OK, message_id=%s", msg.message_id) return msg.message_id def edit_text(self, message_id: int, text: str, keyboard: Keyboard | None = None): log.info("Telegram edit_message_text → message_id=%s", message_id) self._bot.edit_message_text( text, self._channel_id, message_id, parse_mode="Markdown", disable_web_page_preview=True, reply_markup=self._to_markup(keyboard), ) log.info("Telegram edit_message_text OK, message_id=%s", message_id) def edit_keyboard(self, message_id: int, keyboard: Keyboard | None): log.info("Telegram edit_message_reply_markup → message_id=%s", message_id) self._bot.edit_message_reply_markup( self._channel_id, message_id, reply_markup=self._to_markup(keyboard), ) def delete(self, message_id: int): log.info("Telegram delete_message → message_id=%s", message_id) try: self._bot.delete_message(self._channel_id, message_id) log.info("Telegram delete_message OK, message_id=%s", message_id) except Exception as e: log.warning("Telegram delete_message failed (message_id=%s): %s", message_id, e) def send_animation(self, url: str, disable_notification: bool = True) -> int: log.info("Telegram send_animation → channel %s", self._channel_id) msg = self._bot.send_animation( self._channel_id, url, disable_notification=disable_notification, ) log.info("Telegram send_animation OK, message_id=%s", msg.message_id) return msg.message_id def is_stale_error(self, exc: Exception) -> bool: s = str(exc).lower() return any(x in s for x in ( "message to edit not found", "message is not modified", "message can't be edited", "bad request", ))