Server SDK
Установка
Требования
Для использования SDK Telesend вам понадобится:
- Python версии 3.7 или выше
- Проект на Python
- API ключ
Установка пакета
pip install telesend-server-sdkПолучение API ключа
API ключ необходим для аутентификации вашего приложения при взаимодействии с Telesend. Для получения API ключа:
- Войдите в свой аккаунт на сайте Telesend
- Перейдите к своему проекту:
- Если у вас уже есть проект - выберите его в личном кабинете.
- Если у вас еще нет проекта - создайте новый:
- Нажмите на кнопку “Создать проект”.
- Заполните все поля и нажмите на кнопку “Создать”.
- Найдите кнопку “Редактировать” и скопируйте ваш API ключ
Храните ваш API ключ в безопасном месте и не публикуйте его в открытом доступе.
Инициализация клиента
Для начала работы с SDK необходимо создать экземпляр класса TelesendClient:
from telesend.client import TelesendClient, compose_messagefrom telesend.types import TelesendClientConfig, MessageQueueItem
# Функция для отправки сообщения в Telegramasync 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 requestsimport jsonfrom telesend.client import TelesendClient, compose_messagefrom telesend.types import TelesendClientConfig, MessageQueueItem
# Функция для отправки сообщения в Telegramdef 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) -> Noneidentify
Идентифицирует пользователя.
identify(user: UserDetails) -> Noneis_connected
Проверяет, установлено ли соединение с сервисом.
is_connected() -> boolclose
Закрывает соединение с сервисом при завершении работы с SDK.
close() -> NoneСобытия
on_message_sent
Подписка на события отправки сообщений.
on_message_sent(callback: Callable[[str, bool], None]) -> Noneon_error
Подписка на события ошибок.
on_error(callback: Callable[[Exception], None]) -> Noneon_connected
Подписка на события успешного подключения.
on_connected(callback: Callable[[], None]) -> Noneon_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 # Устройство пользователя ): # ...Классы для работы с сообщениями
# Кнопка для клавиатуры Telegramclass 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)