How to Automate Twitch Subscriber Notifications using the API

1import websocket
2import requests
3import json, threading, os, time
4from datetime import datetime
5
6CLIENT_ID = os.environ['TWITCH_CLIENT_ID']
7ACCESS_TOKEN = os.environ['TWITCH_ACCESS_TOKEN']
8BROADCASTER_ID = os.environ['TWITCH_BROADCASTER_ID']
9DISCORD_WEBHOOK = os.environ.get('DISCORD_WEBHOOK_URL')
10
11HEADERS = {'Authorization': f'Bearer {ACCESS_TOKEN}', 'Client-Id': CLIENT_ID}
12TIER = {'1000': 'Tier 1', '2000': 'Tier 2', '3000': 'Tier 3'}
13SUB_TYPES = ['channel.subscribe', 'channel.subscription.gift', 'channel.subscription.message']
14
15def validate_token():
16    r = requests.get('https://id.twitch.tv/oauth2/validate', headers={'Authorization': f'OAuth {ACCESS_TOKEN}'})
17    if r.status_code != 200: raise Exception('Token invalid')
18    print('Token validated')
19
20def subscribe_eventsub(session_id):
21    for stype in SUB_TYPES:
22        r = requests.post('https://api.twitch.tv/helix/eventsub/subscriptions',
23            headers={**HEADERS, 'Content-Type': 'application/json'},
24            json={'type': stype, 'version': '1',
25                  'condition': {'broadcaster_user_id': BROADCASTER_ID},
26                  'transport': {'method': 'websocket', 'session_id': session_id}})
27        print(f'Subscribed {stype}: {r.status_code}')
28
29def send_alert(title, desc):
30    if DISCORD_WEBHOOK:
31        requests.post(DISCORD_WEBHOOK, json={'embeds': [{'title': title, 'description': desc, 'color': 9520895}]})
32
33def handle_event(sub_type, event):
34    if sub_type == 'channel.subscribe':
35        tier = TIER.get(event.get('tier', '1000'), 'Tier 1')
36        if not event.get('is_gift'):
37            send_alert('⭐ New Subscriber!', f"{event['user_name']} just subscribed at {tier}!")
38    elif sub_type == 'channel.subscription.gift':
39        n = event.get('total', 1)
40        gifter = 'Anonymous' if event.get('is_anonymous') else event.get('gifter_user_name', '?')
41        tier = TIER.get(event.get('tier', '1000'), 'Tier 1')
42        send_alert('🎁 Gift Subs!', f"{gifter} gifted {n}x {tier} sub{'s' if n>1 else ''}!")
43    elif sub_type == 'channel.subscription.message':
44        months = event.get('cumulative_months', 1)
45        tier = TIER.get(event.get('tier', '1000'), 'Tier 1')
46        msg = event.get('message', {}).get('text', '')
47        desc = f"{event['user_name']} resubbed for {months} months ({tier})!"
48        if msg: desc += f' "{msg[:100]}"'
49        send_alert('🔄 Resub!', desc)
50
51def on_message(ws, message):
52    data = json.loads(message)
53    mtype = data.get('metadata', {}).get('message_type')
54    if mtype == 'session_welcome':
55        sid = data['payload']['session']['id']
56        threading.Thread(target=subscribe_eventsub, args=(sid,), daemon=True).start()
57    elif mtype == 'notification':
58        handle_event(data['metadata']['subscription_type'], data['payload']['event'])
59    elif mtype == 'session_keepalive':
60        pass  # heartbeat, no action needed
61
62def on_error(ws, error):
63    print(f'WebSocket error: {error}')
64
65def on_close(ws, code, reason):
66    print(f'Disconnected: {code} {reason}. Reconnecting in 5s...')
67    time.sleep(5)
68    start_bot()
69
70def start_bot():
71    validate_token()
72    ws = websocket.WebSocketApp('wss://eventsub.wss.twitch.tv/ws',
73        on_message=on_message, on_error=on_error, on_close=on_close)
74    ws.run_forever(ping_interval=30, ping_timeout=10)
75
76if __name__ == '__main__':
77    start_bot()