Telegram Provider¶
Обёртка над Aiogram для создания Telegram ботов.
Возможности¶
- Упрощённый API для типичных операций
- Встроенная обработка ошибок
- FSM (Finite State Machine)
- Inline keyboards
- Медиа сообщения
- Webhooks и polling
Использование¶
Базовый бот¶
from kit.providers.telegram import TelegramService
service = TelegramService(token="123456:ABC...")
# Простая команда
@service.command("start")
async def start_handler(message):
await message.answer("Привет! Я бот.")
# Текстовые сообщения
@service.on_text()
async def text_handler(message):
await message.answer(f"Вы написали: {message.text}")
# Запуск
await service.start_polling()
Inline клавиатуры¶
from kit.providers.telegram import InlineKeyboard
@service.command("menu")
async def menu_handler(message):
keyboard = InlineKeyboard()
keyboard.add_button("Опция 1", callback_data="option_1")
keyboard.add_button("Опция 2", callback_data="option_2")
keyboard.row() # Новый ряд
keyboard.add_button("Назад", callback_data="back")
await message.answer("Выберите опцию:", reply_markup=keyboard.markup)
@service.on_callback("option_1")
async def option1_handler(callback):
await callback.answer("Выбрана опция 1!")
await callback.message.edit_text("Вы выбрали опцию 1")
FSM (состояния)¶
from kit.providers.telegram import FSM, State
class OrderStates(FSM):
waiting_name = State()
waiting_address = State()
waiting_confirm = State()
@service.command("order")
async def order_start(message, state):
await state.set(OrderStates.waiting_name)
await message.answer("Введите ваше имя:")
@service.on_state(OrderStates.waiting_name)
async def process_name(message, state):
await state.update(name=message.text)
await state.set(OrderStates.waiting_address)
await message.answer("Введите адрес доставки:")
@service.on_state(OrderStates.waiting_address)
async def process_address(message, state):
await state.update(address=message.text)
await state.set(OrderStates.waiting_confirm)
data = await state.get_data()
await message.answer(
f"Подтвердите заказ:\nИмя: {data['name']}\nАдрес: {data['address']}",
reply_markup=InlineKeyboard()
.add_button("Подтвердить", "confirm")
.add_button("Отмена", "cancel")
.markup
)
Отправка медиа¶
# Фото
await service.send_photo(
chat_id=user_id,
photo="path/to/photo.jpg",
caption="Описание фото"
)
# Видео
await service.send_video(
chat_id=user_id,
video="path/to/video.mp4",
caption="Ваше видео готово!"
)
# Документ
await service.send_document(
chat_id=user_id,
document="path/to/file.pdf"
)
# Голосовое
await service.send_voice(
chat_id=user_id,
voice="path/to/audio.ogg"
)
Webhooks¶
from fastapi import FastAPI
from kit.providers.telegram import TelegramService
app = FastAPI()
service = TelegramService(token="...")
@app.on_event("startup")
async def startup():
await service.set_webhook(
url="https://yourdomain.com/webhook",
secret_token="your-secret"
)
@app.post("/webhook")
async def webhook(request: Request):
return await service.process_webhook(request)
API Reference¶
TelegramService¶
class TelegramService:
def __init__(
self,
token: str,
parse_mode: str = "HTML"
)
# Декораторы
def command(self, command: str) -> Callable
def on_text(self, filter: Callable = None) -> Callable
def on_callback(self, data: str = None) -> Callable
def on_state(self, state: State) -> Callable
# Отправка сообщений
async def send_message(
self,
chat_id: int,
text: str,
reply_markup: Any = None
) -> Message
async def send_photo(
self,
chat_id: int,
photo: str,
caption: str = None
) -> Message
async def send_video(
self,
chat_id: int,
video: str,
caption: str = None
) -> Message
async def send_document(
self,
chat_id: int,
document: str,
caption: str = None
) -> Message
async def send_voice(
self,
chat_id: int,
voice: str,
caption: str = None
) -> Message
# Lifecycle
async def start_polling(self) -> None
async def set_webhook(self, url: str, secret_token: str = None) -> None
async def process_webhook(self, request: Request) -> Response
InlineKeyboard¶
class InlineKeyboard:
def add_button(
self,
text: str,
callback_data: str = None,
url: str = None
) -> "InlineKeyboard"
def row(self) -> "InlineKeyboard"
@property
def markup(self) -> InlineKeyboardMarkup
Примеры из production¶
DedMoroz.ai — бот поздравлений¶
class DedMorozBot:
def __init__(self, token: str):
self.service = TelegramService(token)
self.llm = OpenAIProvider(...)
self.tts = EdgeTTSProvider(...)
self.setup_handlers()
def setup_handlers(self):
@self.service.command("start")
async def start(message):
await message.answer(
"Привет! Я Дед Мороз! 🎅\n"
"Отправь мне имя и пожелания, и я создам персональное поздравление!"
)
@self.service.on_text()
async def create_greeting(message):
# Показываем "печатает..."
await self.service.send_chat_action(message.chat.id, "typing")
# Генерируем текст
greeting = await self.llm.generate(
prompt=f"Создай новогоднее поздравление для: {message.text}",
system="Ты Дед Мороз..."
)
# Генерируем озвучку
audio = await self.tts.synthesize(
greeting.content,
f"./temp/{message.chat.id}.mp3"
)
# Отправляем
await message.answer(greeting.content)
await self.service.send_voice(
message.chat.id,
audio.audio_path
)
async def run(self):
await self.service.start_polling()
AI Sales Manager — бот продаж¶
class SalesBot:
def __init__(self):
self.service = TelegramService(token=settings.telegram_token)
self.db = Database()
def setup(self):
@self.service.command("catalog")
async def show_catalog(message):
products = await self.db.get_products()
keyboard = InlineKeyboard()
for product in products[:10]:
keyboard.add_button(
f"{product.name} - ${product.price}",
callback_data=f"product_{product.id}"
)
keyboard.row()
await message.answer("Наш каталог:", reply_markup=keyboard.markup)
@self.service.on_callback_prefix("product_")
async def show_product(callback):
product_id = callback.data.replace("product_", "")
product = await self.db.get_product(product_id)
await callback.message.answer_photo(
photo=product.image_url,
caption=f"<b>{product.name}</b>\n\n{product.description}\n\nЦена: ${product.price}",
reply_markup=InlineKeyboard()
.add_button("Купить", callback_data=f"buy_{product_id}")
.add_button("Назад", callback_data="catalog")
.markup
)