How to Automate Twitch Follower Reports using the API

1import requests, json, os, time, logging
2from datetime import datetime
3
4logging.basicConfig(level=logging.INFO)
5
6CLIENT_ID = os.environ['TWITCH_CLIENT_ID']
7ACCESS_TOKEN = os.environ['TWITCH_ACCESS_TOKEN']
8DISCORD_WEBHOOK = os.environ.get('DISCORD_WEBHOOK_URL')
9CHANNEL_NAME = os.environ['TWITCH_CHANNEL_NAME']
10SNAPSHOT_FILE = f'snapshot_{CHANNEL_NAME}.json'
11
12HEADERS = {'Authorization': f'Bearer {ACCESS_TOKEN}', 'Client-Id': CLIENT_ID}
13
14def validate_token():
15    r = requests.get('https://id.twitch.tv/oauth2/validate', headers={'Authorization': f'OAuth {ACCESS_TOKEN}'})
16    if r.status_code != 200:
17        raise Exception('Token invalid or expired — re-authorize')
18    logging.info('Token valid')
19
20def get_broadcaster_id(channel_name):
21    r = requests.get('https://api.twitch.tv/helix/users', headers=HEADERS, params={'login': channel_name})
22    return r.json()['data'][0]['id']
23
24def get_all_followers(broadcaster_id):
25    followers, cursor = [], None
26    while True:
27        params = {'broadcaster_id': broadcaster_id, 'first': 100}
28        if cursor: params['after'] = cursor
29        r = requests.get('https://api.twitch.tv/helix/channels/followers', headers=HEADERS, params=params)
30        if r.status_code == 429:
31            reset = int(r.headers.get('Ratelimit-Reset', time.time() + 60)) - int(time.time())
32            time.sleep(max(reset + 1, 1))
33            continue
34        data = r.json()
35        followers.extend(data.get('data', []))
36        cursor = data.get('pagination', {}).get('cursor')
37        logging.info(f'Fetched {len(followers)}/{data.get("total", 0)}')
38        if not cursor: break
39        time.sleep(0.1)
40    return followers
41
42def generate_and_send_report(channel_name):
43    validate_token()
44    broadcaster_id = get_broadcaster_id(channel_name)
45    followers = get_all_followers(broadcaster_id)
46    current_ids = {f['user_id']: f for f in followers}
47    
48    previous = {}
49    prev_time = datetime.utcnow()
50    if os.path.exists(SNAPSHOT_FILE):
51        with open(SNAPSHOT_FILE) as f:
52            snap = json.load(f)
53            previous = snap.get('followers', {})
54            prev_time = datetime.fromisoformat(snap.get('timestamp', datetime.utcnow().isoformat()))
55    
56    new_ids = set(current_ids) - set(previous)
57    lost_ids = set(previous) - set(current_ids)
58    days = max((datetime.utcnow() - prev_time).total_seconds() / 86400, 1)
59    
60    with open(SNAPSHOT_FILE, 'w') as f:
61        json.dump({'timestamp': datetime.utcnow().isoformat(), 'followers': current_ids}, f)
62    
63    if DISCORD_WEBHOOK:
64        requests.post(DISCORD_WEBHOOK, json={'embeds': [{
65            'title': f'Twitch Follower Report — {channel_name}',
66            'color': 9520895,
67            'fields': [
68                {'name': 'Total', 'value': f'{len(current_ids):,}', 'inline': True},
69                {'name': 'New', 'value': f'+{len(new_ids)}', 'inline': True},
70                {'name': 'Lost', 'value': f'-{len(lost_ids)}', 'inline': True},
71                {'name': 'Net', 'value': str(len(new_ids) - len(lost_ids)), 'inline': True},
72                {'name': 'Daily Rate', 'value': f'{len(new_ids)/days:.1f}/day', 'inline': True},
73            ]
74        }]})
75        logging.info('Report delivered to Discord')
76
77generate_and_send_report(CHANNEL_NAME)