Real-Time DonationAlerts Notifications with Python, OAuth & WebSockets 🚀

# .env
APP_ID="YOUR_APP_ID_HERE"
API_KEY="YOUR_API_KEY_HERE"
REDIRECT_URI="http://localhost:8000/api/auth/donationalerts/callback"
SESSION_SECRET_KEY="GENERATE_A_STRONG_RANDOM_KEY_HERE" # For session security

# Usually fixed
DA_SCOPES="oauth-user-show oauth-donation-subscribe"
DA_AUTHORIZATION_URL="https://www.donationalerts.com/oauth/authorize"
DA_TOKEN_URL="https://www.donationalerts.com/oauth/token"
DA_API_BASE_URL="https://www.donationalerts.com/api/v1"
# ... other URLs if needed

# main.py (Simplified Snippets)
import httpx
import secrets
from fastapi import FastAPI, Request, Depends
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware # For storing state

# Assume 'settings' is loaded from config.py or .env
# settings = load_app_config() # From our full example's config.py

app = FastAPI()

# IMPORTANT: SessionMiddleware is needed to store the OAuth state temporarily
app.add_middleware(SessionMiddleware, secret_key=settings["SESSION_SECRET_KEY"])

# --- Login Endpoint ---
@app.get("/api/auth/donationalerts/login")
async def login_donationalerts(request: Request):
    state = secrets.token_urlsafe(16)
    request.session['oauth_state'] = state # Store state to prevent CSRF

    params = {
        "client_id": settings["APP_ID"],
        "redirect_uri": settings["REDIRECT_URI"],
        "response_type": "code",
        "scope": settings["DA_SCOPES"],
        "state": state,
    }
    auth_request = httpx.Request('GET', settings["DA_AUTHORIZATION_URL"], params=params)
    authorization_url = str(auth_request.url)

    print(f"Redirecting to: {authorization_url}") # For debugging
    return RedirectResponse(authorization_url)

# --- Callback Endpoint ---
@app.get("/api/auth/donationalerts/callback")
async def auth_donationalerts_callback(request: Request, code: str = None, state: str = None, error: str = None):
    if error:
        # Handle authorization error from DonationAlerts
        return {"error": error}

    # --- Security Check: Verify State ---
    stored_state = request.session.pop('oauth_state', None)
    if not state or state != stored_state:
        # State mismatch, potential CSRF attack!
        return {"error": "Invalid state"}

    if not code:
        return {"error": "Missing authorization code"}

    # --- Exchange Code for Tokens ---
    token_data = await exchange_code_for_token(code) # See function below

    if not token_data:
        return {"error": "Failed to exchange code for token"}

    # --- Success! Store Tokens Securely ---
    # In a real app: save token_data['access_token'], token_data['refresh_token'],
    # and calculate token_data['expires_at'] in a database linked to the user.
    # For this example, we might use a simple file storage (like token_storage.py).
    # save_token(token_data) # From our full example's token_storage.py

    print("OAuth successful, tokens obtained!")
    # Redirect user back to the main page of your app
    return RedirectResponse(url="/?status=success", status_code=303)

# --- Helper: Token Exchange ---
async def exchange_code_for_token(code: str):
    data = {
        "grant_type": "authorization_code",
        "client_id": settings["APP_ID"],
        "client_secret": settings["API_KEY"],
        "code": code,
        "redirect_uri": settings["REDIRECT_URI"],
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(settings["DA_TOKEN_URL"], data=data)
            response.raise_for_status() # Check for HTTP errors
            print("Token exchange successful!")
            return response.json()
        except httpx.HTTPStatusError as e:
            print(f"Token exchange failed: {e.response.status_code} - {e.response.text}")
            return None
        except Exception as e:
            print(f"An error occurred during token exchange: {e}")
            return None

# donationalerts_client.py (Simplified Snippets)
import websockets
import json
import httpx

# Assume 'settings' is loaded, and we have a valid 'access_token'
# Also assume '_make_api_request' is a helper function using httpx and access_token

async def connect_and_listen():
    # --- Stage 1: Get User ID and Socket Token (HTTP API) ---
    try:
        user_info_response = await _make_api_request('GET', '/user/oauth', access_token)
        if not user_info_response or 'data' not in user_info_response:
             print("Failed to get user info")
             return
        user_id = user_info_response['data'].get('id')
        socket_token = user_info_response['data'].get('socket_connection_token')
        if not user_id or not socket_token:
            print("Missing user_id or socket_token in API response")
            return
    except Exception as e:
        print(f"Error getting user info/socket token: {e}")
        return

    print(f"Got User ID: {user_id}, Socket Token: {socket_token[:10]}...") # Don't log full token

    # --- Stage 2: Connect and Authenticate WebSocket ---
    websocket_url = settings["DA_CENTRIFUGO_URL"]
    try:
        async with websockets.connect(websocket_url) as ws:
            print(f"Connected to WebSocket: {websocket_url}")

            # --- Stage 2a: Send Socket Token ---
            # IMPORTANT: The format is specific! {"params": {"token": ...}, "id": 1}
            auth_payload = {"params": {"token": socket_token}, "id": 1}
            await ws.send(json.dumps(auth_payload))
            print("Sent WebSocket authentication request.")

            # --- Stage 2b: Receive Auth Response & Client ID ---
            auth_response_str = await ws.recv()
            auth_response = json.loads(auth_response_str)
            print(f"Received auth response: {auth_response}")

            client_id = None
            if auth_response.get("id") == 1 and auth_response.get("result", {}).get("client"):
                client_id = auth_response["result"]["client"]
                print(f"WebSocket authenticated! Client ID: {client_id}")
            else:
                print("WebSocket authentication failed!")
                return

            # --- Stage 3: Get Subscription Token (HTTP API) ---
            channel_name = f"$alerts:donation_{user_id}"
            sub_token_payload = {"client": client_id, "channels": [channel_name]}
            try:
                sub_response = await _make_api_request('POST', '/centrifuge/subscribe', access_token, json=sub_token_payload)
                if not sub_response or "channels" not in sub_response:
                     print("Failed to get subscription token response")
                     return

                subscription_token = None
                for channel_info in sub_response.get("channels", []):
                    if channel_info.get("channel") == channel_name and "token" in channel_info:
                        subscription_token = channel_info["token"]
                        break

                if not subscription_token:
                     print(f"Subscription token for {channel_name} not found in response.")
                     return

                print(f"Got Subscription Token: {subscription_token[:10]}...")
            except Exception as e:
                print(f"Error getting subscription token: {e}")
                return

            # --- Stage 4: Subscribe to Channel (WebSocket) ---
            subscribe_payload = {
                "id": 2, # Use a new ID for this request
                "method": 1, # 1 = subscribe
                "params": {"channel": channel_name, "token": subscription_token}
            }
            await ws.send(json.dumps(subscribe_payload))
            print(f"Sent subscribe request for channel {channel_name}")

            # --- Stage 5: Listen for Messages ---
            print("Waiting for donation messages...")
            async for message_str in ws:
                print(f"RAW MESSAGE RECEIVED: {message_str}")
                # Now parse the message (see next step)
                # handle_donation_message(message_str) 

    except websockets.exceptions.ConnectionClosedOK:
        print("WebSocket connection closed normally.")
    except Exception as e:
        print(f"WebSocket Error: {e}")

# --- Helper for API Requests (Example) ---
async def _make_api_request(method, endpoint, token, **kwargs):
     headers = {"Authorization": f"Bearer {token}"}
     url = f"{settings['DA_API_BASE_URL']}{endpoint}"
     async with httpx.AsyncClient() as client:
         response = await client.request(method, url, headers=headers, **kwargs)
         response.raise_for_status()
         return response.json()

{
  "result": {
    "channel": "$alerts:donation_YOUR_USER_ID",
    "data": {
      "seq": 15, // Sequence number from Centrifugo
      "data": {
        // ---> THIS is the actual donation object <---
        "id": 164405432, // Unique Donation ID! Important for deduplication.
        "name": "DonatorName", // Name entered by donator (fallback)
        "username": "DonatorDA_Username", // DA username if logged in, else null
        "message": "Your message here! \r\n Can contain newlines.",
        "message_type": "text",
        "amount": 22.0,
        "currency": "USD",
        "is_shown": 1, // If shown in standard DA widgets
        "amount_in_user_currency": 1500.0,
        "recipient_name": "YourStreamerName",
        "recipient": { // Info about the recipient (you)
            "user_id": YOUR_USER_ID,
            "code": "YourStreamerLogin",
            "name": "YourStreamerName",
            "avatar": "URL_to_avatar"
         },
        "created_at": "2025-04-12 07:15:49" // Server time
        // ... potentially other fields ...
      }
    }
  }
}

def handle_donation_message(message_str):
    try:
        message = json.loads(message_str)
        # Check if it's a push message (no 'id') and has the expected structure
        if "id" not in message and "result" in message:
            result_data = message.get("result", {})
            outer_data = result_data.get("data", {})
            donation_data = outer_data.get("data") # The actual donation payload

            if donation_data and result_data.get("channel", "").startswith("$alerts:donation_"):
                # Extract useful fields (use .get for safety)
                donation_id = donation_data.get("id")
                sender = donation_data.get("username") or donation_data.get("name", "Anonymous")
                amount = donation_data.get("amount")
                currency = donation_data.get("currency")
                text = donation_data.get("message", "")
                created_at = donation_data.get("created_at")

                print("-" * 20)
                print(f"🎉 New Donation! (ID: {donation_id})")
                print(f"   From: {sender}")
                print(f"   Amount: {amount} {currency}")
                print(f"   Message: {text.strip()}") # Strip whitespace/newlines
                print(f"   Time: {created_at}")
                print("-" * 20)

                # --- TODO: Send this data to your frontend or trigger actions! ---
                # Example: await websocket_manager.broadcast({"type": "donation", "data": donation_data})

            elif result_data.get("channel", "").startswith("$alerts:donation_"):
                 # Might be a subscribe confirmation or other message on the channel
                 print(f"Received non-donation push on channel: {message}")

        elif "id" in message:
             # This is likely an ack for one of our requests (id: 1 or 2) or a ping
             print(f"Received message with ID: {message}")
        else:
             print(f"Received unhandled message format: {message}")

    except json.JSONDecodeError:
        print(f"Error decoding JSON: {message_str}")
    except Exception as e:
        print(f"Error handling message: {e}")

# Inside the WebSocket listening loop:
# async for message_str in ws:
#     handle_donation_message(message_str)

// Simplified Frontend JS Example
let ws = new WebSocket(`ws://${window.location.host}/ws`); // Connect to your backend WS

ws.onmessage = function(event) {
    try {
        const message = JSON.parse(event.data);
        if (message.type === 'donation' && message.data) {
            // Function to add the donation to the page
            displayDonation(message.data); 
        }
    } catch (e) {
        console.error("Error processing message from backend:", e);
    }
};

function displayDonation(donation) {
    const log = document.getElementById('donations-log');
    const item = document.createElement('div');
    const sender = donation.username || donation.name || 'Anonymous';
    item.innerHTML = `${sender} donated ${donation.amount} ${donation.currency}: ${donation.message || ''}`;
    log.appendChild(item);
}