Awful Twitch v3 API CLI thingamagoop

twitch_api.py

# Create a localvar.py and define CLIENT_ID and TOKEN for API stuff
# You're on your own for getting those.

import requests
import ssl

from urllib.parse import quote_plus

from localvar import CLIENT_ID, TOKEN

# Replace with your own API Client ID here
TWITCH_URI_BASE = "https://api.twitch.tv/kraken/"
req_headers = {"Accept": "application/vnd.twitchtv.v3+json",
               "Client-ID": CLIENT_ID,
               "Authorization": TOKEN}

URI_BASE = "https://api.twitch.tv/kraken/"
USER_NAME = None

def raw_api_call(uri):
    # Passed URIs should be base-relative, but if we get a full one, use it
    if uri.startswith("http"):
        url = uri
    else:
        url = URI_BASE + uri

    # Prepare our request and shunt return data JSON->Dict
    req = urllib2.Request(url, headers=req_headers)
    gcontext = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
    data = urllib2.urlopen(req, context=gcontext).read()
    result = json.loads(data)

    return result


def paginating_api_call(uri, to_yield="streams"):
    """ Generic API call function that expects a "streams" object and channel
        paginate in response to over 100 results. Useful for searches, follower
        lists, and other multi-stream searches.

        Can also handle single-response searches that return one "stream"
        object.

        To return something besides "stream" and "streams", override to_yield
        """
    # Passed URIs should be base-relative, but if we get a full one, use it
    if uri.startswith("http"):
        url = uri
    else:
        url = TWITCH_URI_BASE + uri

    # Prepare our request and shunt return data JSON->Dict
    req = requests.get(url, headers=req_headers)
    if req.status_code != 200:
        print("Return code: " + str(req.status_code))
        return None
    result = req.json()

    # If we called a one-stream search, return just that stream and run away
    if to_yield.endswith('s'):
        to_yield_singular = to_yield[:-1]
        if to_yield_singular in result:
            yield result[to_yield_singular]
            return

    # Yield each stream individually
    # We yield because if we have to paginate, we don't want to block
    for s in result[to_yield]:
        yield s

    # Do pagination if response has more than max items
    if "_total" in result and result["_total"] > 100:
        for i in range(int(result["_total"] / 100)):
            if '?' in url:
                offset_url = "%s&offset=%d" % (url, (i + 1) * 100)
            else:
                offset_url = "%s?offset=%d" % (url, (i + 1) * 100)
            req = requests.get(offset_url, headers=req_headers)
            result = req.json()

            for s in result[to_yield]:
                yield s


def get_authenticated_user():
    for u in paginating_api_call("", to_yield="tokens"):
        return u["user_name"]


def get_user_favorites():
    url = "streams/followed?limit=100"
    for r in paginating_api_call(url):
        yield r


def get_user_all_favorites():
    """ Unfortunate name for backward compatability. get_user_favorites() only
        return active streams. This one gets ALL favorites. """
    global USER_NAME
    if USER_NAME is None:
        USER_NAME = get_authenticated_user()

    url = "/users/%s/follows/channels?limit=100" % (USER_NAME)
    for r in paginating_api_call(url, to_yield="follows"):
        yield r


def get_channel_status(channel, raw=False):
    url = "streams/%s" % channel
    for r in paginating_api_call(url):
        yield r


def get_game_streams(game):
    game = quote_plus(game)
    url = "streams?limit=100&game=%s" % game
    for r in paginating_api_call(url):
        yield r


def search_api(search):
    search = quote_plus(search)
    url = "search/streams?limit=100&q=%s" % search

    for r in paginating_api_call(url):
        yield r

twitch_cli.py

#!/usr/bin/env python3

import twitch_api
import shutil
import argparse
import time
from termcolor import cprint

highlights = [
    "brossentia",
    "macaw45",
    "nescardinality",
    "vandaeron"
]

parser = argparse.ArgumentParser()
parser.add_argument("-f",
                    "--follow",
                    help=("Follow along with the API and print "
                          "online/offline events"),
                    action="store_true")
args = parser.parse_args()

termwidth = shutil.get_terminal_size((80, 24))[0]

def print_table_border():
    """ Prints a pretty header for the formatted table. """
    headers = "{:5} {:25.25} {:42.42} {}".format("View", "Streamer", "Game", "Title")
    padding = " " * (termwidth - len(headers))
    cprint(headers + padding, attrs=['underline'])


def print_formatted_table(streams):
    """ Take a list of twitch account API data objects and prints a neat table
        containing their viewer count, account name, game, and title. """
    # Sort by popularity
    streams.sort(key=lambda stream: stream['viewers'], reverse=True)

    # Length of status cell, defined by termwidth - space for everything else
    celllength = termwidth - 75

    # Print the table, each row alternating between none and grey background
    # Also don't forget to highlight our defined highlights
    color = False
    for s in streams:
        name = s["channel"]["name"]
        game = s["channel"]["game"]
        status = s["channel"]["status"][0:celllength].replace("\n", " ")
        viewers = s["viewers"]

        if color:
            color = False
            bg = "on_grey"
        else:
            color = True
            bg = None

        if name.lower() in highlights:
            fg = "green"
            attrs = ['bold']
        else:
            fg = None
            attrs = []

        dataline = "{:5} {:25.25} {:42.42} {}".format(viewers, name, game, status)
        padding = " " * (termwidth - len(dataline))
        cprint(dataline + padding, fg, bg, attrs=attrs)

def get_live_follows():
    """ Returns a list of twitch API objects representing account objects
        of live streamers. Differs from follow_list_from_api in that this
        returns full data, not just account name. """
    streams = []

    # This is a generator so we need to unspool it into a list
    for i in twitch_api.get_user_favorites():
        streams.append(i)

    return streams

def follow_list_from_api(result):
    """ Returns a list of twitch account names that are live. """
    live = []
    for stream in result:
        live.append(stream["channel"]["name"])

    return live

def print_live_list(streams):
    print_table_border()
    print_formatted_table(streams)


if __name__ == "__main__":
    # Non -f invocation: print a nice table of live streams
    if not args.follow:
        result = get_live_follows()
        print_live_list(result)
    # -f invocation. Follow mode, like tail -f but colorful and dumb
    else:
        last_list = set()

        while True:
            result = get_live_follows()

            this_list = set(follow_list_from_api(result))

            new_streams = this_list.difference(last_list)
            old_streams = last_list.difference(this_list)

            now = time.strftime("%H:%M")
            if old_streams:
                print(now + "> \033[0;31m-" + ", ".join(sorted(old_streams)) + "\033[0;0m")
            if new_streams:
                print(now + "> \033[0;32m+" + ", ".join(sorted(new_streams)) + "\033[0;0m")

            last_list = this_list
            time.sleep(60)