transport.py

Go to the documentation of this file.

"""Thin raw Telegram Bot API transport over httpx.
 
Plan 236 adds the three pieces an inline-keyboard decision needs, and nothing
else: ``getUpdates`` names ``callback_query`` in ``allowed_updates`` (the Bot
API remembers the last filter a caller set, so leaving it implicit would let
an earlier caller's filter silently drop taps); ``sendMessage`` accepts an
already-built ``reply_markup``; and ``answerCallbackQuery`` exists so a stored
tap can be acknowledged. :func:`inline_keyboard` is the only builder of a
keyboard and refuses any label outside the reply vocabulary: buttons are the
vocabulary, nothing else.
 
Dependencies
------------
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (REPLY_VOCABULARY - the only labels a keyboard button may carry)
- httpx (Client - the HTTP client for Bot API calls, environment trust off)
"""
from __future__ import annotations
 
from collections.abc import Mapping, Sequence
from typing import Any
 
import httpx
from oc_diagnostics import _dbg
from option_c_connector_contracts import REPLY_VOCABULARY
 
DEFAULT_API_ROOT = "https://api.telegram.org"
 
#: The update kinds the inbound poller handles. Sent on every ``getUpdates``
#: so the server-side filter is always the one this code expects.
ALLOWED_UPDATES: tuple[str, ...] = ("message", "callback_query")
 
#: Callback data is ``<TOKEN><separator><audit_id>``; the poller splits on
#: the first separator and refuses anything that is not exactly that.
CALLBACK_SEPARATOR = ":"
 
#: Bot API bound on ``callback_data`` (bytes).
MAX_CALLBACK_DATA = 64
 
 
class TelegramTransportError(RuntimeError):
    """A sanitized Bot API failure that never includes the request URL."""
 
 
def callback_data(token: str, audit_id: str) -> str:
    """Return the callback data for one vocabulary button on one question."""
    if token not in REPLY_VOCABULARY:
        raise ValueError(
            f"button label {token!r} is not a reply-vocabulary token; buttons may "
            f"carry only {', '.join(REPLY_VOCABULARY)}"
        )
    reference = audit_id.strip()
    if not reference or reference != audit_id or CALLBACK_SEPARATOR in reference:
        raise ValueError("inline keyboard callback data needs the send's audit id")
    data = f"{token}{CALLBACK_SEPARATOR}{reference}"
    if len(data.encode("utf-8")) > MAX_CALLBACK_DATA:
        raise ValueError("inline keyboard callback data exceeds the Bot API bound")
    return data
 
 
def inline_keyboard(tokens: Sequence[str], audit_id: str) -> dict[str, Any]:
    """Return a one-row ``InlineKeyboardMarkup`` of vocabulary buttons.
 
    Every button's callback data is ``<TOKEN>:<audit_id>`` for the SAME audit
    id, so a tap names the question it answers and nothing else. Labels are
    the tokens themselves; no URL, web-app, or free-label button can be built
    here, which is what keeps the structured-only ruling mechanical.
    """
    if not tokens:
        raise ValueError("an inline keyboard needs at least one button")
    row = [
        {"text": token, "callback_data": callback_data(token, audit_id)}
        for token in tokens
    ]
    return {"inline_keyboard": [row]}
 
 
class TelegramTransport:
    """Implement the Bot API methods oc_connect needs, environment trust off."""
 
    def __init__(
        self,
        raw_value: str,
        *,
        client: httpx.Client | None = None,
        api_root: str = DEFAULT_API_ROOT,
    ) -> None:
        if not raw_value.strip():
            raise ValueError("Telegram credential must not be blank")
        self._raw_value = raw_value.strip()
        self._api_root = api_root.rstrip("/")
        self._client = client or httpx.Client(trust_env=False, timeout=10.0)
 
    def _request(self, method: str, payload: Mapping[str, Any]) -> Any:
        url = f"{self._api_root}/bot{self._raw_value}/{method}"
        try:
            response = self._client.post(url, json=dict(payload))
        except httpx.HTTPError as exc:
            raise TelegramTransportError(
                f"Telegram {method} request failed ({type(exc).__name__})"
            ) from None
        try:
            body = response.json()
        except ValueError as exc:
            raise TelegramTransportError(
                f"Telegram {method} returned invalid JSON ({type(exc).__name__})"
            ) from None
        if response.status_code >= 400 or not isinstance(body, dict) or not body.get("ok"):
            raise TelegramTransportError(
                f"Telegram {method} was rejected (HTTP {response.status_code})"
            )
        _dbg(
            "OC_CONNECT.TELEGRAM",
            "_request",
            f"Bot API method={method} status={response.status_code}",
            3,
        )
        return body.get("result")
 
    def get_updates(
        self,
        *,
        offset: int | None = None,
        timeout: int = 0,
    ) -> list[dict[str, Any]]:
        """Return available updates, optionally long-polling from *offset*.
 
        *offset* is the standard Bot API cursor: passing it acknowledges every
        earlier update, which is what makes the inbound poller's deliberate
        refusal to advance past a deferred message actually defer it. A
        non-zero *timeout* is the §4 long poll; setup and one-shot CLI reads
        leave it at zero. ``allowed_updates`` is sent explicitly every time
        (plan 236): the Bot API persists the last filter it was given, so an
        omitted list would inherit whatever an earlier caller set.
        """
        if timeout < 0:
            raise ValueError("long-poll timeout must not be negative")
        payload: dict[str, Any] = {
            "timeout": timeout,
            "allowed_updates": list(ALLOWED_UPDATES),
        }
        if offset is not None:
            payload["offset"] = offset
        result = self._request("getUpdates", payload)
        if not isinstance(result, list):
            raise TelegramTransportError("Telegram getUpdates returned a non-list result")
        return [item for item in result if isinstance(item, dict)]
 
    def send_message(
        self,
        chat_id: str,
        text: str,
        *,
        reply_markup: Mapping[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Send one text message to an already-pinned chat.
 
        *reply_markup*, when given, is an already-built keyboard (see
        :func:`inline_keyboard`); a plain send carries no markup key at all.
        """
        payload: dict[str, Any] = {"chat_id": chat_id, "text": text}
        if reply_markup is not None:
            payload["reply_markup"] = dict(reply_markup)
        result = self._request("sendMessage", payload)
        if not isinstance(result, dict):
            raise TelegramTransportError("Telegram sendMessage returned a non-object result")
        return result
 
    def answer_callback_query(self, callback_query_id: str) -> bool:
        """Acknowledge one button tap so the operator's client stops spinning.
 
        No text, no alert: the acknowledgement carries nothing but the id.
        """
        if not callback_query_id.strip():
            raise ValueError("a callback acknowledgement needs the callback query id")
        result = self._request(
            "answerCallbackQuery", {"callback_query_id": callback_query_id}
        )
        return bool(result)
 
    def close(self) -> None:
        """Close the owned or injected HTTP client."""
        self._client.close()
 
 
__all__ = [
    "ALLOWED_UPDATES",
    "CALLBACK_SEPARATOR",
    "DEFAULT_API_ROOT",
    "MAX_CALLBACK_DATA",
    "TelegramTransport",
    "TelegramTransportError",
    "callback_data",
    "inline_keyboard",
]