Download Twitch channel emotes

download_emotes.py

#!/usr/bin/env python3.6
# -*- coding: utf-8 -*-

#
# put all channels you want to download the emotes of
# into a text file called "channel_list.txt", separated
# by line breaks
# also download subscriber.json from
# https://twitchemotes.com/apidocs
# and put it into a folder called "api" next to this script
# all emotes, sub/cheer badges, and cheermotes
# will be downloaded into Channels/<channel_name>/
# @SamusAranX 2017
#

import requests
import json
import sys
import os
from os.path import basename, dirname, join, abspath, exists
from twapiv5 import TwitchAPI

TWITCH_API_KEY = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

PATH_SUPFLDR = "Channels"
PATH_EMOTES = "Emotes"
PATH_CHEERMOTES = "Cheermotes"
PATH_BADGE_SUB = "SubBadges"
PATH_BADGE_CHEER = "CheerBadges"
INDENT_CHAR = "\t"

def download(url, path):
	response = requests.get(url)
	if response.status_code == 200:
		os.makedirs(dirname(path), exist_ok=True)

		with open(path, 'wb') as f:
			f.write(response.content)
			return True

	return False

def main():
	t = TwitchAPI(TWITCH_API_KEY)

	# In a low-amount-of-RAM env, this is the only JSON file we can load into memory
	# If we load anything else, Python runs out of memory
	with open("api/subscriber.json", "r", encoding="utf8") as f:
		tw_subscriber = json.load(f)

	try:
		with open("channel_list.txt", "r", encoding="utf8") as ch_list:
			channel_list = [ch.strip() for ch in ch_list.readlines()]
	except FileNotFoundError as e:
		raise e

	all_channels_to_crawl = [tw_subscriber[ch_id] for ch_id in tw_subscriber if tw_subscriber[ch_id]["channel_name"] in channel_list]
	for new_channel in all_channels_to_crawl:
		channel_name = new_channel["channel_name"]
		channel_id = new_channel["channel_id"]

		print(f"New channel: {channel_name}! ({channel_id})")

		def get_channel_path(c):
			return f"{PATH_SUPFLDR}/{channel_name}/{c}/"
		
		channel_emotes = new_channel["emotes"]
		if channel_emotes:
			fldr_emotes = get_channel_path("Emotes")
			os.makedirs(fldr_emotes, exist_ok=True)
			print("Downloading emotes…")
			for ce in channel_emotes:
				emote_id = ce["id"]
				emote_code = ce["code"]
				for scale in ["1.0", "2.0", "3.0"]:
					emote_url = f"https://static-cdn.jtvnw.net/emoticons/v1/{emote_id}/{scale}"
					emote_path = fldr_emotes + f"{emote_code}-{scale[:1]}.png"
					if download(emote_url, emote_path):
						print(f"{INDENT_CHAR}{emote_path} downloaded.")
					else:
						print(f"{INDENT_CHAR}{emote_path} failed.")
		else:
			print("No channel emotes.")

		subscriber_badges = new_channel["subscriber_badges"]
		if subscriber_badges:
			fldr_badges_sub = get_channel_path("Sub Badges")
			os.makedirs(fldr_badges_sub, exist_ok=True)
			print("Downloading sub badges…")
			for sb in subscriber_badges:
				sub_badge = subscriber_badges[sb]
				for scale in ["image_url_1x", "image_url_2x", "image_url_4x"]:
					sb_url = sub_badge[scale]
					sb_fname = fldr_badges_sub + f"{sb}_{scale[10:]}.png"
					if download(sb_url, sb_fname):
						print(f"{INDENT_CHAR}{sb_fname} downloaded.")
					else:
						print(f"{INDENT_CHAR}{sb_fname} failed.")
		else:
			print("No subscriber badges.")

		cheer_badges = new_channel["bits_badges"]
		if cheer_badges:
			fldr_badges_cheer = get_channel_path("Cheer Badges")
			os.makedirs(fldr_badges_cheer, exist_ok=True)
			print(f"Downloading cheer badges…")
			for bb in cheer_badges:
				bit_badge = cheer_badges[bb]
				for scale in ["image_url_1x", "image_url_2x", "image_url_4x"]:
					bb_url = bit_badge[scale]
					bb_fname = fldr_badges_cheer + f"{bb}_{scale[10:]}.png"
					if download(bb_url, bb_fname):
						print(f"{INDENT_CHAR}{bb_fname} downloaded.")
					else:
						print(f"{INDENT_CHAR}{bb_fname} failed.")
		else:
			print("No cheer badges.")

		print("Fetching cheermotes from API…")
		cheermotes = t.get_cheermotes(channel_id)
		cheermotes = cheermotes["actions"]
		cheermotes = [cm for cm in cheermotes if cm["type"] == "channel_custom"]

		if len(cheermotes) > 0:
			fldr_cheermotes = get_channel_path("Cheermotes")

			cheermotes = cheermotes[0]
			scales = cheermotes["scales"]
			bgs = cheermotes["backgrounds"]
			states = cheermotes["states"]
			print("Downloading cheermotes…")
			for tier in cheermotes["tiers"]:
				tier_id = tier["id"]
				# for bg in [bgs]: # ignore dark bg cheermotes
				for bg in ["light"]:
					# for state in states: # ignore static cheermotes
					for state in ["animated"]:
						for scale in scales:
							os.makedirs(fldr_cheermotes, exist_ok=True)
							cm_url = tier["images"][bg][state][scale]
							cm_fname = join(fldr_cheermotes, f"{tier_id}-{scale}.gif")
							if download(cm_url, cm_fname):
								print(f"{INDENT_CHAR}{cm_fname} downloaded.")
							else:
								print(f"{INDENT_CHAR}{cm_fname} failed.")
		else:
			print("No cheermotes.")

if __name__ == '__main__':
	main()

twapiv5.py

#!/usr/bin/env python3.6
# -*- coding: utf-8 -*-

#
# smol api wrapper I wrote to download channels' cheermotes
# @SamusAranX 2017
#

import os
import sys
import json
import requests
from urllib.parse import urljoin, parse_qs, urlsplit, urlunsplit, urlencode

class TwitchAPIError(Exception):
	def __init__(self, error_json):
		self.message = error_json["message"]
		self.status = error_json["status"]
		self.error = error_json["error"]

class TwitchAPI:
	_ROOT_URL = "https://api.twitch.tv/kraken/"
	_ACCEPT = "application/vnd.twitchtv.v5+json"

	def __init__(self, client_id):
		self.session = requests.Session()
		self.session.headers.update({"Accept": self._ACCEPT})
		self.session.headers.update({"Client-ID": client_id})

	def _get(self, endpoint, **params):
		scheme, netloc, path, query_string, fragment = urlsplit(endpoint)
		query_params = parse_qs(query_string)

		for p in { k:v for k,v in params.items() if v is not None }:
			# print(p, params[p])
			query_params[p] = params[p]

		new_query_string = urlencode(query_params, doseq=True)
		endpoint = urlunsplit((scheme, netloc, path, new_query_string, fragment))

		# print(endpoint)
		
		r = self.session.get(endpoint)

		try:
			response = r.json()
		except Exception as e:
			print(r.text)
			raise e
		
		if "error" in response:
			raise TwitchAPIError(response)

		return response

	#
	# BITS
	#
	def get_cheermotes(self, channel_id = None):
		endpoint = urljoin(self._ROOT_URL, "bits/actions", allow_fragments=False)
		return self._get(endpoint, channel_id = channel_id)

	#
	# CHAT
	#
	def get_chat_badges_by_channel(self, channel_id):
		endpoint = urljoin(self._ROOT_URL, f"chat/{channel_id}/badges", allow_fragments=False)
		return self._get(endpoint)

	def get_chat_emoticons_by_set(self, emotesets = None):
		endpoint = urljoin(self._ROOT_URL, "chat/emoticon_images", allow_fragments=False)
		return self._get(endpoint, emotesets = emotesets)

	def get_all_chat_emoticons(self):
		endpoint = urljoin(self._ROOT_URL, "chat/emoticons", allow_fragments=False)
		return self._get(endpoint)

	#
	# USERS
	#
	def get_user_by_id(self, user_id):
		endpoint = urljoin(self._ROOT_URL, f"users/{user_id}", allow_fragments=False)
		return self._get(endpoint)

	def get_users(self, user_ids):
		endpoint = urljoin(self._ROOT_URL, "users", allow_fragments=False)
		return self._get(endpoint, login = ",".join(user_ids))

	def get_user_emotes(self, user_id):
		endpoint = urljoin(self._ROOT_URL, f"users/{user_id}/emotes", allow_fragments=False)
		return self._get(endpoint)