41 lines
1.6 KiB
Python
41 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
from urllib import parse, request as urllib_request
|
|
|
|
|
|
class TelegramAPI:
|
|
def __init__(self, token: str, proxy_url: str | None = None) -> None:
|
|
self.base_url = f"https://api.telegram.org/bot{token}"
|
|
handlers = []
|
|
if proxy_url:
|
|
handlers.append(urllib_request.ProxyHandler({"https": proxy_url, "http": proxy_url}))
|
|
self.opener = urllib_request.build_opener(*handlers) if handlers else urllib_request.build_opener()
|
|
|
|
def _post(self, method: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
req = urllib_request.Request(
|
|
f"{self.base_url}/{method}",
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with self.opener.open(req, timeout=90) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
def _get(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
|
|
query = parse.urlencode(params)
|
|
with self.opener.open(f"{self.base_url}/{method}?{query}", timeout=90) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
def get_updates(self, offset: int | None, timeout: int) -> list[dict[str, Any]]:
|
|
params: dict[str, Any] = {"timeout": timeout}
|
|
if offset is not None:
|
|
params["offset"] = offset
|
|
response = self._get("getUpdates", params)
|
|
return response.get("result", [])
|
|
|
|
def send_message(self, chat_id: int, text: str) -> None:
|
|
self._post("sendMessage", {"chat_id": chat_id, "text": text})
|