Messengers.telegram

  1"""
  2Реализация Telegram-бота для отправки уведомлений.
  3
  4Взаимодействует с Telegram Bot API: отправляет сообщения,
  5поддерживает long polling для получения входящих команд,
  6обрабатывает очередь сообщений в отдельном потоке.
  7"""
  8
  9import threading
 10import time
 11import requests
 12
 13from loguru import logger
 14
 15from Messengers.bot import Bot as MessengerBot
 16
 17
 18class TelegramTimeouts:
 19    """Хранит настройки таймаутов для Telegram Bot API.
 20
 21    Позволяет раздельно настроить таймаут для обычных запросов
 22    и для long polling (getUpdates).
 23
 24    Attributes:
 25        request_timeout: Таймаут для обычных запросов (сек).
 26        polling_timeout: Таймаут для long polling (сек).
 27    """
 28
 29    REQUEST_TIMEOUT = 10
 30    POLLING_TIMEOUT = 5
 31
 32    def __init__(
 33        self, request_timeout: int = REQUEST_TIMEOUT, polling_timeout: int = POLLING_TIMEOUT
 34    ):
 35        """Инициализирует настройки таймаутов.
 36
 37        Args:
 38            request_timeout: Таймаут для обычных запросов (сек).
 39            polling_timeout: Таймаут для long polling (сек).
 40        """
 41        self.request_timeout = request_timeout
 42        self.polling_timeout = polling_timeout
 43
 44
 45class TelegramBot(MessengerBot):
 46    """Реализация бота для Telegram Bot API.
 47
 48    Поддерживает отправку сообщений, ответы на команды /start и /chatid,
 49    long polling для получения обновлений, работает в фоновом потоке.
 50
 51    Attributes:
 52        queue: Очередь сообщений для отправки.
 53        token: Токен Telegram-бота.
 54        chat_id: ID чата для отправки уведомлений.
 55        base_url: Базовый URL Telegram Bot API.
 56    """
 57
 58    def __init__(
 59        self,
 60        name: str,
 61        token: str,
 62        chat_id: str,
 63        timeouts: TelegramTimeouts = TelegramTimeouts(),
 64        polling_enabled: bool = False,
 65        polling_interval: int = 1,
 66    ):
 67        """Инициализирует Telegram-бота.
 68
 69        Args:
 70            name: Имя бота.
 71            token: Токен Telegram-бота (полученный от BotFather).
 72            chat_id: ID чата для отправки сообщений.
 73            timeouts: Настройки таймаутов.
 74            polling_enabled: Включить long polling для получения команд.
 75            polling_interval: Интервал между проверками обновлений.
 76        """
 77        super().__init__(name)
 78        self.queue: list[str] = []
 79        self.token = token
 80        self.chat_id = chat_id
 81        self.base_url = f"https://api.telegram.org/bot{self.token}"
 82
 83        self._validate_token()
 84
 85        self.thread = None
 86        self._is_running = False
 87        self._stop_requested = False
 88        self._polling_interval = polling_interval
 89        self._polling_enabled = polling_enabled
 90
 91        self._request_timeout = timeouts.REQUEST_TIMEOUT
 92        self._polling_timeout = timeouts.POLLING_TIMEOUT
 93
 94    def _validate_token(self):
 95        """Проверяет валидность токена Telegram-бота.
 96
 97        Выполняет запрос getMe к Telegram API. В случае неверного
 98        токена выбрасывает ValueError.
 99
100        Raises:
101            ValueError: Если токен недействителен.
102            requests.RequestException: При ошибке соединения.
103        """
104        try:
105            response = requests.get(f"{self.base_url}/getMe", timeout=10)
106            if response.status_code == 200:
107                data = response.json()
108                if data.get("ok"):
109                    logger.debug(f"Токен валиден. Бот: @{data['result']['username']}")
110                    return
111            logger.error(f"Неверный токен: {response.text}")
112            raise ValueError(f"Invalid Telegram token: {response.text}")
113        except requests.RequestException as e:
114            logger.error(f"Ошибка проверки токена: {e}")
115            raise
116
117    def _make_request(self, method: str, params: dict = None, timeout: float = None):
118        """Выполняет запрос к Telegram Bot API.
119
120        Args:
121            method: Метод Telegram API (например, sendMessage, getUpdates).
122            params: Параметры запроса.
123            timeout: Таймаут запроса (если None, используется request_timeout).
124
125        Returns:
126            dict или list: Результат запроса или None при ошибке.
127        """
128        if timeout is None:
129            timeout = self._request_timeout
130
131        try:
132            url = f"{self.base_url}/{method}"
133
134            # Для getUpdates используем более длинный таймаут, но с контролем
135            if method == "getUpdates" and "timeout" in params:
136                # Используем таймаут из параметров, но не более заданного значения
137                actual_timeout = min(params["timeout"] + 5, timeout)  # +5 секунд буфер
138            else:
139                actual_timeout = timeout
140
141            response = requests.post(url, json=params, timeout=actual_timeout)
142
143            if response.status_code == 200:
144                data = response.json()
145                if data.get("ok"):
146                    return data["result"]
147                else:
148                    error_description = data.get("description", "Unknown error")
149                    logger.error(f"Telegram API error in {method}: {error_description}")
150                    return None
151            else:
152                logger.error(f"HTTP error {response.status_code} in {method}: {response.text}")
153                return None
154
155        except requests.exceptions.Timeout:
156            # Таймаут - это нормально для long polling, не логируем как ошибку
157            if method == "getUpdates":
158                logger.debug("getUpdates timeout (expected for long polling)")
159            else:
160                logger.warning(f"Request timeout in {method}")
161            return None
162
163        except requests.exceptions.ConnectionError as e:
164            logger.warning(f"Connection error in {method}: {e}")
165            return None
166
167        except requests.RequestException as e:
168            logger.warning(f"Request error in {method}: {e}")
169            return None
170
171    def send_message(self, text: str):
172        """Отправляет сообщение в чат (публичный метод).
173
174        Добавляет сообщение в очередь и запускает обработку очереди.
175
176        Args:
177            text: Текст сообщения.
178        """
179        self.queue.append(text)
180        self.process_queue()
181
182    def _send_message(self, text: str):
183        """Отправляет сообщение в чат через Telegram API.
184
185        Args:
186            text: Текст сообщения.
187
188        Returns:
189            bool: True если сообщение отправлено успешно, False при ошибке.
190        """
191        if not self._is_running:
192            logger.warning("Бот не запущен, невозможно отправить сообщение")
193            return False
194
195        params = {"chat_id": self.chat_id, "text": text, "parse_mode": "HTML"}
196
197        result = self._make_request("sendMessage", params)
198        if result:
199            logger.debug(f"Сообщение отправлено в {self.chat_id}")
200            return True
201        else:
202            self.queue.append(text)
203            logger.error(f"Не удалось отправить сообщение в {self.chat_id}")
204            return False
205
206    def process_queue(self):
207        """Обрабатывает очередь сообщений для отправки.
208
209        Пытается отправить все сообщения из очереди по порядку.
210        При первой неудачной попытке отправка прекращается.
211        """
212        if not self.queue:
213            return
214
215        logger.debug(f"Попытка отправки {len(self.queue)} сообщений из очереди")
216        for _ in range(len(self.queue)):
217            message = self.queue[0]
218            if self._send_message(message):
219                self.queue.pop(0)
220            else:
221                break  # Прекращаем попытки при первой неудаче
222
223    def _handle_update(self, update):
224        """Обрабатывает входящее обновление от Telegram.
225
226        Поддерживает команды:
227        - /start — подтверждение активности бота
228        - /chatid — получение ID текущего чата
229
230        Args:
231            update: Объект обновления от Telegram API.
232        """
233        try:
234            if "message" in update:
235                message = update["message"]
236                text = message.get("text", "")
237                chat_id = message["chat"]["id"]
238
239                # Обработка команды /start
240                if text == "/start":
241                    self._make_request("sendMessage", {"chat_id": chat_id, "text": "Бот активен!"})
242
243                # Обработка команды /chatid
244                elif text == "/chatid":
245                    self._make_request(
246                        "sendMessage", {"chat_id": chat_id, "text": f"Chat ID: {chat_id}"}
247                    )
248
249        except Exception as e:
250            logger.error(f"Ошибка обработки сообщения: {e}")
251
252    def _get_updates(self, offset: int = None):
253        """Получает обновления от Telegram через getUpdates.
254
255        Args:
256            offset: Идентификатор последнего обработанного update_id + 1.
257
258        Returns:
259            list: Список обновлений или None при ошибке.
260        """
261        params = {
262            "timeout": self._polling_timeout,  # Короткий таймаут для частых проверок
263            "allowed_updates": ["message"],
264        }
265
266        if offset:
267            params["offset"] = offset
268
269        return self._make_request("getUpdates", params, timeout=self._polling_timeout + 5)
270
271    def _run_polling(self):
272        """Запускает основной цикл long polling с устойчивостью к ошибкам.
273
274        При последовательных ошибках (более max_consecutive_errors)
275        делает паузу перед повторными попытками.
276        """
277        logger.debug("Запускаем polling...")
278        offset = None
279        consecutive_errors = 0
280        max_consecutive_errors = 3
281
282        while not self._stop_requested:
283            try:
284                updates = self._get_updates(offset)
285
286                if updates is not None:  # None означает ошибку
287                    if updates:  # Пустой список - нет сообщений, это нормально
288                        for update in updates:
289                            offset = update["update_id"] + 1
290                            self._handle_update(update)
291
292                    # Сбрасываем счетчик ошибок при успешном запросе
293                    consecutive_errors = 0
294
295                    # Короткая пауза после успешного запроса
296                    time.sleep(0.1)
297
298                else:
299                    # Ошибка запроса
300                    consecutive_errors += 1
301                    logger.warning(
302                        f"Ошибка получения updates ({consecutive_errors}/{max_consecutive_errors})"
303                    )
304
305                    if consecutive_errors >= max_consecutive_errors:
306                        logger.error("Слишком много ошибок подряд, пауза...")
307                        time.sleep(5)
308                        consecutive_errors = 0  # Сбрасываем после паузы
309                    else:
310                        time.sleep(1)  # Короткая пауза перед повторной попыткой
311
312            except Exception as e:
313                logger.error(f"Неожиданная ошибка в polling цикле: {e}")
314                time.sleep(5)  # Длинная пауза при критической ошибке
315
316    def _run_bot(self):
317        """Запускает бота и цикл polling (если включён) в текущем потоке.
318
319        Устанавливает флаг _is_running и запускает polling,
320        если polling_enabled = True.
321        """
322        self._is_running = True
323        self._stop_requested = False
324
325        logger.debug("Бот запущен")
326        try:
327            if self._polling_enabled:
328                self._run_polling()
329        except Exception as e:
330            if not self._stop_requested:
331                logger.error(f"Бот упал с ошибкой: {e}")
332        finally:
333            self._is_running = False
334            logger.debug("Бот полностью остановлен")
335
336    def start(self):
337        """Запускает бота в фоновом потоке (daemon).
338
339        Returns:
340            threading.Thread: Поток бота или None, если бот уже запущен.
341        """
342        if self._is_running:
343            logger.warning("Бот уже запущен")
344            return
345
346        self.thread = threading.Thread(target=self._run_bot, daemon=True)
347        self.thread.start()
348
349        # Ждем инициализации
350        time.sleep(2)
351        if self._is_running:
352            logger.info(f'Бот "{self.name}" запущен в фоновом режиме')
353        return self.thread
354
355    def stop(self):
356        """Корректно останавливает бота.
357
358        Устанавливает флаг остановки и ожидает завершения потока
359        до max_wait секунд.
360        """
361        if not self._is_running:
362            logger.warning("Бот уже остановлен")
363            return
364
365        logger.debug("Запрашиваем остановку бота...")
366        self._stop_requested = True
367
368        # Ждем завершения
369        max_wait = 5
370        start_time = time.time()
371
372        while self._is_running and (time.time() - start_time) < max_wait:
373            time.sleep(0.1)
374
375        if self._is_running:
376            logger.warning(f"Бот не остановился в течение {max_wait} секунд")
377        else:
378            logger.info(f'Бот "{self.name}" успешно остановлен')
379
380    def is_running(self):
381        """Проверяет, запущен ли бот.
382
383        Returns:
384            bool: True если бот запущен, False в противном случае.
385        """
386        return self._is_running
387
388    def get_bot_info(self):
389        """Получает информацию о боте через API getMe.
390
391        Returns:
392            dict: Информация о боте или None при ошибке.
393        """
394        return self._make_request("getMe")
395
396    def test_connection(self):
397        """Тестирует соединение с Telegram Bot API.
398
399        Returns:
400            bool: True если соединение установлено, False при ошибке.
401        """
402        info = self.get_bot_info()
403        if info:
404            logger.info(f"Бот подключен: @{info['username']} ({info['first_name']})")
405            return True
406        else:
407            logger.error("Не удалось подключиться к боту")
408            return False