core/telegram_bot.py

# core/tools/telegram_bot.py
from __future__ import annotations

import os
import requests
from typing import Any, Dict, Optional


def _env(name: str) -> str:
    return os.getenv(name, "").strip()


def tg_token() -> str:
    return _env("TELEGRAM_BOT_TOKEN")


def tg_api(method: str) -> str:
    return f"https://api.telegram.org/bot{tg_token()}/{method}"


def send_message(
    chat_id: int,
    text: str,
    reply_to_message_id: Optional[int] = None,
    reply_markup: Optional[Dict[str, Any]] = None,
    disable_web_page_preview: Optional[bool] = None,
) -> Dict[str, Any]:
    """
    sendMessage wrapper.
    - reply_markup: e.g. {"inline_keyboard": [[{"text":"...", "callback_data":"..."}]]}
    """
    if not tg_token():
        return {"ok": False, "error": "missing_token"}

    payload: Dict[str, Any] = {"chat_id": chat_id, "text": text}

    # Optional: reply to the triggering message
    if reply_to_message_id is not None:
        payload["reply_to_message_id"] = reply_to_message_id

    if reply_markup is not None:
        payload["reply_markup"] = reply_markup

    if disable_web_page_preview is not None:
        payload["disable_web_page_preview"] = bool(disable_web_page_preview)

    r = requests.post(tg_api("sendMessage"), json=payload, timeout=12)
    try:
        return r.json()
    except Exception:
        return {"ok": False, "status": r.status_code, "text": r.text}


def edit_message_text(
    chat_id: int,
    message_id: int,
    text: str,
    reply_markup: Optional[Dict[str, Any]] = None,
    disable_web_page_preview: Optional[bool] = None,
) -> Dict[str, Any]:
    """
    editMessageText wrapper (useful to update status messages after button click)
    """
    if not tg_token():
        return {"ok": False, "error": "missing_token"}

    payload: Dict[str, Any] = {"chat_id": chat_id, "message_id": message_id, "text": text}

    if reply_markup is not None:
        payload["reply_markup"] = reply_markup

    if disable_web_page_preview is not None:
        payload["disable_web_page_preview"] = bool(disable_web_page_preview)

    r = requests.post(tg_api("editMessageText"), json=payload, timeout=12)
    try:
        return r.json()
    except Exception:
        return {"ok": False, "status": r.status_code, "text": r.text}


def answer_callback_query(
    callback_query_id: str,
    text: Optional[str] = None,
    show_alert: bool = False,
) -> Dict[str, Any]:
    """
    answerCallbackQuery wrapper: stops the Telegram button spinner.
    """
    if not tg_token():
        return {"ok": False, "error": "missing_token"}

    payload: Dict[str, Any] = {"callback_query_id": callback_query_id, "show_alert": bool(show_alert)}
    if text:
        payload["text"] = str(text)

    r = requests.post(tg_api("answerCallbackQuery"), json=payload, timeout=10)
    try:
        return r.json()
    except Exception:
        return {"ok": False, "status": r.status_code, "text": r.text}


def send_chat_action(chat_id: int, action: str = "typing") -> None:
    if not tg_token():
        return
    try:
        requests.post(tg_api("sendChatAction"), json={"chat_id": chat_id, "action": action}, timeout=8)
    except Exception:
        pass


def is_allowed(chat_id: int) -> bool:
    allow = _env("TELEGRAM_ALLOWED_CHAT_IDS")
    if not allow:
        return True
    allowed = {c.strip() for c in allow.split(",") if c.strip()}
    return str(chat_id) in allowed