# -*- coding: utf-8 -*- """Utilidades para envío de resultados por Telegram.""" from __future__ import annotations import os from pathlib import Path import requests TELEGRAM_BOT_TOKEN_ENV = "TELEGRAM_BOT_TOKEN" TELEGRAM_CHAT_ID_ENV = "TELEGRAM_CHAT_ID" def telegram_configured() -> bool: return bool(os.getenv(TELEGRAM_BOT_TOKEN_ENV) and os.getenv(TELEGRAM_CHAT_ID_ENV)) def send_telegram_message(text: str) -> dict | None: token = os.getenv(TELEGRAM_BOT_TOKEN_ENV) chat_id = os.getenv(TELEGRAM_CHAT_ID_ENV) if not token or not chat_id: return None url = f"https://api.telegram.org/bot{token}/sendMessage" response = requests.post(url, data={"chat_id": chat_id, "text": text}, timeout=60) response.raise_for_status() return response.json() def send_telegram_document(file_path: str | Path, caption: str | None = None) -> dict | None: token = os.getenv(TELEGRAM_BOT_TOKEN_ENV) chat_id = os.getenv(TELEGRAM_CHAT_ID_ENV) if not token or not chat_id: return None file_path = Path(file_path) if not file_path.exists(): raise FileNotFoundError(f"No existe el archivo para Telegram: {file_path}") url = f"https://api.telegram.org/bot{token}/sendDocument" with file_path.open("rb") as f: files = {"document": (file_path.name, f)} data = {"chat_id": chat_id} if caption: data["caption"] = caption[:1024] response = requests.post(url, data=data, files=files, timeout=180) response.raise_for_status() return response.json() def send_result_files(current_xlsx: str | Path, incremental_xlsx: str | Path | None, summary: str) -> list[str]: """Envía resumen y archivos disponibles. Devuelve mensajes de estado.""" statuses: list[str] = [] if not telegram_configured(): return ["Telegram no configurado: faltan TELEGRAM_BOT_TOKEN y/o TELEGRAM_CHAT_ID."] send_telegram_message(summary) statuses.append("Resumen enviado por Telegram.") send_telegram_document(current_xlsx, caption="Excel de la consulta actual") statuses.append("Excel de consulta actual enviado por Telegram.") if incremental_xlsx: send_telegram_document(incremental_xlsx, caption="Excel de la base incremental Neon") statuses.append("Excel de base incremental enviado por Telegram.") return statuses