Приём вебхуков с помощью SDK

import asyncio
import json
import os
from fastapi import FastAPI, Request, Response
from socialift.webhooks import parse_events, verify_challenge, verify_request

app = FastAPI()
events: asyncio.Queue = asyncio.Queue()   # в продакшене: Redis, RQ, Celery и т. п.


@app.get("/webhooks/meta")
async def meta_verify(request: Request):
    challenge = verify_challenge(dict(request.query_params), os.environ["META_VERIFY_TOKEN"])
    if challenge is None:
        return Response(status_code=403)
    return Response(challenge, media_type="text/plain")   # не JSON, а именно строка


@app.post("/webhooks/meta")
async def meta_receive(request: Request):
    body = await request.body()                           # сырые байты
    if not verify_request(os.environ["FACEBOOK_APP_SECRET"], body, request.headers):
        return Response(status_code=403)
    for event in parse_events(json.loads(body)):
        events.put_nowait(event)
    return Response(status_code=200)

import json
import os
from fastapi import FastAPI, Request, Response
from socialift.webhooks import parse_tiktok_event, verify_tiktok_request

app = FastAPI()


@app.post("/webhooks/tiktok")
async def tiktok_receive(request: Request):
    body = await request.body()
    if not verify_tiktok_request(os.environ["TIKTOK_CLIENT_SECRET"], body, request.headers):
        return Response(status_code=403)
    event = parse_tiktok_event(json.loads(body))
    if event.event == "authorization.removed":
        print("Доступ отозван:", event.user_openid)   # удалите токены пользователя
    return Response(status_code=200)

import os
from fastapi import FastAPI, Request, Response
from socialift import Telegram
from socialift.webhooks import verify_telegram_request

app = FastAPI()
SECRET = os.environ["TELEGRAM_WEBHOOK_SECRET"]    # 1–256 символов: A-Z, a-z, 0-9, _ и -


async def register_webhook():                      # один раз при деплое
    async with Telegram.from_env() as bot:
        await bot.set_webhook("https://example.uz/webhooks/telegram",
                              secret_token=SECRET, allowed_updates=["message"])


@app.post("/webhooks/telegram")
async def telegram_receive(request: Request):
    if not verify_telegram_request(request.headers, SECRET):
        return Response(status_code=403)
    update = await request.json()
    print(update.get("message", {}).get("text"))     # поставьте в очередь
    return Response(status_code=200)

import asyncio
from socialift.webhooks import WebhookEvent, dedupe_key

seen: set[str] = set()   # в продакшене: UNIQUE-колонка в БД или Redis SET NX


async def worker(events: asyncio.Queue):
    while True:
        event: WebhookEvent = await events.get()
        key = dedupe_key(event)                  # то же, что event.key
        if key not in seen:
            seen.add(key)
            if event.field == "leadgen" and event.leadgen_id:
                print("Новый лид:", event.leadgen_id, "форма:", event.form_id)
        events.task_done()