Подключения Telegram-бота через webhook

application.run_webhook(
    listen='0.0.0.0',
    port=8443,
    secret_token='ASecretTokenIHaveChangedByNow',
    key='private.key',
    cert='cert.pem',
    webhook_url='https://example.com:8443'
)

# или

updater.start_webhook(
    listen='0.0.0.0',
    port=8443,
    secret_token='ASecretTokenIHaveChangedByNow',
    key='private.key',
    cert='cert.pem',
    webhook_url='https://example.com:8443'
)

import os

TOKEN = "TOKEN"
PORT = int(os.environ.get('PORT', '8443'))
# добавим обработчики
application.run_webhook(
    listen="0.0.0.0",
    port=PORT,
    secret_token='ASecretTokenIHaveChangedByNow',
    webhook_url="https://<appname>.herokuapp.com/"
)

application.run_webhook(
    listen='127.0.0.1',
    port=5000,
    url_path='1',
    secret_token='ASecretTokenIHaveChangedByNow',
    webhook_url='https://example.com/1',
    cert='cert.pem'
)

server {
    listen              443 ssl;
    server_name         example.com;
    ssl_certificate     cert.pem;
    ssl_certificate_key private.key;

    location /TOKEN1 {
        proxy_pass http://127.0.0.1:5000/1/;
    }

    location /TOKEN2 {
        proxy_pass http://127.0.0.1:5001/2/;
    }
}

application.run_webhook(
    listen='127.0.0.1',
    port=5000,
    secret_token='ASecretTokenIHaveChangedByNow',
    webhook_url='https://bot1.example.com',
    cert='cert_bot1.pem')
)

frontend  public-https
    bind        0.0.0.0:443 ssl crt cert_key_bot1.pem crt cert_key_bot2.pem
    option      httpclose

    use_backend bot1 if  { ssl_fc_sni bot1.example.com }
    use_backend bot2 if  { ssl_fc_sni bot2.example.com }

backend bot1
    mode            http
    option          redispatch
    server          bot1.example.com 127.0.0.1:5000 check inter 1000

backend bot2
    mode            http
    option          redispatch
    server          bot2.example.com 127.0.0.1:5001 check inter 1000

from telegram import Bot
from telegram.ext import Application

application = Application.builder().token('TOKEN').build()
    
# Регистрация обработчиков здесь

# Получаем `update_queue`, из которого приложение получает обновления для обработки.
update_queue = application.update_queue
start_fetching_updates(update_queue)

# Запускаем приложение
async with application:
    application.start()
    # и останавливаем, когда срабатывает 
    # какой-либо механизм отключения:
    application.stop()

import asyncio
import html
import logging
from dataclasses import dataclass
from http import HTTPStatus

import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import PlainTextResponse, Response
from starlette.routing import Route

from telegram import __version__ as TG_VER

try:
    from telegram import __version_info__
except ImportError:
    __version_info__ = (0, 0, 0, 0, 0)  # type: ignore[assignment]

if __version_info__ < (20, 0, 0, "alpha", 1):
    raise RuntimeError(
        f"Пример несовместим с текущей версией PTB {TG_VER}. "
        f"Чтобы просмотреть {TG_VER} версия этого примера, "
        f"посетите https://docs.python-telegram-bot.org/en/v{TG_VER}/examples.html"
    )

from telegram import Update
from telegram.constants import ParseMode
from telegram.ext import (
    Application,
    CallbackContext,
    CommandHandler,
    ContextTypes,
    ExtBot,
    TypeHandler,
)

# Enable logging
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)


@dataclass
class WebhookUpdate:
    """Класс данных для переноса пользовательского типа обновления"""

    user_id: int
    payload: str


class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
    """
    Пользовательский класс CallbackContext, который делает 
    user_data доступным для обновлений типа `WebhookUpdate`.
    """

    @classmethod
    def from_update(
        cls,
        update: object,
        application: "Application",
    ) -> "CustomContext":
        if isinstance(update, WebhookUpdate):
            return cls(application=application, user_id=update.user_id)
        return super().from_update(update, application)


async def start(update: Update, context: CustomContext) -> None:
    """Показать сообщение с инструкциями по использованию этого бота."""
    url = context.bot_data["url"]
    payload_url = html.escape(f"{url}/submitpayload?user_id=<your user id>&payload=<payload>")
    text = (
        f"Чтобы проверить, все ли еще запущен бот, вызовите <code>{url}/healthcheck</code>.\n\n"
        f"Чтобы опубликовать пользовательское обновление, вызовите <code>{payload_url}</code>."
    )
    await update.message.reply_html(text=text)


async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
    """Обратный вызов, который обрабатывает пользовательские обновления."""
    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
    payloads = context.user_data.setdefault("payloads", [])
    payloads.append(update.payload)
    combined_payloads = "</code>\n• <code>".join(payloads)
    text = (
        f"Пользователь {chat_member.user.mention_html()} отправил новую полезную нагрузку. "
        f"На данный момент они отправили следующее: \n\n• <code>{combined_payloads}</code>"
    )
    await context.bot.send_message(
        chat_id=context.bot_data["admin_chat_id"], text=text, parse_mode=ParseMode.HTML
    )


async def main() -> None:
    """Настройки приложения и пользовательского веб-сервера."""
    url = "https://domain.tld"
    admin_chat_id = 123456
    port = 8000

    context_types = ContextTypes(context=CustomContext)
    # Здесь устанавливаем для `updater` значение `None`, потому что нужно, чтобы пользовательский
    # сервер обрабатывал обновления, и, следовательно, не нужен экземпляр `Updater`.
    application = (
        Application.builder().token("TOKEN").updater(None).context_types(context_types).build()
    )
    # сохраним значения в `bot_data` таким образом, чтобы можно было легко получить 
    # к ним доступ при обратных вызовах
    application.bot_data["url"] = url
    application.bot_data["admin_chat_id"] = admin_chat_id

    # регистрируем обработчики
    application.add_handler(CommandHandler("start", start))
    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))

    # Передать настройки вебхука в телеграмм
    await application.bot.set_webhook(url=f"{url}/telegram")

    # Поднимаем веб-сервер
    async def telegram(request: Request) -> Response:
        """Обрабатываем входящие обновления Telegram, помещая их в `update_queue`"""
        await application.update_queue.put(
            Update.de_json(data=await request.json(), bot=application.bot)
        )
        return Response()

    async def custom_updates(request: Request) -> PlainTextResponse:
        """
        Обрабатываем входящие обновления веб-перехватчика, также помещая их в очередь 
       `update_queue`, если необходимые параметры были переданы правильно.
        """
        try:
            user_id = int(request.query_params["user_id"])
            payload = request.query_params["payload"]
        except KeyError:
            return PlainTextResponse(
                status_code=HTTPStatus.BAD_REQUEST,
                content="Передайте как `user_id`, так и полезную "
                        "нагрузку в качестве параметров запроса.",
            )
        except ValueError:
            return PlainTextResponse(
                status_code=HTTPStatus.BAD_REQUEST,
                content="`user_id` должен быть строкой!",
            )

        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
        return PlainTextResponse("Спасибо за отправку! Оно пересылается.")

    async def health(_: Request) -> PlainTextResponse:
        """Для конечной точки работоспособности отвечаем текстовым сообщением."""
        return PlainTextResponse(content="Бот по-прежнему работает нормально :)")

    starlette_app = Starlette(
        routes=[
            Route("/telegram", telegram, methods=["POST"]),
            Route("/healthcheck", health, methods=["GET"]),
            Route("/submitpayload", custom_updates, methods=["POST", "GET"]),
        ]
    )
    webserver = uvicorn.Server(
        config=uvicorn.Config(
            app=starlette_app,
            port=port,
            use_colors=False,
            host="127.0.0.1",
        )
    )

    # Запуск приложения и веб-сервера вместе
    async with application:
        await application.start()
        await webserver.serve()
        await application.stop()


if __name__ == "__main__":
    asyncio.run(main())