.gitignore

config.py

README.md

Twitch.tv subscription hooks with python.

This is only a proof of concept.

1. Start ngrok server and note URL (preferably the HTTPS one)

2. Rename config.py.example to config.py and fill out the relevent keys / URL. UIDS should be a list of user IDs as strings, or None to automatically get some User IDs from the 'Get Stream' endpoint.

3. Run python server.py leave this running to monitor requests coming back from twitch.

4. Run python notify.py in another terminal, and output should look like:

{'access_token': '_REDACTED_', 'expires_in': 4852660, 'token_type': 'bearer'}
---------
<Response [202]>
---------
b'{"total":1,"data":[{"topic":"https://api.twitch.tv/helix/streams?user_id=5678","callback":"https://_REDACTED_.ngrok.io/","expires_at":"2019-10-02T11:26:33Z"}],"pagination":{}}'
---------

5. Subsequent runs of notify.py should add further items to the response data list of the third request.

## Notes to self

A stream offline event looks like this in the server console:

Host: __REDACTED__.ngrok.io
User-Agent: Go-http-client/1.1
Connection: close
Content-Length: 11
Content-Type: application/json; charset=utf-8
Link: <https://api.twitch.tv/helix/webhooks/hub>; rel="hub", <https://api.twitch.tv/helix/streams?user_id=40017619>; rel="self"
Twitch-Notification-Id: __REDACTED__
Twitch-Notification-Timestamp: 2019-10-03T00:09:45Z
Accept-Encoding: gzip
X-Forwarded-Proto: https
X-Forwarded-For: __REDACTED__


b'{"data":[]}'

A stream change event looks like this in the server console:

Host: __REDACTED__.ngrok.io
User-Agent: Go-http-client/1.1
Connection: close
Content-Length: 353
Content-Type: application/json; charset=utf-8
Link: <https://api.twitch.tv/helix/webhooks/hub>; rel="hub", <https://api.twitch.tv/helix/streams?user_id=26301881>; rel="self"
Twitch-Notification-Id: __REDACTED__
Twitch-Notification-Timestamp: 2019-10-03T01:07:55Z
Accept-Encoding: gzip
X-Forwarded-Proto: https
X-Forwarded-For: __REDACTED__


b'{"data":[{"game_id":"505705","id":"35849957184","language":"en","started_at":"2019-10-02T19:08:56Z","tag_ids":["6ea6bca4-4712-4ab9-a906-e3336a9d8039"],"thumbnail_url":"https://static-cdn.jtvnw.net/previews-ttv/live_user_sodapoppin-{width}x{height}.jpg","title":"Stream","type":"live","user_id":"26301881","user_name":"sodapoppin","viewer_count":18798}]}'

Some info on hub.secret which is so far misterious:

https://discuss.dev.twitch.tv/t/how-to-use-the-hub-secret-with-web-hooks/13356

config.py.example

CLIENT_ID='your_key'
CLIENT_SECRET='your_secret'

# Get this when you run the ngrok server:
NGROK_URL='https://your-ngrok-url.ngrok.io/'

# Set `UIDS` to a list of user IDs as strings.
#  if set to `None` the notify.py script automatically
#  grabs some UIDs from the 'Get Streams' endpoint
UIDS=None
#UIDS=['5678']

notify.py

from config import CLIENT_ID, CLIENT_SECRET, NGROK_URL, UIDS
import requests

auth = requests.post("https://id.twitch.tv/oauth2/token", headers={'Client-ID':CLIENT_ID},data={"client_secret":CLIENT_SECRET,"grant_type":"client_credentials"})

ACCESS_TOKEN = "Bearer %s" % (auth.json()['access_token'],)


if not UIDS:
	# Get some user IDs for the 'Get Streams' endpoint
	# This is a convenient method to get some user IDs incase we don't know any.

	streams = requests.get("https://api.twitch.tv/helix/streams",  headers={'Client-ID':CLIENT_ID},data={"first":"10"})
	uids = [item['user_id'] for item in streams.json()['data']]
else:
	# Expect a list of user ID's from the config.
	uids = UIDS

# Subscribe to those uids

for uid in uids:
	hub_topic = "https://api.twitch.tv/helix/streams?user_id=%s" % (uid,)
	r = requests.post("https://api.twitch.tv/helix/webhooks/hub", headers={'Client-ID': CLIENT_ID}, data={"hub.callback": NGROK_URL, "hub.mode":"subscribe",'hub.topic':hub_topic,"hub.lease_seconds":600})

# Sleep then check the subscriptions.

import time
time.sleep(2)

webhook = requests.get("https://api.twitch.tv/helix/webhooks/subscriptions?", headers={"Authorization": ACCESS_TOKEN}, data={"first":"10"})

# Print diag

print(auth.json())
print("---------")

print(r)
print("---------")

print(webhook.content)
print("---------")

server.py

from flask import Flask, request

app = Flask(__name__)

@app.route('/', methods=['GET','POST'])
def result():

    # Check if `hub.challenge` is a GET arg in the request.
    # If it is, assume  this is an initial response to our
    # subscription.  Thus return it as per docs.
    
    if request.args.get('hub.challenge'):
        print ('Returning hub challenge.')
        return request.args.get('hub.challenge')
    
    # Next make sure there is a json body to the request

    try:
        resp = request.get_json()
    except Exception as e:
        print (e)
        return ''

    # print some diag

    print('==============')
    print(request.headers)
    print(request.data) 
    print('==============')


    # If the request payload is `{'data':[]}` the stream has gone offline 
    if 'data' in resp and len(resp['data']) == 0:
        print ('Stream went offline.')
        # TODO: Identify which stream here, possible from this header (see readme for full sample):
        # Link: <https://api.twitch.tv/helix/webhooks/hub>; rel="hub", <https://api.twitch.tv/helix/streams?user_id=40017619>; rel="self"
        return ''
    else:
        # At this stage we have 'Other Stream Change Event'
        print ('Stream changed.')
        # Do something with `resp` dictionary here.
        return ''   

if __name__=='__main__':
    app.run(host='0.0.0.0')