"""Thin async client for the Bale Bot API. Bale's bot surface is a Telegram Bot API clone: ``POST {base}/bot{token}/{method}`` and file downloads at ``{base}/file/bot{token}/{file_path}``. Long polling via ``getUpdates`` avoids any webhook/TLS setup on the server. HTML parse_mode is requested first; if Bale rejects it for a given payload we transparently retry as plain text so messages never get lost. """ import asyncio import logging import httpx logger = logging.getLogger("bot.bale") REPLY_MARKUP = "reply_markup" class BaleError(Exception): def __init__(self, description: str, error_code: int | None = None): super().__init__(description) self.description = description self.error_code = error_code class BaleClient: def __init__(self, base_url: str, token: str, timeout: float = 60.0): self._base = base_url.rstrip("/") self._token = token self._timeout = timeout # ------------------------------------------------------------------ core def _method_url(self, method: str) -> str: return f"{self._base}/bot{self._token}/{method}" def _file_url(self, file_path: str) -> str: return f"{self._base}/file/bot{self._token}/{file_path.lstrip('/')}" async def _call(self, method: str, payload: dict | None = None, timeout: float | None = None) -> dict: async with httpx.AsyncClient(timeout=timeout or self._timeout) as client: try: resp = await client.post(self._method_url(method), json=payload or {}) except httpx.HTTPError as exc: raise BaleError(f"network error calling {method}: {exc}") from exc try: body = resp.json() except ValueError as exc: raise BaleError(f"non-JSON response for {method} (HTTP {resp.status_code})") from exc if not body.get("ok"): raise BaleError(str(body.get("description") or "unknown bale error"), resp.status_code) return body.get("result") # ------------------------------------------------------------- receiving async def get_updates(self, offset: int, timeout_seconds: int) -> list[dict]: """Long poll. ``timeout_seconds=0`` returns immediately (drain mode).""" result = await self._call( "getUpdates", {"offset": offset, "timeout": timeout_seconds, "allowed_updates": ["message", "callback_query"]}, timeout=timeout_seconds + 15, ) return result or [] # -------------------------------------------------------------- sending async def send_message(self, chat_id: int, text: str, reply_markup: dict | None = None, reply_to: int | None = None) -> dict: payload: dict = {"chat_id": chat_id, "text": text[:4000]} if reply_markup: payload[REPLY_MARKUP] = reply_markup if reply_to: payload["reply_to_message_id"] = reply_to return await self._call("sendMessage", payload) async def edit_message_text(self, chat_id: int, message_id: int, text: str, reply_markup: dict | None = None) -> dict: payload: dict = {"chat_id": chat_id, "message_id": message_id, "text": text[:4000]} if reply_markup: payload[REPLY_MARKUP] = reply_markup return await self._call("editMessageText", payload) async def delete_message(self, chat_id: int, message_id: int) -> bool: payload: dict = {"chat_id": chat_id, "message_id": message_id} try: return bool(await self._call("deleteMessage", payload)) except BaleError as exc: logger.warning("deleteMessage failed for chat %s msg %s: %s", chat_id, message_id, exc) return False except Exception as exc: logger.warning("unexpected error in deleteMessage for chat %s msg %s: %s", chat_id, message_id, exc) return False async def answer_callback_query(self, callback_query_id: str, text: str | None = None) -> None: payload: dict = {"callback_query_id": callback_query_id} if text: payload["text"] = text[:200] try: await self._call("answerCallbackQuery", payload) except BaleError as exc: logger.warning("answerCallbackQuery failed: %s", exc) async def send_chat_action(self, chat_id: int, action: str = "typing") -> None: try: await self._call("sendChatAction", {"chat_id": chat_id, "action": action}) except BaleError as exc: logger.debug("sendChatAction failed: %s", exc) # ---------------------------------------------------------------- files async def get_file(self, file_id: str) -> dict: return await self._call("getFile", {"file_id": file_id}) async def download_file(self, file_id: str) -> bytes: """Resolve a file_id to bytes (getFile → file_path → download URL).""" info = await self.get_file(file_id) file_path = info.get("file_path") if not file_path: raise BaleError("getFile returned no file_path") async with httpx.AsyncClient(timeout=120) as client: resp = await client.get(self._file_url(file_path)) resp.raise_for_status() return resp.content async def poll_loop(client: BaleClient, on_update, offset_key: str = "bale_offset") -> None: """Long-poll supervisor: resumable offset, one update at a time, never dies. ``on_update`` must be an async callable receiving the raw update dict. """ from app.db import SessionLocal from app.models import BotMeta async def load_offset() -> int: async with SessionLocal() as session: row = (await session.execute( __import__("sqlalchemy").select(BotMeta.value).where(BotMeta.key == offset_key) )).scalar() return int(row) if row else 0 async def save_offset(value: int) -> None: from sqlalchemy.dialects.postgresql import insert as pg_insert async with SessionLocal() as session: stmt = pg_insert(BotMeta).values(key=offset_key, value=str(value)) stmt = stmt.on_conflict_do_update(index_elements=[BotMeta.key], set_={"value": str(value)}) await session.execute(stmt) await session.commit() offset = await load_offset() if offset == 0: # Skip any backlog accumulated while the bot was down: peek once # without waiting and jump straight to the newest update. backlog = await client.get_updates(0, 0) if backlog: offset = max(u["update_id"] for u in backlog) + 1 logger.info("polling started (offset=%s)", offset) while True: try: updates = await client.get_updates(offset, 25) for update in updates: offset = max(offset, update["update_id"] + 1) await save_offset(offset) try: await on_update(update) except Exception: logger.exception("handler crashed on update %s", update.get("update_id")) except asyncio.CancelledError: raise except BaleError as exc: logger.warning("poll error: %s", exc) await asyncio.sleep(3) except Exception: logger.exception("unexpected poll error") await asyncio.sleep(3)