ZatsuDachi Kick Plugin (readonly)

kick.py

# Requirements: pip install tls_client

# Import required packages
from api import PluginBase, Message, Color
from flask import escape
import asyncio
import websockets
import json
import sys
import tls_client
import re


# Define the plugin class
# Based on: https://github.com/mumixam/kick.com-chatlogger-python
# Updated with code from: https://github.com/lukemvc/kickbot
class Plugin(PluginBase):
	# Set plugin name
	name = "Kick Chat Plugin"
	# Set image URL for the plugin
	pluginImageURL = "data:image/webp;base64,UklGRgAEAABXRUJQVlA4WAoAAAAIAAAA5QAA5QAAVlA4IHwDAADQIgCdASrmAOYAPpVAkkkkIgmPnCIEonpbuF1UQgZH8i1oLl/5W8PTI6dj8If5n7ufeT/d/ZT5gH6K/qB1s/MB/H/8V+2nvE/439Vfc36AH9m8532LvQP8uD2a/3E/Z32gNVs8s/yIwTtVRzKW66Z/77LxrpS3Om503Om503Om503OmgsYKQVlMeljt5B44K8siNs5Ydl4XivrTVkW/qdGcf4DllU3NnzZfYSYoAAT0uvRSq3NnzbOAoWB4vdmndKSLEjSr/DQe0hcRznTPcqOLrQ9gmo8er1Yy5SlF/g1P/Ko9c+Sa8a02ZrTpRGXrpKDUQIluaxJj53wHD1VugVuL/Kau/skQsvGulPLf5ZzpudNzpudNzpudNzpudM8AAD+2MAAABQ1dwhqklBFtJ1hSH2wyw+LvQtahBtg2bySrhEDmMU2cCYXNPGHy3IPK4erX7pX2cO12XjzDgmBILmaSm3kHUZVNTmNRZQsRTn+FRI3Ff+5n5kKWDrpg/KYKLWHrK+mHUVf8Orfe25U6zp/XPdh52c+EKw43YhrmssBdO3N713I01U5PSOXHv1QB6BED92QVCfCPYFlCTrjFjQ+dFmmz46X6OY6O68afGpzlozfxCPXHBT5SHR2xElpnzOC0L9jiU81O7zQkFrBB0JUGAMVJ9+NZE0+0DTR0vkTTxuUuNoU4sgDt2RGdpPPPFZnPkoSzv350PoOYQRkmsd2BGjTe+bX3+pEpiGXXk14LvH+enrrt5XVqZi7CdVGvueqWD+fPGFteWKmITTyu/TQOo+15pAdDbdpzJ4C0+ADfjR2Z3mUN+eos6exNbud+bEfpPTlEI9IbhTNWLCpqcGVxM2yK33KxYUB5Kzsu37AXbJKc3+azjS4e7RwKhL/+L3VzsOTqo4W/u7obeM04i22fHLdJnBv4ESUjXIB/HEbCYYtezSz2NHCNYM1ZVQN8QnfGZDVTVa16oLuybtiOf+A9rx0KpZ/uO1kTgH9dPiiOYFrmmg/Mg3OoLjJveTBiKkyFKcxq4xl9HdE3hVyu3LoJl5YHWqbZqXJdmMghotRf9T8SID4oLdFw0AxCAfe+7jyaMBn7tNJSjnVUz/yTrZ5Y/rSmYMwYYcAvls7PhsBoXALl/x3giBNW8B70hB+89sBLEz7PkSuuOZem3yMeLLKaO+n4XQAAAAAAAAARVhJRl4AAABJSSoACAAAAAEAaYcEAAEAAAAaAAAAAAAAAAEAhpIHADIAAAAsAAAAAAAAAEFTQ0lJAAAAMS41LjEtMjNELUNTN1RIUFdJSEFUVFJHQVNVRVY0Q1JXUUNJLjAuMi0w"
	# Reference to the current websocket
	_ws = None


	# CONNECTION NOTE: Paste the name of your livestream channel here!
	targetChannel = "pandaskills"


	async def go(self):
		if self._ws is not None: return

		slug = self.targetChannel.replace('_', '-')
		url = f"https://kick.com/api/v2/channels/{slug}"

		api = tls_client.Session(
			client_identifier="chrome_116",
			random_tls_extension_order=True
		)

		response = api.get(url, headers=self._BASE_HEADERS)
		status = response.status_code
		match status:
			case 403 | 429:
				raise RuntimeError(f"Error retrieving streamer info. Blocked By cloudflare. ({status})")
			case 404:
				raise RuntimeError(f"Streamer info for '{self.targetChannel}' not found. (404 error) ")
		data = response.json()
		channelID = data["chatroom"]["id"]

		# TODO: Need to register badges

		self._ws = await websockets.connect("wss://ws-us2.pusher.com/app/eb1d5f283081a78b932c?protocol=7&client=js&version=7.6.0&flash=false")
		connection_response = json.loads(await self._ws.recv())
		socketId = json.loads(connection_response.get("data"))["socket_id"]
		await self._ws.send('{"event":"pusher:subscribe","data":{"auth":"","channel":"chatrooms.' + str(channelID) + '.v2"}}')
		join_response = json.loads(await self._ws.recv())
		print(f"Connected to Kick channel '{self.targetChannel}'")
		while self._keepAlive:
			self.on_message(await self._ws.recv())

	def stop(self):
		"""
		Stop the plugin and halt the websocket listener
		"""
		PluginBase.stop(self)
		# Plugin._lock.acquire()
		try:
			if Plugin._ws is not None: Plugin._ws.close()
			Plugin._ws = None
		finally: 
			# Plugin._lock.release()
			print("Kick stopped! Waiting for connection to timeout...")


	def on_message(self, message):
		msg = json.loads(message)
		if 'ChatMessageEvent' in msg['event']:
			print(msg)
			data = json.loads(msg['data'])
			content = data["content"]
			if not content:
				pass
			user = data['sender']['username']
			color = Color(data['sender']['identity']['color'])
			badges = data['sender']['identity']['badges']
			badges = [badge["type"] for badge in badges]

			emotes = re.findall(self._pattern, content)
			for (sub, id, text) in emotes:
				self.register_emote(text, f"https://files.kick.com/emotes/{id}/fullsize")
				content = content.replace(sub, text)

			# Create a Message object with the constructed content, sender information, color, and plugin image URL
			self.recieve_message(Message(content=content, sender=user, senderColor=color, senderBadges=badges, pluginImageUrl=self.pluginImageURL))


	_BASE_HEADERS = {
		"Accept": "application/json, text/plain, */*",
		"Accept-Encoding": "gzip, deflate, br",
		"Accept-Language": "en-US,en;q=0.9",
		"Sec-Fetch-Dest": "empty",
		"Sec-Fetch-Mode": "cors",
		"Sec-Fetch-Site": "same-origin",
		"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
					"(KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36",
	}
	_pattern = r'(\[emote:(\d*?):(\S*?)\])'