How to Automate Twitch Stream Alerts using the API

1import os
2import hmac
3import hashlib
4import time
5import logging
6import requests
7from flask import Flask, request, Response
8
9logging.basicConfig(level=logging.INFO)
10logger = logging.getLogger(__name__)
11
12CLIENT_ID = os.environ['TWITCH_CLIENT_ID']
13CLIENT_SECRET = os.environ['TWITCH_CLIENT_SECRET']
14WEBHOOK_SECRET = os.environ['TWITCH_WEBHOOK_SECRET']
15CALLBACK_URL = os.environ['TWITCH_CALLBACK_URL']
16DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL']
17
18app = Flask(__name__)
19
20def get_access_token():
21    resp = requests.post('https://id.twitch.tv/oauth2/token', data={
22        'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET,
23        'grant_type': 'client_credentials'
24    })
25    resp.raise_for_status()
26    return resp.json()['access_token']
27
28def helix_headers(token):
29    return {'Authorization': f'Bearer {token}', 'Client-Id': CLIENT_ID}
30
31ACCESS_TOKEN = get_access_token()
32
33def get_broadcaster_id(username):
34    resp = requests.get('https://api.twitch.tv/helix/users',
35                        headers=helix_headers(ACCESS_TOKEN), params={'login': username})
36    resp.raise_for_status()
37    data = resp.json()['data']
38    if not data:
39        raise ValueError(f'User not found: {username}')
40    return data[0]['id']
41
42def subscribe(event_type, broadcaster_id):
43    resp = requests.post('https://api.twitch.tv/helix/eventsub/subscriptions',
44        headers={**helix_headers(ACCESS_TOKEN), 'Content-Type': 'application/json'},
45        json={
46            'type': event_type, 'version': '1',
47            'condition': {'broadcaster_user_id': broadcaster_id},
48            'transport': {'method': 'webhook', 'callback': CALLBACK_URL, 'secret': WEBHOOK_SECRET}
49        }
50    )
51    if resp.status_code == 409:
52        logger.info(f'{event_type} subscription already exists')
53        return
54    resp.raise_for_status()
55    logger.info(f'Subscribed to {event_type}')
56
57def verify_signature(raw_body, headers):
58    msg = (headers.get('Twitch-Eventsub-Message-Id', '').encode() +
59           headers.get('Twitch-Eventsub-Message-Timestamp', '').encode() + raw_body)
60    expected = 'sha256=' + hmac.new(WEBHOOK_SECRET.encode(), msg, hashlib.sha256).hexdigest()
61    return hmac.compare_digest(expected, headers.get('Twitch-Eventsub-Message-Signature', ''))
62
63@app.route('/twitch/eventsub', methods=['POST'])
64def eventsub():
65    raw = request.get_data()
66    if not verify_signature(raw, dict(request.headers)):
67        return Response('Forbidden', status=403)
68    data = request.get_json()
69    msg_type = request.headers.get('Twitch-Eventsub-Message-Type')
70    if msg_type == 'webhook_callback_verification':
71        return Response(data['challenge'], content_type='text/plain', status=200)
72    if msg_type == 'notification':
73        event = data['event']
74        sub_type = data['subscription']['type']
75        if sub_type == 'stream.online':
76            username = event['broadcaster_user_name']
77            bid = event['broadcaster_user_id']
78            stream = requests.get('https://api.twitch.tv/helix/streams',
79                                  headers=helix_headers(ACCESS_TOKEN),
80                                  params={'user_id': bid}).json().get('data', [])
81            if stream:
82                s = stream[0]
83                thumb = s['thumbnail_url'].replace('{width}','320').replace('{height}','180')
84                thumb += f'?t={int(time.time())}'
85                requests.post(DISCORD_WEBHOOK_URL, json={'embeds': [{
86                    'title': f'{username} is LIVE!',
87                    'description': f'Playing **{s["game_name"]}**\n{s["title"]}',
88                    'url': f'https://twitch.tv/{username}',
89                    'color': 9520895, 'image': {'url': thumb},
90                    'fields': [{'name': 'Viewers', 'value': str(s['viewer_count']), 'inline': True}]
91                }]})
92                logger.info(f'Alert posted for {username}')
93        elif sub_type == 'stream.offline':
94            requests.post(DISCORD_WEBHOOK_URL, json={
95                'content': f':red_circle: {event["broadcaster_user_name"]} has gone offline.'
96            })
97    return Response(status=204)
98
99if __name__ == '__main__':
100    broadcaster_id = get_broadcaster_id(os.environ.get('TWITCH_CHANNEL', 'ninja'))
101    subscribe('stream.online', broadcaster_id)
102    subscribe('stream.offline', broadcaster_id)
103    app.run(port=8080)