How to Automate Twitch Chat Moderation using the API

1import asyncio
2import json
3import os
4import re
5import time
6import requests
7import websockets
8import logging
9from collections import defaultdict, deque
10
11logging.basicConfig(level=logging.INFO)
12logger = logging.getLogger(__name__)
13
14CLIENT_ID = os.environ['TWITCH_CLIENT_ID']
15ACCESS_TOKEN = os.environ['MOD_ACCESS_TOKEN']
16BROADCASTER_ID = os.environ['BROADCASTER_ID']
17MOD_USER_ID = os.environ['MOD_USER_ID']
18
19# Moderation rules
20BANNED_PHRASES = ['buy followers', 'free nitro']
21SPAM_RE = re.compile(r'(.)\1{4,}')
22user_msg_times = defaultdict(deque)
23mod_actions = deque()
24
25def hdrs():
26    return {'Authorization': f'Bearer {ACCESS_TOKEN}', 'Client-Id': CLIENT_ID,
27            'Content-Type': 'application/json'}
28
29def circuit_ok():
30    now = time.time()
31    while mod_actions and mod_actions[0] < now - 60:
32        mod_actions.popleft()
33    return len(mod_actions) < 400
34
35def get_violation(text, user_id):
36    t = text.lower()
37    for p in BANNED_PHRASES:
38        if p in t:
39            return ('ban', f'Banned phrase: {p}', None)
40    if SPAM_RE.search(text):
41        return ('timeout', 'Character spam', 60)
42    now = time.time()
43    times = user_msg_times[user_id]
44    while times and times[0] < now - 5:
45        times.popleft()
46    times.append(now)
47    if len(times) > 3:
48        return ('timeout', 'Message flood', 30)
49    return None
50
51def take_action(action, user_id, msg_id, reason, duration=None):
52    if not circuit_ok():
53        logger.warning('Circuit breaker: pausing moderation')
54        time.sleep(5)
55    mod_actions.append(time.time())
56    try:
57        if action in ('ban', 'timeout'):
58            body = {'data': {'user_id': user_id, 'reason': reason}}
59            if action == 'timeout':
60                body['data']['duration'] = duration
61            resp = requests.post(
62                f'https://api.twitch.tv/helix/moderation/bans'
63                f'?broadcaster_id={BROADCASTER_ID}&moderator_id={MOD_USER_ID}',
64                headers=hdrs(), json=body
65            )
66            resp.raise_for_status()
67        if msg_id:
68            requests.delete(
69                f'https://api.twitch.tv/helix/moderation/chat'
70                f'?broadcaster_id={BROADCASTER_ID}&moderator_id={MOD_USER_ID}&message_id={msg_id}',
71                headers={'Authorization': f'Bearer {ACCESS_TOKEN}', 'Client-Id': CLIENT_ID}
72            )
73        logger.info(f'{action} {user_id}: {reason}')
74    except Exception as e:
75        logger.error(f'Moderation action failed: {e}')
76
77async def run_bot():
78    while True:
79        try:
80            async with websockets.connect('wss://eventsub.wss.twitch.tv/ws') as ws:
81                welcome = json.loads(await ws.recv())
82                session_id = welcome['payload']['session']['id']
83                requests.post(
84                    'https://api.twitch.tv/helix/eventsub/subscriptions',
85                    headers=hdrs(),
86                    json={'type': 'channel.chat.message', 'version': '1',
87                          'condition': {'broadcaster_user_id': BROADCASTER_ID, 'user_id': MOD_USER_ID},
88                          'transport': {'method': 'websocket', 'session_id': session_id}}
89                ).raise_for_status()
90                logger.info('Moderation bot connected')
91                async for raw in ws:
92                    ev = json.loads(raw)
93                    if ev.get('metadata', {}).get('message_type') != 'notification':
94                        continue
95                    e = ev['payload']['event']
96                    violation = get_violation(e['message']['text'], e['chatter_user_id'])
97                    if violation:
98                        action, reason, duration = violation
99                        take_action(action, e['chatter_user_id'], e['message_id'], reason, duration)
100        except Exception as e:
101            logger.error(f'Connection error: {e}. Reconnecting in 5s...')
102            await asyncio.sleep(5)
103
104asyncio.run(run_bot())