176 lines
7.2 KiB
Python
176 lines
7.2 KiB
Python
"""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], "parse_mode": "HTML"}
|
|
if reply_markup:
|
|
payload[REPLY_MARKUP] = reply_markup
|
|
if reply_to:
|
|
payload["reply_to_message_id"] = reply_to
|
|
try:
|
|
return await self._call("sendMessage", payload)
|
|
except BaleError:
|
|
# parse_mode / markup rejection fallback: retry plain, keep the keyboard.
|
|
payload.pop("parse_mode", None)
|
|
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], "parse_mode": "HTML"}
|
|
if reply_markup:
|
|
payload[REPLY_MARKUP] = reply_markup
|
|
try:
|
|
return await self._call("editMessageText", payload)
|
|
except BaleError:
|
|
payload.pop("parse_mode", None)
|
|
return await self._call("editMessageText", payload)
|
|
|
|
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)
|