""" Envío de avisos a Telegram. Las llamadas HTTP nunca bloquean los bucles de detección: se encolan y un hilo aparte las envía por orden. Si la red falla, reintenta con retroceso exponencial; si la cola se llena (Telegram caído mucho rato), se descartan los avisos viejos en favor de los nuevos, porque un aviso de hace cinco minutos ya no sirve. El token se trata como secreto: jamás se imprime, y todo texto de error pasa por el enmascarador (la URL de la API lo lleva dentro). """ import json import queue import threading import time import requests from pet_common import mask_secrets API = "https://api.telegram.org/bot%s/%s" def descubrir_chat_id(token, timeout=(5, 15)): """ Deduce el chat_id a partir del último mensaje recibido por el bot. Evita tener que buscarlo a mano: basta con escribirle algo al bot y lanzar esto. Devuelve (chat_id, descripcion) o (None, motivo). """ try: respuesta = requests.get(API % (token, "getUpdates"), timeout=timeout) respuesta.raise_for_status() datos = respuesta.json() except Exception as exc: # noqa: BLE001 return None, mask_secrets(exc, [token]) if not datos.get("ok"): return None, mask_secrets(datos.get("description", "respuesta sin ok"), [token]) for actualizacion in reversed(datos.get("result", [])): mensaje = actualizacion.get("message") or actualizacion.get("edited_message") or {} chat = mensaje.get("chat") or {} if chat.get("id") is not None: nombre = chat.get("title") or chat.get("first_name") or chat.get("username") or "?" return chat["id"], nombre return None, "el bot no ha recibido ningún mensaje todavía" class TelegramNotifier: """Cola de avisos con un hilo que los envía a Telegram.""" def __init__( self, token, chat_id=None, queue_size=16, connect_timeout=5, read_timeout=30, max_retries=3, ): self._token = token self._chat_id = chat_id self._timeout = (connect_timeout, read_timeout) self._max_retries = max_retries self._queue = queue.Queue(maxsize=queue_size) self._thread = None self.sent = 0 self.errors = 0 self.dropped = 0 self.last_error = None # --- ciclo de vida --------------------------------------------------- def start(self): """Comprueba la configuración y arranca el hilo de envío.""" if self._chat_id is None: chat_id, detalle = descubrir_chat_id(self._token, self._timeout) if chat_id is None: raise RuntimeError(f"no se pudo deducir el chat_id: {detalle}") self._chat_id = chat_id print(f"[INFO] Telegram: chat detectado ({detalle}, id {chat_id}).") self._thread = threading.Thread(target=self._loop, name="telegram", daemon=True) self._thread.start() return self def stop(self, timeout=15.0): """Deja de aceptar avisos y espera a que se envíe lo que queda en cola.""" self._queue.put(None) if self._thread is not None: self._thread.join(timeout=timeout) # --- API pública ----------------------------------------------------- def send_message(self, texto): return self._encola(("message", None, texto)) def send_photo(self, jpeg, caption=""): return self._encola(("photo", jpeg, caption)) # --- interno --------------------------------------------------------- def _encola(self, item): try: self._queue.put_nowait(item) return True except queue.Full: # Nos quedamos con el aviso nuevo y tiramos el más viejo try: self._queue.get_nowait() self._queue.task_done() self.dropped += 1 except queue.Empty: pass try: self._queue.put_nowait(item) except queue.Full: self.dropped += 1 return False return True def _loop(self): while True: item = self._queue.get() try: if item is None: return self._envia(item) finally: self._queue.task_done() def _envia(self, item): tipo, datos, texto = item url = API % (self._token, "sendPhoto" if tipo == "photo" else "sendMessage") for intento in range(1, self._max_retries + 1): try: if tipo == "photo": respuesta = requests.post( url, data={"chat_id": self._chat_id, "caption": texto}, files={"photo": ("captura.jpg", datos, "image/jpeg")}, timeout=self._timeout, ) else: respuesta = requests.post( url, data={"chat_id": self._chat_id, "text": texto}, timeout=self._timeout, ) # Límite de velocidad: Telegram dice cuánto hay que esperar if respuesta.status_code == 429: espera = self._retry_after(respuesta) print(f"[WARN] Telegram limitó la frecuencia; esperando {espera:.0f}s", flush=True) time.sleep(espera) continue respuesta.raise_for_status() cuerpo = respuesta.json() if not cuerpo.get("ok"): raise RuntimeError(cuerpo.get("description", "respuesta sin ok")) self.sent += 1 return True except Exception as exc: # noqa: BLE001 self.last_error = mask_secrets(exc, [self._token]) if intento < self._max_retries: time.sleep(min(2 ** intento, 10)) self.errors += 1 print(f"[WARN] No se pudo enviar el aviso a Telegram: {self.last_error}", flush=True) return False @staticmethod def _retry_after(respuesta): """Segundos que pide Telegram ante un 429 (por defecto, 5).""" try: parametros = respuesta.json().get("parameters", {}) return float(parametros.get("retry_after", 5)) except (ValueError, json.JSONDecodeError): return 5.0