Skip to content

Server SDK

Установка

Требования

Для использования SDK Telesend вам понадобится:

  • Python версии 3.7 или выше
  • Проект на Python
  • API ключ

Установка пакета

Terminal window
pip install telesend-server-sdk

Получение API ключа

API ключ необходим для аутентификации вашего приложения при взаимодействии с Telesend. Для получения API ключа:

  1. Войдите в свой аккаунт на сайте Telesend
  2. Перейдите к своему проекту:
    • Если у вас уже есть проект - выберите его в личном кабинете.
    • Если у вас еще нет проекта - создайте новый:
      1. Нажмите на кнопку “Создать проект”.
      2. Заполните все поля и нажмите на кнопку “Создать”.
  3. Найдите кнопку “Редактировать” и скопируйте ваш API ключ

Храните ваш API ключ в безопасном месте и не публикуйте его в открытом доступе.

Инициализация клиента

Для начала работы с SDK необходимо создать экземпляр класса TelesendClient:

from telesend.client import TelesendClient, compose_message
from telesend.types import TelesendClientConfig, MessageQueueItem
# Функция для отправки сообщения в Telegram
async def send_telegram_message(message_data: MessageQueueItem):
user_id = message_data.user_id
message = message_data.message
message_format = compose_message(message_data)
method = message_format["type"]
body = message_format["body"]
import requests
response = requests.post(
f"https://api.telegram.org/bot{BOT_TOKEN}/{method}",
json=body
)
if not response.ok:
error_data = response.json()
raise Exception(json.dumps(error_data))
# Создаем конфигурацию клиента
config = TelesendClientConfig(
api_key="ваш_api_ключ",
callback_hook_send_message=send_telegram_message
)
# Инициализируем клиент
client = TelesendClient(config)

Отправка аналитических событий

Идентификация пользователя

Прежде чем отправлять аналитические события, рекомендуется идентифицировать пользователя:

from telesend.types import UserDetails
# Идентификация пользователя
client.identify(UserDetails(
id="123456789",
username="username",
first_name="Имя",
last_name="Фамилия",
is_premium=False
))

Подробнее о методе client.identify

Отправка события

from telesend.types import Event
# Отправка аналитического события
client.track(
"123456789",
"button_click",
Event(
start_parameter="",
path="/start",
params={
"buttonId": "start"
},
language="ru",
device="mobile"
)
)

Подробнее о методе client.track

Обработка событий

SDK Telesend использует систему событий для отслеживания состояния подключения и отправки сообщений:

# Отслеживание отправки сообщений
def on_message_sent(user_id, success):
if success:
print(f"Сообщение успешно отправлено пользователю {user_id}")
else:
print(f"Ошибка при отправке сообщения пользователю {user_id}")
# Отслеживание ошибок
def on_error(error):
print(f"Произошла ошибка: {error}")
# Подписка на события
client.on_message_sent(on_message_sent)
client.on_error(on_error)

Примеры использования

Отправка аналитического события

Пример отправки события о нажатии на кнопку:

from telesend.types import UserDetails, Event
# Идентификация пользователя
client.identify(UserDetails(
id="123456789",
username="username",
first_name="Имя",
last_name="Фамилия",
is_premium=True
))
# Отправка события
client.track(
"123456789",
"button_click",
Event(
start_parameter="",
path="/start",
params={
"buttonId": "start_button",
"buttonText": "Начать"
},
language="ru",
device="mobile"
)
)

Интеграция с Telegram API

SDK Telesend не взаимодействует напрямую с Telegram API. Вместо этого он использует переданную в конструкторе функцию callback_hook_send_message для отправки сообщений. Это позволяет разработчикам иметь полный контроль над процессом отправки сообщений.

Пример реализации колбэка для отправки сообщений:

import requests
import json
from telesend.client import TelesendClient, compose_message
from telesend.types import TelesendClientConfig, MessageQueueItem
# Функция для отправки сообщения в Telegram
def send_telegram_message(message_data: MessageQueueItem):
message_format = compose_message(message_data)
method = message_format["type"]
body = message_format["body"]
response = requests.post(
f"https://api.telegram.org/bot{BOT_TOKEN}/{method}",
json=body
)
if not response.ok:
error_data = response.json()
raise Exception(json.dumps(error_data))

API и типы данных

Класс TelesendClient

Основной класс для работы с SDK Telesend.

Конструктор

TelesendClient(config: TelesendClientConfig)
class TelesendClientConfig:
def __init__(
self,
api_key: str,
base_url: Optional[str] = None,
migrate_users_hook: Optional[Callable[[], List[UserData]]] = None,
callback_hook_send_message: Optional[Callable[[MessageQueueItem], None]] = None
):
# ...

Методы

track

Отправляет аналитическое событие на сервер.

track(user_id: Union[str, int], event_type: str, payload: Event) -> None
identify

Идентифицирует пользователя.

identify(user: UserDetails) -> None
is_connected

Проверяет, установлено ли соединение с сервисом.

is_connected() -> bool
close

Закрывает соединение с сервисом при завершении работы с SDK.

close() -> None

События

on_message_sent

Подписка на события отправки сообщений.

on_message_sent(callback: Callable[[str, bool], None]) -> None
on_error

Подписка на события ошибок.

on_error(callback: Callable[[Exception], None]) -> None
on_connected

Подписка на события успешного подключения.

on_connected(callback: Callable[[], None]) -> None
on_disconnected

Подписка на события отключения.

on_disconnected(callback: Callable[[], None]) -> None

Вспомогательные функции

compose_message

Формирует данные для запроса к Telegram API на основе сообщения.

compose_message(message_data: MessageQueueItem) -> Dict[str, Any]

Типы данных

UserDetails и Event

class UserDetails:
def __init__(
self,
id: str, # Уникальный идентификатор пользователя
username: Optional[str] = None, # Имя пользователя в Telegram (без @)
first_name: Optional[str] = None, # Имя пользователя
last_name: Optional[str] = None, # Фамилия пользователя
is_premium: Optional[bool] = None,# Статус премиум-подписки пользователя
time_zone: Optional[int] = None # Часовой пояс пользователя (от -12 до +12)
):
# ...
class Event:
def __init__(
self,
start_parameter: str, # Начальный параметр события
path: str, # Путь, на котором произошло событие
params: Dict[str, Any], # Дополнительные параметры события
language: Optional[str] = None, # Язык пользователя
device: Optional[str] = None # Устройство пользователя
):
# ...

Классы для работы с сообщениями

# Кнопка для клавиатуры Telegram
class Button:
def __init__(
self,
text: str, # Текст кнопки
url: Optional[str] = None, # URL для перехода
callback_data: Optional[str] = None # Данные для обратного вызова
):
# ...
# Медиаэлемент для группы медиа
class MediaItem:
def __init__(
self,
type: Literal['audio', 'document', 'photo', 'video'], # Тип медиа
url: str, # URL медиафайла
caption: Optional[str] = None # Подпись к медиафайлу
):
# ...
# Базовый класс для всех типов сообщений
class BaseMessage:
def __init__(
self,
disable_notification: bool = False, # Отключить уведомление
parse_mode: Optional[str] = None # Режим парсинга (Markdown, HTML)
):
# ...
# Текстовое сообщение
class MessageText(BaseMessage):
def __init__(
self,
text: str, # Текст сообщения
buttons: Optional[List[Button]] = None, # Кнопки
disable_notification: bool = False,
parse_mode: Optional[str] = None
):
# ...
# Группа медиафайлов
class MessageMediaGroup(BaseMessage):
def __init__(
self,
items: List[MediaItem], # Медиафайлы
disable_notification: bool = False
):
# ...
# Фото
class MessagePhoto(BaseMessage):
def __init__(
self,
photo: str, # URL фото
caption: Optional[str] = None, # Подпись к фото
buttons: Optional[List[Button]] = None, # Кнопки
disable_notification: bool = False,
parse_mode: Optional[str] = None
):
# ...
# Видео-заметка
class MessageVideoNote(BaseMessage):
def __init__(
self,
video_note: str, # URL видео-заметки
buttons: Optional[List[Button]] = None, # Кнопки
disable_notification: bool = False
):
# ...
# Голосовое сообщение
class MessageVoice(BaseMessage):
def __init__(
self,
voice: str, # URL голосового сообщения
caption: Optional[str] = None, # Подпись
disable_notification: bool = False,
parse_mode: Optional[str] = None
):
# ...
# Опрос
class MessagePoll(BaseMessage):
def __init__(
self,
question: str, # Вопрос
options: List[str], # Варианты ответов
is_anonymous: bool = True, # Анонимный опрос
allows_multiple_answers: bool = False, # Множественный выбор
disable_notification: bool = False
):
# ...
# Объединение всех типов сообщений
TelegramMessage = Union[
MessageText,
MessageMediaGroup,
MessagePhoto,
MessageVideoNote,
MessageVoice,
MessagePoll
]

Прочие типы

# Данные пользователя
class UserData:
def __init__(self, tg: Union[str, int]): # Telegram ID пользователя
# ...
# Элемент очереди сообщений
class MessageQueueItem:
def __init__(self, user_id: str, message: TelegramMessage):
# ...

Создание рассылки

Для отправки сообщений группе пользователей, используйте метод broadcast:

from telesend.types import BroadcastOptions, MessageText
# Создание текстового сообщения
message = MessageText(
text="Привет! Это сообщение от Telesend SDK",
parse_mode="HTML"
)
# Создание параметров рассылки
options = BroadcastOptions(
users=["123456789", "987654321"], # ID пользователей Telegram
content=message
)
# Отправка рассылки
result = client.broadcast(options)
if result:
print(f"Сообщение отправлено, ID рассылки: {result['broadcastId']}")

Для отправки сообщения всем пользователям, используйте специальное значение 'all' и реализуйте хук migrate_users_hook:

from telesend.types import BroadcastOptions, MessageText, TelesendClientConfig, UserData
# Функция для получения списка всех пользователей
def get_all_users():
# Загрузка пользователей из вашей базы данных
users_from_db = [{"id": "123456789"}, {"id": "987654321"}]
return [UserData(tg=user["id"]) for user in users_from_db]
# Создание конфигурации с хуком для получения пользователей
config = TelesendClientConfig(
api_key="ваш_api_ключ",
callback_hook_send_message=send_telegram_message,
migrate_users_hook=get_all_users
)
client = TelesendClient(config)
# Создание параметров рассылки для всех пользователей
options = BroadcastOptions(
users="all", # Специальное значение для отправки всем пользователям
content=message
)
# Отправка рассылки
result = client.broadcast(options)