Как я запустил Telegram-бота в браузере, не переписывая его обработчики
У меня было несколько ботов на 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, собственных команд приложения или полностью выносили сценарии из фреймворка?
Ссылки
Если эта публикация вас вдохновила и вы хотите поддержать автора — не стесняйтесь нажать на кнопку
KioskNews shows a cleaned-up reading view extracted from the publisher’s page — the original always lives on their site, not ours.