153 lines
4.9 KiB
Python
153 lines
4.9 KiB
Python
"""HTTP client for MAX Bot API (platform-api.max.ru)."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MaxApiError(Exception):
|
|
def __init__(self, status: int, payload: Any):
|
|
super().__init__(f"MAX API error {status}: {payload}")
|
|
self.status = status
|
|
self.payload = payload
|
|
|
|
|
|
class MaxBotApi:
|
|
"""
|
|
Minimal async client. Auth: raw token in Authorization header (same as official TS SDK).
|
|
"""
|
|
|
|
def __init__(self, token: str, base_url: str = "https://platform-api.max.ru") -> None:
|
|
self._token = token
|
|
self._base = base_url.rstrip("/")
|
|
self._client = httpx.AsyncClient(
|
|
base_url=self._base,
|
|
headers={"Authorization": token},
|
|
timeout=httpx.Timeout(120.0, connect=30.0),
|
|
)
|
|
|
|
async def aclose(self) -> None:
|
|
await self._client.aclose()
|
|
|
|
async def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
json: Any | None = None,
|
|
) -> Any:
|
|
response = await self._client.request(method, path, params=params, json=json)
|
|
payload: Any
|
|
try:
|
|
payload = response.json()
|
|
except Exception:
|
|
payload = response.text
|
|
if response.status_code >= 400:
|
|
if isinstance(payload, dict):
|
|
raise MaxApiError(
|
|
response.status_code,
|
|
{"code": payload.get("code"), "message": payload.get("message", payload)},
|
|
)
|
|
raise MaxApiError(response.status_code, payload)
|
|
return payload
|
|
|
|
async def get_me(self) -> dict[str, Any]:
|
|
data = await self._request("GET", "/me")
|
|
return dict(data) if isinstance(data, dict) else {}
|
|
|
|
async def get_updates(
|
|
self,
|
|
*,
|
|
marker: int | None = None,
|
|
limit: int = 100,
|
|
timeout: int = 30,
|
|
types: list[str] | None = None,
|
|
) -> tuple[list[dict[str, Any]], int | None]:
|
|
params: dict[str, Any] = {"limit": limit, "timeout": timeout}
|
|
if marker is not None:
|
|
params["marker"] = marker
|
|
if types:
|
|
params["types"] = ",".join(types)
|
|
data = await self._request("GET", "/updates", params=params)
|
|
if not isinstance(data, dict):
|
|
return [], None
|
|
raw_updates = data.get("updates") or []
|
|
updates = [u for u in raw_updates if isinstance(u, dict)]
|
|
marker_out = data.get("marker")
|
|
return updates, marker_out if isinstance(marker_out, int) else None
|
|
|
|
async def send_message_to_chat(
|
|
self,
|
|
chat_id: int,
|
|
*,
|
|
text: str | None = None,
|
|
attachments: list[dict[str, Any]] | None = None,
|
|
fmt: str | None = None,
|
|
) -> dict[str, Any]:
|
|
params: dict[str, Any] = {"chat_id": chat_id}
|
|
body: dict[str, Any] = {}
|
|
if text is not None:
|
|
body["text"] = text
|
|
if attachments is not None:
|
|
body["attachments"] = attachments
|
|
if fmt:
|
|
body["format"] = fmt
|
|
return await self._request("POST", "/messages", params=params, json=body)
|
|
|
|
async def send_message_to_user(
|
|
self,
|
|
user_id: int,
|
|
*,
|
|
text: str | None = None,
|
|
attachments: list[dict[str, Any]] | None = None,
|
|
fmt: str | None = None,
|
|
) -> dict[str, Any]:
|
|
params: dict[str, Any] = {"user_id": user_id}
|
|
body: dict[str, Any] = {}
|
|
if text is not None:
|
|
body["text"] = text
|
|
if attachments is not None:
|
|
body["attachments"] = attachments
|
|
if fmt:
|
|
body["format"] = fmt
|
|
return await self._request("POST", "/messages", params=params, json=body)
|
|
|
|
async def send_chat_action(self, chat_id: int, action: str) -> Any:
|
|
return await self._request(
|
|
"POST",
|
|
f"/chats/{chat_id}/actions",
|
|
json={"action": action},
|
|
)
|
|
|
|
async def get_upload_url(self, upload_type: str) -> dict[str, Any]:
|
|
data = await self._request("POST", "/uploads", params={"type": upload_type})
|
|
return dict(data) if isinstance(data, dict) else {}
|
|
|
|
async def answer_callback(
|
|
self,
|
|
callback_id: str,
|
|
*,
|
|
message: dict[str, Any] | None = None,
|
|
notification: str | None = None,
|
|
) -> Any:
|
|
body: dict[str, Any] = {}
|
|
if message is not None:
|
|
body["message"] = message
|
|
if notification is not None:
|
|
body["notification"] = notification
|
|
return await self._request(
|
|
"POST",
|
|
"/answers",
|
|
params={"callback_id": callback_id},
|
|
json=body if body else {},
|
|
)
|
|
|
|
async def download_file(self, url: str) -> bytes:
|
|
response = await self._client.get(url)
|
|
response.raise_for_status()
|
|
return response.content
|