How to Automate Twitch Charity Donation Alerts using the API

1import asyncio
2import json
3import os
4import time
5import logging
6import requests
7import websockets
8
9logging.basicConfig(level=logging.INFO)
10logger = logging.getLogger(__name__)
11
12CLIENT_ID = os.environ['TWITCH_CLIENT_ID']
13ACCESS_TOKEN = os.environ['BROADCASTER_ACCESS_TOKEN']
14BROADCASTER_ID = os.environ['BROADCASTER_ID']
15DISCORD_WEBHOOK_URL = os.environ.get('DISCORD_WEBHOOK_URL', '')
16LOG_FILE = 'charity_donations.json'
17
18def hdrs(json_content=False):
19    h = {'Authorization': f'Bearer {ACCESS_TOKEN}', 'Client-Id': CLIENT_ID}
20    if json_content:
21        h['Content-Type'] = 'application/json'
22    return h
23
24def fmt_amount(value, decimal_places, currency):
25    return f'{currency} {value / (10 ** decimal_places):,.2f}'
26
27def log_donation(donation):
28    donations = []
29    try:
30        with open(LOG_FILE, 'r') as f:
31            donations = json.load(f)
32    except FileNotFoundError:
33        pass
34    donations.append({'timestamp': time.time(), **donation})
35    with open(LOG_FILE, 'w') as f:
36        json.dump(donations, f, indent=2)
37
38def get_campaign():
39    resp = requests.get('https://api.twitch.tv/helix/charity/campaigns',
40                        headers=hdrs(), params={'broadcaster_id': BROADCASTER_ID})
41    resp.raise_for_status()
42    data = resp.json()['data']
43    return data[0] if data else None
44
45async def on_donation(donation):
46    donor = donation.get('user_name') or 'Anonymous'
47    amount_str = fmt_amount(donation['amount']['value'],
48                            donation['amount']['decimal_places'],
49                            donation['amount']['currency'])
50    current_str = fmt_amount(donation['campaign_current_amount']['value'],
51                             donation['campaign_current_amount']['decimal_places'],
52                             donation['campaign_current_amount']['currency'])
53    target_str = fmt_amount(donation['campaign_target_amount']['value'],
54                            donation['campaign_target_amount']['decimal_places'],
55                            donation['campaign_target_amount']['currency'])
56    charity = donation.get('charity_name', 'the charity')
57    broadcaster = donation.get('broadcaster_user_name', '')
58    desc = (f'**{donor}** donated **{amount_str}** to {charity}!\n'
59            f'Total raised: **{current_str} / {target_str}**')
60    logger.info(f'Donation: {desc}')
61    log_donation(donation)
62    if DISCORD_WEBHOOK_URL:
63        requests.post(DISCORD_WEBHOOK_URL, json={'embeds': [{
64            'title': ':heart: Charity Donation!',
65            'description': desc,
66            'color': 16711680,
67            'footer': {'text': f'Donate at twitch.tv/{broadcaster.lower()}'}
68        }]})
69
70async def run():
71    campaign = get_campaign()
72    if not campaign:
73        logger.warning('No active charity campaign — waiting for one to start')
74    else:
75        logger.info(f'Campaign: {campaign["charity_name"]}')
76    while True:
77        try:
78            async with websockets.connect('wss://eventsub.wss.twitch.tv/ws') as ws:
79                welcome = json.loads(await ws.recv())
80                session_id = welcome['payload']['session']['id']
81                requests.post(
82                    'https://api.twitch.tv/helix/eventsub/subscriptions',
83                    headers=hdrs(True),
84                    json={'type': 'channel.charity.campaign.donate', 'version': '1',
85                          'condition': {'broadcaster_user_id': BROADCASTER_ID},
86                          'transport': {'method': 'websocket', 'session_id': session_id}}
87                ).raise_for_status()
88                logger.info('Listening for charity donations...')
89                async for raw in ws:
90                    ev = json.loads(raw)
91                    if ev.get('metadata', {}).get('message_type') == 'notification':
92                        d = ev['payload']['event']
93                        await on_donation(d)
94        except Exception as e:
95            logger.error(f'Connection error: {e}. Reconnecting in 5s')
96            await asyncio.sleep(5)
97
98asyncio.run(run())