40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
from typing import Annotated
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
from ..main import QBotApp
|
|
|
|
|
|
from ..db import get_db
|
|
from aiogram.types import Update
|
|
from logging import getLogger
|
|
|
|
|
|
logger = getLogger(__name__)
|
|
logger.setLevel("DEBUG")
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/webhook")
|
|
async def telegram_webhook(
|
|
db_session: Annotated[AsyncSession, Depends(get_db)], request: Request
|
|
):
|
|
logger.debug("Webhook request %s", await request.json())
|
|
app: QBotApp = request.app
|
|
|
|
request_token = request.headers.get("X-Telegram-Bot-Api-Secret-Token")
|
|
if request_token != app.bot_auth_token:
|
|
logger.warning("Unauthorized request %s", request)
|
|
return Response(status_code=403)
|
|
try:
|
|
update = Update(**await request.json())
|
|
except Exception:
|
|
logger.error("Invalid request", exc_info=True)
|
|
return Response(status_code=400)
|
|
try:
|
|
await app.dp.feed_webhook_update(
|
|
app.bot, update, db_session=db_session, app=app
|
|
)
|
|
except Exception:
|
|
logger.error("Error processing update", exc_info=True)
|
|
return Response(status_code=200)
|