fleast/twitch.py

import cherrypy
import requests
import threading
import time


class TwitchClient:
    def __init__(self, token, freq=2):
        self.token = token
        self.lock = threading.Lock()

        self.header_v5 = {'Client-ID': self.token,
                          'Accept': 'application/vnd.twitchtv.v5+json'}
        self.header_v6 = {'Client-ID': self.token}
        self.urlbase_v5 = 'https://api.twitch.tv/kraken'
        self.urlbase_v6 = 'https://api.twitch.tv/helix'

        self.last_q = time.time()
        self.delay = 1/freq

    def do_q(self, base, header):
        """
            Do query for twitch server
            :param base: string with requesting URL
            :param header: dictionary of http headers
            :return: string with response or None
        """
        self.lock.acquire()                                     # Lock for 1 at time query
        try:
            cherrypy.log('Request: %s' % base)
            delta = time.time() - self.last_q                   # Delta for correct query freq
            if delta < self.delay:
                time.sleep(delta)                               # Sleep remaining time
            r = requests.get(base, headers=header).json()
            self.last_q = time.time()
            cherrypy.log('Request: OK')
        except requests.exceptions.RequestException as e:
            cherrypy.log('Request: FAIL')
            cherrypy.log('Error: {}'.format(e))
            r = None
        finally:
            self.lock.release()                                 # Do not forget to release lock
            return r
        
    def get_base(self, ver):
        """
            Get base which is depended on API version
            :param ver: string with API version ('v5' or 'v6')
            :return: tuple with list of headers and URL string
            :raises: value error on incorrect API version
        """
        if ver == 'v5':
            return self.header_v5, self.urlbase_v5
        elif ver == 'v6':
            return self.header_v6, self.urlbase_v6
        else:
            raise ValueError('Not supported API version')

    # - # - #
    
    def raw_query_v6(self, q):
        """
            Do a query with API v6
            :param q: query string
            :return: string with get query result or None
        """
        header, base = self.get_base('v6')
        return self.do_q(base+q, header)

    def raw_query_v5(self, q):
        """
            Do a query with API v5
            :param q: query string
            :return: string with get query result or None
        """
        header, base = self.get_base('v5')
        return self.do_q(base+q, header)

    def get_game_id(self, name):
        """
            Getting game id
            :param name: string with the name of game
            :return: tuple of integer with game id and string with game name or None, None
        """
        return self.get_game_id_v5(name)

    def get_live_streams(self, name, lang):
        """
            Getting list of livestreams
            :param name: string with the name of game
            :param lang: string with the shortcut of language
            :return: list of all streams which are live
        """
        return self.get_live_streams_v5(name, lang)

    def get_game_id_v5(self, name):
        """
            Getting game id with API v5
            :param name: string with the name of game
            :return: tuple of integer with game id and string with game name or None, None
        """
        header, base = self.get_base('v5')
        r = self.do_q('{}/search/games?query={}'.format(base, name), header)
        if r and r.get('games'):
            return r['games'][0]['_id'], r['games'][0]['name']
        return None, None

    def get_live_streams_v5(self, name, lang):
        """
            Getting list of livestreams with API v5
            :param name: string with the name of game
            :param lang: string with the shortcut of language
            :return: list of all streams which are live with this format -
                https://dev.twitch.tv/docs/v5/reference/search/#search-streams
        """
        header, base = self.get_base('v5')
        init_q_template = '{}/streams/?game={}&language={}&limit={}&stream_type=live'
        q_template = '{}/streams/?game={}&language={}&limit={}&stream_type=live&offset={}'
        data = self.do_q(init_q_template.format(base, name, lang, 100), header)
        if data is None:
            return []
        total = data['_total']
        streams = len(data['streams'])

        while total > streams:
            r = self.do_q(q_template.format(base, name, lang, 100, streams), header)
            if r is None:
                return []
            data['streams'].extend(r['streams'])
            total = r['_total']
            streams = len(data['streams'])
        # Tweak for getting only live sterams
        data['streams'] = [x for x in data['streams'] if x['stream_type'] == 'live']
        data['_total'] = len(data['streams'])
        return data