ZatsuDachi Twitch Plugin (readonly)

twitch.py

# NOTE: You must update your twitch URL by editing this file!

# Import required packages
from api import PluginBase, Message, Color
from flask import escape
import threading
import itertools
import asyncio
import irc.client_aio
import irc.client
import sys

# Define the plugin class
class Plugin(PluginBase):
    # Set plugin name and target stream
    name = "Twitch Chat Plugin (readonly)"
    # Set image URL for the plugin
    pluginImageURL = "https://brand.twitch.tv/assets/logos/svg/glitch/purple.svg"
    # Lock
    _lock = threading.Lock()
    # Plugin loop
    _connection = None

    # CONNECTION NOTE: Paste the id of your livestream here!
    targetStream = "doxy_ai"

    # Asynchronous method for the plugin's main functionality
    async def go(self):
        # If the plugin is already running... don't start it again!
        Plugin._lock.acquire()
        try:
            if self._connection is not None:
                return
            # Initialize IRC reactor and connect to Twitch IRC server
            reactor = irc.client.Reactor()
            try:
                self._connection = reactor.server().connect("irc.chat.twitch.tv", 6667, "justinfan1234")
            except irc.client.ServerConnectionError:
                print(sys.exc_info()[1])
                raise SystemExit(1)
        finally:
            Plugin._lock.release()

        # Add event handlers
        self._connection.add_global_handler("welcome", self.on_connect)
        self._connection.add_global_handler("join", self.on_join)
        self._connection.add_global_handler("disconnect", self.on_disconnect)
        self._connection.add_global_handler("pubmsg", self.on_message)

        # Register Twitch badges if not already registered
        self.register_badge_if_not_registered("broadcaster", "https://static-cdn.jtvnw.net/badges/v1/5527c58c-fb7d-422d-b71b-f309dcb85cc1/3")
        self.register_badge_if_not_registered("moderator", "https://static-cdn.jtvnw.net/badges/v1/3267646d-33f0-4b17-b3df-f923a41db1d0/3")
        self.register_badge_if_not_registered("vip", "https://static-cdn.jtvnw.net/badges/v1/b817aba4-fad8-49e2-b88a-7cc744dfa6ec/3")
        self.register_badge_if_not_registered("founder", "https://static-cdn.jtvnw.net/badges/v1/511b78a9-ab37-472f-9569-457753bbe7d3/3")

        # Main loop to process events
        while self._keepAlive:
            reactor.process_once(.1)

    # Method to stop the plugin
    def stop(self):
        PluginBase.stop(self)
        Plugin._lock.acquire()
        try:
            if self._connection is not None:
                self._connection.quit()
            self._connection = None
        finally:
            Plugin._lock.release()
            print("Twitch stopped!")

    # Disable GUI updating 
    def setupGUI(self, content, applyButton):
        return None

    # Event handler for when the plugin connects to Twitch channel
    def on_connect(self, connection, event):
        # Request to see commands
        connection.cap("REQ", "twitch.tv/commands")
        # Request to see more information about messages!
        connection.cap("REQ", "twitch.tv/tags")
        # Attempt to join the target channel!
        target = "#" + self.targetStream
        if irc.client.is_channel(target):
            connection.join(target)
        else:
            raise RuntimeError("Failed to connect to the twitch channel `" + target + "` it doesn't seam to exist!")

    # Event handler for when the plugin successfully joins the channel
    def on_join(self, connection, event):
        print("Connected to Twitch Channel " + self.targetStream)

    # Event handler for incoming messages
    def on_message(self, connection, event):
        # Extract message information
        print(event.arguments)
        tags = {}
        for pair in event.tags:
            tags[pair['key']] = pair['value']
        print(tags)
        
        messageContent: str = event.arguments[0]

        # Extract badge information
        badges = []
        if tags["badges"] is not None:
            badges = [badge.split("/")[0] for badge in tags["badges"].split(",")]
        
        # Handle emojis in the message!
        if tags["emotes"] is not None:
            for emote in tags["emotes"].split('/'):
                emoteId, positions_str = emote.split(':')

                textPositions = [
                    {
                        'startPosition': int(start),
                        'endPosition': int(end)
                    }
                    for start, end in (position.split('-') for position in positions_str.split(','))
                ]

                start_pos = textPositions[0]['startPosition']
                end_pos = textPositions[0]['endPosition'] + 1

                emote_url = f'https://static-cdn.jtvnw.net/emoticons/v2/{emoteId}/default/dark/3.0'
                emote_text = messageContent[start_pos:end_pos]
                self.register_emote(emote_text, emote_url)
        
        # Pass message information to receive_message method
        self.recieve_message(Message(content=messageContent, sender=tags["display-name"], senderColor=Color(tags["color"]), senderBadges=badges, pluginImageUrl=self.pluginImageURL))

    # Event handler for disconnection
    def on_disconnect(self, connection, event):
        raise SystemExit()