Telegram-бот на Python: от токена до продакшена

import asyncio
from socialift import Telegram

async def main():
    async with Telegram.from_env() as bot:
        me = await bot.get_me()
        print("Бот:", me["username"])

        # chat_id — числовой ID или @username для канала
        await bot.send_message("@my_channel", "<b>Привет!</b> Бот запущен.",
                               parse_mode="HTML")

asyncio.run(main())

from socialift import Telegram

async def run_polling():
    async with Telegram.from_env() as bot:
        await bot.delete_webhook()  # при активном вебхуке getUpdates не работает
        offset = None
        while True:
            updates = await bot.get_updates(offset=offset, timeout=25)
            for update in updates:
                offset = update["update_id"] + 1
                msg = update.get("message") or {}
                if msg.get("text") == "/start":
                    await bot.send_message(msg["chat"]["id"], "Добро пожаловать!")

import os
from aiohttp import web
from socialift import Telegram
from socialift.webhooks import verify_telegram_request

SECRET = os.environ["TG_WEBHOOK_SECRET"]  # 1–256 символов: A-Z, a-z, 0-9, _ и -

async def setup(bot: Telegram) -> None:
    await bot.set_webhook("https://example.com/tg/webhook", secret_token=SECRET,
                          allowed_updates=["message", "callback_query"])
    info = await bot.get_webhook_info()
    print(info["pending_update_count"], info.get("last_error_message"))

async def handle(request: web.Request) -> web.Response:
    if not verify_telegram_request(request.headers, SECRET):
        return web.Response(status=403)
    update = await request.json()
    await queue.put(update)  # тяжёлую работу — в очередь, ответ не задерживаем
    return web.Response(text="ok")

async def send_catalog(bot, chat_id: int) -> None:
    inline = {"inline_keyboard": [[
        {"text": "Каталог", "url": "https://example.com/catalog"},
        {"text": "Оформить заказ", "callback_data": "order:42"},
    ]]}
    await bot.send_photo(chat_id, "https://example.com/banner.jpg",
                         caption="Новая коллекция", reply_markup=inline)

    menu = {"keyboard": [[{"text": "Контакты"}, {"text": "Корзина"}]],
            "resize_keyboard": True}
    await bot.send_message(chat_id, "Выберите раздел:", reply_markup=menu)

async def on_callback(bot, query: dict) -> None:
    await bot.answer_callback_query(query["id"], text="Принято")

import asyncio
import socialift

async def broadcast(bot, chat_ids: list[int], text: str) -> None:
    for chat_id in chat_ids:
        try:
            await bot.send_message(chat_id, text)
        except socialift.RateLimitError as exc:
            await asyncio.sleep(exc.retry_after or 5)
            await bot.send_message(chat_id, text)
        except socialift.PermissionError:
            mark_inactive(chat_id)  # пользователь заблокировал бота (403)
        await asyncio.sleep(0.05)  # ~20 сообщений/с — ниже порога в 30/с