Messengers.telegram

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