telegram_utils.py

# -*- 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