How to promote group admins in Python with the Wapito WhatsApp API

"""Group Admins with the Wapito WhatsApp API.

Read the current roles, promote from your own source of truth, demote what is stale, and follow role changes on the webhook.

Run it with:
    export WAPITO_TOKEN="wpt_..."
    python group-admins.py
"""
import os

import requests

BASE_URL = "https://api.wapito.com/v1"
TOKEN = os.environ["WAPITO_TOKEN"]

# --- See who holds a role today ---
url = BASE_URL + "/groups/120363041234567890@g.us/participants"

headers = {"Authorization": "Bearer " + TOKEN}

response = requests.get(url, headers=headers)

print(response.json())

# --- Promote the people who should moderate ---
url = BASE_URL + "/groups/120363041234567890@g.us/admins"

payload = { "participants": ["+15551234567"] }
headers = {
    "Authorization": "Bearer " + TOKEN,
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)

print(response.json())

# --- Demote anyone who no longer needs it ---
url = BASE_URL + "/groups/120363041234567890@g.us/admins/+15551234567"

headers = {"Authorization": "Bearer " + TOKEN}

response = requests.delete(url, headers=headers)

print(response.status_code)

import hashlib
import hmac
import os
import time

from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()
SECRET = os.environ["WAPITO_WEBHOOK_SECRET"].encode()
EVENTS = ["groups.participants"]


def verify(raw: bytes, header: str | None) -> bool:
    """Checks the X-Wapito-Signature header: t=<ms>,v1=<hex hmac-sha256>."""
    if not header:
        return False
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    ts, sig = parts.get("t"), parts.get("v1")
    if not ts or not sig:
        return False
    if abs(time.time() * 1000 - int(ts)) > 300_000:  # 5 minute clock skew
        return False
    expected = hmac.new(SECRET, ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, sig)


@app.post("/wapito")
async def wapito(request: Request, signature: str | None = Header(default=None, alias="X-Wapito-Signature")):
    raw = await request.body()
    if not verify(raw, signature):
        raise HTTPException(status_code=401, detail="bad signature")
    payload = await request.json()
    if payload["event"] in EVENTS:
        print(payload["event"], payload["data"])
    return {"ok": True}  # answer 2xx fast; do the real work in a queue