SavePearlHarbor

Как я запустил Telegram-бота в браузере, не переписывая его обработчики

—

от автора

admin

У меня было несколько ботов на python-telegram-bot: команды, inline-кнопки, состояния, работа с базой и внешними сервисами. Затем понадобился веб-интерфейс с тем же поведением.

Переносить обработчики в отдельный HTTP API означало бы поддерживать две реализации одного сценария. Встраивать Telegram Web App тоже не подходило: пользователь должен был работать в обычном браузере, без Telegram.

Я пошёл другим путём: браузер притворяется Telegram на входе, а локальный transport-слой — Telegram Bot API на выходе. Исходные обработчики получают настоящие объекты Update и вызываются через штатный Application.process_update(). В этой статье покажу, как устроен прототип, на каких границах он держится и что сломалось после первых перезапусков.

Демонстрационный PTB-бот в веб-интерфейсе

Демонстрационный PTB-бот в веб-интерфейсе

Скриншот минимального демонстрационного бота. Интерфейс работает через реальный адаптер; Telegram в этом сценарии не участвует.

Что хотелось сохранить

Исходный бот уже умел:

  • обрабатывать команды и обычный текст;

  • показывать inline-клавиатуры;

  • отвечать на CallbackQuery;

  • редактировать и удалять сообщения;

  • отправлять фотографии и документы;

  • хранить прогресс пользователя в своей базе.

Поэтому главный критерий был таким: обработчики, репозитории и сервисы исходного бота менять нельзя. Веб-слой должен быть транспортом, а не второй бизнес-логикой.

Это важное ограничение. Если веб-обработчик отдельно повторяет start(), choose_language() и confirm(), две версии неизбежно разойдутся. Ошибка может проявиться не сразу, а после следующего изменения сценария.

Где войти в python-telegram-bot

У python-telegram-bot есть удобная точка входа — Application.process_update(). Метод принимает один Update, прогоняет его через зарегистрированные handlers и передаёт исключения error handlers.

Значит, для входящего сообщения достаточно собрать JSON той же формы, которую прислал бы Telegram, затем создать объект через Update.de_json():

update_data = {    "update_id": self._next_update,    "message": {        "message_id": self._next_message,        "date": int(datetime.now(timezone.utc).timestamp()),        "chat": {"id": user_id, "type": "private"},        "from": {            "id": user_id,            "is_bot": False,            "first_name": "Web",        },        "text": text,    },}update = Update.de_json(update_data, application.bot)await application.process_update(update)

Для команды я дополнительно создаю entity типа bot_command. Для кнопки — объект callback_query с тем же callback_data, которое исходный бот положил в InlineKeyboardButton.

update_data = {    "update_id": self._next_update,    "callback_query": {        "id": str(uuid4()),        "from": user,        "chat_instance": str(user_id),        "data": str(payload["value"]),        "message": message,    },}

Для обработчика это обычный Telegram update. CommandHandler, MessageHandler и CallbackQueryHandler продолжают работать без веб-веток внутри исходного проекта.

Вторая половина задачи: перехватить ответы

Одного process_update() мало. Обработчик почти сразу вызовет reply_text(), edit_message_text() или send_photo(). По умолчанию PTB отправит HTTP-запрос настоящему Bot API.

В моём прототипе перехватывается HTTPXRequest.do_request — точка, через которую созданные исходным ботом объекты Bot уходят в сеть:

def install_bot_api() -> None:    HTTPXRequest.do_request = _bot_api_requestasync def _bot_api_request(self, url, method, request_data=None, **kwargs):    action = urlsplit(url).path.rsplit("/", 1)[-1].lower()    if action == "getme":        return ok({            "id": 123456,            "is_bot": True,            "first_name": "Web Bot",            "username": "local_web_bot",        })    session = active_session.get()    if session is None:        raise RuntimeError(            f"Bot API {action} outside an active web conversation is disabled"        )    if action in {"sendmessage", "sendphoto", "senddocument"}:        return ok(session.send(action, request_data.parameters, request_data))    if action in {        "editmessagetext",        "editmessagecaption",        "editmessagereplymarkup",    }:        return ok(session.edit(action, request_data.parameters))    raise RuntimeError(f"Bot API method {action} is not implemented")

Transport не пытается реализовать весь Bot API. Он понимает только операции, необходимые подключённым ботам, а на остальные отвечает явной ошибкой. Это лучше, чем молча проглотить вызов и показать пользователю неправдоподобный результат.

Поток события через веб-адаптер и исходный PTB-бот

Поток события через веб-адаптер и исходный PTB-бот

Вход идёт через настоящий Update, выход перехватывается на HTTP-границе PTB.

Зачем здесь ContextVar

После перехвата появляется вопрос: в какую веб-сессию положить ответ? Глобальная переменная работает ровно до двух одновременных запросов.

Я использовал ContextVar:

_active: ContextVar[PTBSession | None] = ContextVar(    "telegram_web_session",    default=None,)marker = _active.set(self)try:    await application.process_update(update)finally:    _active.reset(marker)

Значение привязано к текущему асинхронному контексту. Когда handler вызывает Bot API, transport достаёт именно ту PTBSession, из которой пришло событие.

У каждой сессии также есть asyncio.Lock. Он сохраняет порядок событий внутри одного диалога: двойной клик не должен одновременно провести state machine через два перехода.

Это не делает весь бот автоматически потокобезопасным. Если исходный проект запускает фоновые задачи или использует concurrent_updates, их поведение нужно проверять отдельно.

Как Telegram-клавиатура превращается в кнопки браузера

reply_markup приходит в параметрах исходящего запроса. Адаптер читает inline_keyboard и превращает кнопки в простой контракт фронтенда:

def keyboard(markup):    if isinstance(markup, str):        markup = json.loads(markup)    return [        {            "label": button.get("text", ""),            "value": button.get("callback_data") or button.get("text", ""),            "row": row_index,        }        for row_index, row in enumerate(markup.get("inline_keyboard", []))        for button in row    ]

При клике браузер возвращает value, transport создаёт CallbackQuery, и исходный CallbackQueryHandler получает знакомый update.callback_query.data.

Ссылочные, платёжные, Web App и другие типы кнопок этот фрагмент не поддерживает. Для моего набора ботов были нужны только callback-кнопки.

Первый фейл: «новый диалог» создавал нового человека

Изначально идентификатор Telegram-пользователя вычислялся из conversationId. В интерфейсе кнопка «Новый диалог» создавала новый разговор — и исходный бот видел нового пользователя. Его прогресс в базе внезапно исчезал.

Правильной границей оказался постоянный ID браузерного пользователя:

def web_user_id(value: str) -> int:    digest = hashlib.sha256(value.encode("utf-8")).digest()    return int.from_bytes(digest[:8], "big") & ((1 << 63) - 1)

conversationId теперь выбирает историю интерфейса, а userId — запись пользователя в исходном боте. Новый диалог отправляет тому же пользователю /start; уже сам бот решает, продолжить сценарий или предложить сброс.

Второй фейл: кнопка пережила сервер, сообщение — нет

История браузера сохраняется дольше процесса адаптера. После рестарта пользователь мог нажать старую кнопку, но новая PTBSession не знала message_id сообщения, которое handler собирался отредактировать.

Сначала это заканчивалось Unknown Telegram message. Затем я добавил восстановление минимальной цели для callback:

def callback_target(self, label: str) -> int:    if self._last_bot_id is not None:        return self._last_bot_id    message_id = self._next_message    self._next_message += 1    self._last_bot_id = message_id    self.messages.append({        "id": str(message_id),        "type": "text",        "text": label,    })    return message_id

Это не восстанавливает Telegram-сообщение во всей полноте. Цель решения скромнее: дать исходному callback-handler отработать и вернуть актуальное состояние вместо 500-й ошибки.

Если handler ничего не прислал, интерфейс показывает понятное сообщение: кнопка устарела, откройте актуальное меню. Молчаливый клик оказался хуже явной деградации.

Что происходит со старой inline-кнопкой после рестарта

Что происходит со старой inline-кнопкой после рестарта

Браузерная история и память процесса живут разное время — это пришлось учесть отдельно.

Третий фейл: совпавшие message_id затирали историю

В первой версии счётчик исходящих сообщений начинался с единицы при каждом запуске. Браузер уже хранил сообщения с ID 1, 2, 3, а новый процесс выдавал те же значения. Фронтенд принимал новый ответ за обновление старого.

Для локального прототипа я выбрал случайную стартовую точку в диапазоне Telegram-подобных ID:

self._next_message = uuid4().int % 2_000_000_000 + 1

Коллизия стала практически невероятной, но это всё ещё не строгая гарантия. В production-версии счётчик нужно хранить устойчиво или перейти на собственный составной ID транспорта.

Что делать с фоновыми уведомлениями

Есть неприятный класс ошибок: handler обрабатывает одного пользователя, а фоновая задача отправляет сообщение другому. Если transport просто положит любой sendMessage в активную вкладку, получится утечка данных между пользователями.

Поэтому адаптер проверяет адресата:

chat_id = int(params["chat_id"])if chat_id != self._current_user:    raise RuntimeError(        "Outgoing message targets a different Telegram chat"    )

Фоновые уведомления требуют отдельного маршрута доставки. В одном из профилей я сделал для них собственный endpoint с cursor-поллингом. Главное — не «удобно» переадресовывать их в текущую сессию.

Проверка без Telegram

Минимальный тест создаёт приложение PTB, отправляет событие в PTBSession и проверяет результат исходного handler:

messages, buttons = await session.dispatch(    application,    {        "type": "command",        "payload": {"value": "start"},    },    user_id,)assert messagesassert "lang_ru" in [button["value"] for button in buttons]

В проекте отдельно проверяются:

  • команда, обычный текст и callback;

  • редактирование сообщения и удаление клавиатуры;

  • стабильный пользователь между веб-диалогами;

  • устаревшая кнопка после рестарта;

  • отсутствие ответа у callback-handler;

  • выдача фотографий через локальный media endpoint.

На момент подготовки статьи прошли 33 теста веб-консоли и два автономных теста transport-слоя. Интеграционные тесты конкретного внешнего бота требуют его базы и сервисов, поэтому я не считаю их частью самодостаточной проверки адаптера.

Почему я пока не называю это библиотекой

Прототип решает мою задачу, но у него есть границы:

  • он рассчитан на python-telegram-bot 20.7;

  • глобальная подмена HTTPXRequest.do_request хрупкая и влияет на весь процесс;

  • реализована только используемая часть Bot API;

  • входящие файлы из браузера пока не превращаются в Telegram media objects;

  • состояние веб-сессий хранится в памяти;

  • _display() удаляет HTML регулярным выражением — годится для контролируемого текста, но не является HTML sanitizer;

  • адаптер слушает только 127.0.0.1 и сам по себе не решает аутентификацию публичного сервиса.

Следующий шаг — реализовать собственный наследник BaseRequest и передавать его через builder там, где я контролирую создание Application. Это уберёт глобальный monkey-patch. Для чужих ботов, которые создают Bot глубоко внутри конструктора, останется отдельный режим совместимости.

Также нужны контрактные тесты на каждую поддержанную операцию Bot API и устойчивое хранилище сессий. Только после этого проект можно предлагать как переиспользуемый пакет.

Итог

Главный результат для меня не в том, что Telegram-бот открылся в браузере. Полезнее оказался сам способ искать границу интеграции:

  1. На входе создать объект, который уже понимает существующий application layer.

  2. На выходе перехватить транспорт до реальной сети.

  3. Не дублировать обработчики и состояние.

  4. Явно падать на неподдержанной функции.

  5. Отдельно тестировать несовпадающие жизненные циклы браузера и серверного процесса.

Такой адаптер подходит для локальных демо, миграции интерфейса и постепенного отделения бизнес-логики от Telegram. Но это не универсальная замена Telegram-клиенту — и чем раньше провести эту границу, тем меньше неожиданных обещаний появится у проекта.

Если вы переносили существующего бота в другой канал, где проводили границу: на уровне Update, собственных команд приложения или полностью выносили сценарии из фреймворка?

Ссылки

ссылка на оригинал статьи https://habr.com/ru/articles/1092800/