У меня было несколько ботов на python-telegram-bot: команды, inline-кнопки, состояния, работа с базой и внешними сервисами. Затем понадобился веб-интерфейс с тем же поведением.
Переносить обработчики в отдельный HTTP API означало бы поддерживать две реализации одного сценария. Встраивать Telegram Web App тоже не подходило: пользователь должен был работать в обычном браузере, без Telegram.
Я пошёл другим путём: браузер притворяется Telegram на входе, а локальный transport-слой — Telegram Bot API на выходе. Исходные обработчики получают настоящие объекты Update и вызываются через штатный Application.process_update(). В этой статье покажу, как устроен прототип, на каких границах он держится и что сломалось после первых перезапусков.
Скриншот минимального демонстрационного бота. Интерфейс работает через реальный адаптер; 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_request async 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. Он понимает только операции, необходимые подключённым ботам, а на остальные отвечает явной ошибкой. Это лучше, чем молча проглотить вызов и показать пользователю неправдоподобный результат.
Вход идёт через настоящий 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 ничего не прислал, интерфейс показывает понятное сообщение: кнопка устарела, откройте актуальное меню. Молчаливый клик оказался хуже явной деградации.
Браузерная история и память процесса живут разное время — это пришлось учесть отдельно.
Третий фейл: совпавшие 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 messages assert "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-бот открылся в браузере. Полезнее оказался сам способ искать границу интеграции:
На входе создать объект, который уже понимает существующий application layer.
На выходе перехватить транспорт до реальной сети.
Не дублировать обработчики и состояние.
Явно падать на неподдержанной функции.
Отдельно тестировать несовпадающие жизненные циклы браузера и серверного процесса.
Такой адаптер подходит для локальных демо, миграции интерфейса и постепенного отделения бизнес-логики от Telegram. Но это не универсальная замена Telegram-клиенту — и чем раньше провести эту границу, тем меньше неожиданных обещаний появится у проекта.
Если вы переносили существующего бота в другой канал, где проводили границу: на уровне Update, собственных команд приложения или полностью выносили сценарии из фреймворка?
