add realtime yt chat support via pytchat

yt_chat.py

@@ -1,11 +1,11 @@
 import os
+import pytchat
 
 # from dotenv import load_dotenv
 from guard import Guardian
 from tv_crew import TVCrew
 import requests
 import time
-import googleapiclient.discovery
 
 # load_dotenv()
 unreal_orchestrator_url = os.getenv("ORCHESTRATOR_URL")
@@ -13,7 +13,8 @@ api_key_unreal = os.getenv("API_KEY_UNREAL")
 
 API_KEY = os.getenv("YOUTUBE_API_KEY")
 VIDEO_ID = "xBVJIJQg1FY"
-POLL_INTERVAL = 15  # seconds
+POLL_INTERVAL = 3  # seconds
+chat = pytchat.create(video_id=VIDEO_ID)
 
 
 class StreamChatHost:
@@ -31,116 +32,79 @@ Keep your responses short, like two or three sentences, and be witty."""
 
         self.messages = [self.system_message]
         self.headers = {"Content-Type": "application/json", "x-api-key": api_key_unreal}
-        self.youtube = googleapiclient.discovery.build(
-            "youtube", "v3", developerKey=API_KEY
-        )
-        self.chat_id = self.get_live_chat_id(self.youtube, VIDEO_ID)
-        self.next_page_token = None
 
     def chat_interaction_loop(self):
         """Checks if the Host is in lobby, then pulls chat messages from the Live Stream. The Host then replies to the message."""
-        while True:
-            time.sleep(POLL_INTERVAL)
-            request = requests.get(
-                f"{unreal_orchestrator_url}/status", headers=self.headers
-            )
-            in_lobby = request.json().get("in_lobby", False)
-            if not in_lobby:
-                continue
-            if len(self.messages) > 20:
-                self.messages = [self.system_message]
-                self.tv_crew.clear_context()
-            try:
-                request = self.youtube.liveChatMessages().list(
-                    liveChatId=self.chat_id,
-                    part="snippet,authorDetails",
-                    maxResults=20,
-                    pageToken=self.next_page_token,
+        while chat.is_alive():
+            for c in chat.get().sync_items():
+                request = requests.get(
+                    f"{unreal_orchestrator_url}/status", headers=self.headers
                 )
-                response = request.execute()
-                user_messages = []
-                for item in response.get("items", []):
-                    author_name = item["authorDetails"]["displayName"]
-                    message_text = item["snippet"]["displayMessage"]
-                    user_messages.append(f"{author_name}: {message_text}")
-
-                first_fetch = True if self.next_page_token is None else False
-                self.next_page_token = response.get("nextPageToken")
-                if first_fetch:
-                    continue
-                if len(user_messages) == 0:
+                in_lobby = request.json().get("in_lobby", False)
+                if not in_lobby:
                     continue
-            except Exception as e:
-                continue
-            # **************************************************************
-
-            user_message = "[YouTube Chat]\n"
-            for msg in user_messages:
-                if len(msg) <= 500:
-                    user_message += f"{msg}\n"
-            is_safe, _ = self.guardian.moderate_message(user_message)
+                if len(self.messages) > 20:
+                    self.messages = [self.system_message]
+                    self.tv_crew.clear_context()
 
-            if not is_safe:
-                continue
+                user_message = f"{c.author.name}]- {c.message}"
+                # print(user_message)
+                # print("Checking message safety...")
+                is_safe, _ = self.guardian.moderate_message(user_message)
 
-            self.tv_crew.add_context(user_message)
-            image_base64, caption = self.tv_crew.suggest_image()
-            if image_base64:
-                is_safe, _ = self.guardian.moderate_message(caption)
                 if not is_safe:
-                    image_base64 = None
-                    caption = None
-                else:
-                    user_message += f"[Crew Updated Television to: {caption}]\n"
+                    # print("Message flagged by Guardian, skipping...")
+                    continue
 
-            self.messages.append({"role": "user", "content": user_message})
-            try:
-                response = self.client.responses.create(
-                    model=self.model, input=self.messages
-                )
-                host_response = response.output_text
-                self.messages.append({"role": "assistant", "content": host_response})
-            except Exception as e:
-                print(f"Error generating host response: {e}")
-                continue
+                self.tv_crew.add_context(user_message)
+                # print("TV Crew suggesting image...")
+                image_base64, caption = self.tv_crew.suggest_image()
+                if image_base64:
+                    # print("Checking image caption safety...")
+                    is_safe, _ = self.guardian.moderate_message(caption)
+                    if not is_safe:
+                        # print("Image caption flagged by Guardian, skipping image...")
+                        image_base64 = None
+                        caption = None
+                    else:
+                        user_message += f"[Crew Updated Television to: {caption}]\n"
+
+                self.messages.append({"role": "user", "content": user_message})
+                try:
+                    # print("Generating host response...")
+                    response = self.client.responses.create(
+                        model=self.model, input=self.messages
+                    )
+                    host_response = response.output_text
+                    self.messages.append(
+                        {"role": "assistant", "content": host_response}
+                    )
+                except Exception as e:
+                    # print(f"Error generating host response: {e}")
+                    continue
 
-            self.tv_crew.add_context(f"Host: {host_response}\n")
-            # Just to make sure, that we're in the lobby
-            request = requests.get(
-                f"{unreal_orchestrator_url}/status", headers=self.headers
-            )
-            in_lobby = request.json().get("in_lobby", False)
-            if not in_lobby:
-                continue
+                self.tv_crew.add_context(f"Host: {host_response}\n")
+                # print(f"Host Response: {host_response}")
+                # Just to make sure, that we're in the lobby
+                request = requests.get(
+                    f"{unreal_orchestrator_url}/status", headers=self.headers
+                )
+                in_lobby = request.json().get("in_lobby", False)
+                if not in_lobby:
+                    continue
 
-            if image_base64:
-                payload_tv = {"base64": image_base64}
+                if image_base64:
+                    # print("Sending Image Unreal Orchestrator...")
+                    payload_tv = {"base64": image_base64}
+                    response = requests.post(
+                        url=f"{unreal_orchestrator_url}/television",
+                        json=payload_tv,
+                        headers=self.headers,
+                    )
+                # print("Sending Host response to Unreal Orchestrator...")
+                payload = {"text": host_response, "is_host": True}
                 response = requests.post(
-                    url=f"{unreal_orchestrator_url}/television",
-                    json=payload_tv,
+                    url=f"{unreal_orchestrator_url}/tts",
+                    json=payload,
                     headers=self.headers,
                 )
-
-            payload = {"text": host_response, "is_host": True}
-            response = requests.post(
-                url=f"{unreal_orchestrator_url}/tts",
-                json=payload,
-                headers=self.headers,
-            )
-
-    def get_live_chat_id(self, youtube, video_id):
-        try:
-            request = youtube.videos().list(part="liveStreamingDetails", id=video_id)
-            response = request.execute()
-            if not response["items"]:
-                print("No video found with the given ID.")
-                return None
-
-            details = response["items"][0]["liveStreamingDetails"]
-            if not details:
-                print("No live streaming details found for this video.")
-                return None
-            return details.get("activeLiveChatId", None)
-        except Exception as e:
-            print(f"An error occurred: {e}")
-            return None