Spaces:
Sleeping
Sleeping
File size: 2,367 Bytes
8b7419b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | # -*- 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
|