158 lines
6.9 KiB
Python
158 lines
6.9 KiB
Python
"""
|
||
Абстракция канала уведомлений.
|
||
|
||
Паттерн «Адаптер»: весь код бота работает с интерфейсом 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",
|
||
))
|