122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import http.client
|
|
import json
|
|
from typing import Any
|
|
from urllib import parse
|
|
|
|
|
|
class TelegramAPI:
|
|
def __init__(self, token: str, proxy_url: str | None = None) -> None:
|
|
self.host = "api.telegram.org"
|
|
self.base_path = f"/bot{token}"
|
|
self.proxy = parse.urlparse(proxy_url) if proxy_url else None
|
|
|
|
def _connection(self) -> http.client.HTTPSConnection:
|
|
if self.proxy:
|
|
conn = http.client.HTTPSConnection(
|
|
self.proxy.hostname,
|
|
self.proxy.port or 443,
|
|
timeout=90,
|
|
)
|
|
conn.set_tunnel(self.host, 443)
|
|
return conn
|
|
return http.client.HTTPSConnection(self.host, 443, timeout=90)
|
|
|
|
def _post(self, method: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
conn = self._connection()
|
|
try:
|
|
conn.request(
|
|
"POST",
|
|
f"{self.base_path}/{method}",
|
|
body=data,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
response = conn.getresponse()
|
|
body = response.read().decode("utf-8")
|
|
finally:
|
|
conn.close()
|
|
return json.loads(body)
|
|
|
|
def _get(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
|
|
query = parse.urlencode(params)
|
|
path = f"{self.base_path}/{method}"
|
|
if query:
|
|
path = f"{path}?{query}"
|
|
conn = self._connection()
|
|
try:
|
|
conn.request("GET", path)
|
|
response = conn.getresponse()
|
|
body = response.read().decode("utf-8")
|
|
finally:
|
|
conn.close()
|
|
return json.loads(body)
|
|
|
|
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,
|
|
*,
|
|
parse_mode: str | None = None,
|
|
reply_markup: dict[str, Any] | None = None,
|
|
) -> int:
|
|
payload: dict[str, Any] = {"chat_id": chat_id, "text": text}
|
|
if parse_mode:
|
|
payload["parse_mode"] = parse_mode
|
|
if reply_markup:
|
|
payload["reply_markup"] = reply_markup
|
|
response = self._post("sendMessage", payload)
|
|
result = response.get("result") or {}
|
|
return int(result.get("message_id", 0))
|
|
|
|
def edit_message_text(
|
|
self,
|
|
chat_id: int,
|
|
message_id: int,
|
|
text: str,
|
|
*,
|
|
parse_mode: str | None = None,
|
|
reply_markup: dict[str, Any] | None = None,
|
|
) -> None:
|
|
try:
|
|
payload: dict[str, Any] = {
|
|
"chat_id": chat_id,
|
|
"message_id": message_id,
|
|
"text": text,
|
|
}
|
|
if parse_mode:
|
|
payload["parse_mode"] = parse_mode
|
|
if reply_markup is not None:
|
|
payload["reply_markup"] = reply_markup
|
|
self._post(
|
|
"editMessageText",
|
|
payload,
|
|
)
|
|
except Exception as exc:
|
|
if "message is not modified" in str(exc).lower():
|
|
return
|
|
raise
|
|
|
|
def send_chat_action(self, chat_id: int, action: str = "typing") -> None:
|
|
self._post(
|
|
"sendChatAction",
|
|
{
|
|
"chat_id": chat_id,
|
|
"action": action,
|
|
},
|
|
)
|
|
|
|
def answer_callback_query(self, callback_query_id: str, text: str | None = None) -> None:
|
|
payload: dict[str, Any] = {"callback_query_id": callback_query_id}
|
|
if text:
|
|
payload["text"] = text
|
|
self._post("answerCallbackQuery", payload)
|