add fastapi

app.py

@@ -13,6 +13,8 @@ from timing import TimeManager
 from validity import is_valid_rpm_url
 from yt_chat import StreamChatHost
 from prompts import system_prompt_tv_crew_guest, system_prompt_tv_crew_chat
+from fastapi import FastAPI, Request
+import uvicorn
 
 # Remove these before uploading on Spaces
 # import dotenv
@@ -50,13 +52,20 @@ streamChatHost = StreamChatHost(client, tv_crew_chat, guardian)
 thread_time_manager = threading.Thread(target=timeManager.guest_time_limit, daemon=True)
 thread_time_manager.start()
 
-thread_stream_chat = threading.Thread(
-    target=streamChatHost.chat_interaction_loop, daemon=True
-)
-thread_stream_chat.start()
+in_construction = False
 
+app = FastAPI()
 
-in_construction = False
+
+@app.post("/receive_chat")
+async def receive_chat(request: Request):
+    api_key = request.headers.get("x-api-key", None)
+    if api_key != api_key_unreal:
+        return {"status": "forbidden"}
+    data = await request.json()
+    user_message = f"{data['user']} - {data['message']}"
+    streamChatHost.reply_to_chat(user_message)
+    return {"status": "received"}
 
 
 def join_show(avatar_url):
@@ -327,4 +336,5 @@ src="https://www.youtube.com/embed/xBVJIJQg1FY?si=tsCm1DdajNCP45ho" title="YouTu
 
 
 if __name__ == "__main__":
-    demo.launch(mcp_server=True)
+    app = gr.mount_gradio_app(app, demo, path="/", mcp_server=True)
+    uvicorn.run(app, host="0.0.0.0", port=7860)

yt_chat.py

@@ -5,14 +5,12 @@ from guard import Guardian
 from tv_crew import TVCrew
 import requests
 import time
-import pytchat
 
 # load_dotenv()
 unreal_orchestrator_url = os.getenv("ORCHESTRATOR_URL")
 api_key_unreal = os.getenv("API_KEY_UNREAL")
 
 API_KEY = os.getenv("YOUTUBE_API_KEY")
-POLL_INTERVAL = 3  # seconds
 
 
 class StreamChatHost:
@@ -31,87 +29,74 @@ 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}
         VIDEO_ID = "xBVJIJQg1FY"
-        try:
-            self.chat = pytchat.create(video_id=VIDEO_ID)
-        except Exception as e:
-            print(f"Error creating pytchat object: {e}")
-            self.chat = None
 
-    def chat_interaction_loop(self):
+    def reply_to_chat(self, user_message):
         """Checks if the Host is in lobby, then pulls chat messages from the Live Stream. The Host then replies to the message."""
-        if not self.chat:
-            time.sleep(5)
+        request = requests.get(
+            f"{unreal_orchestrator_url}/status", headers=self.headers
+        )
+        in_lobby = request.json().get("in_lobby", False)
+        if not in_lobby:
             return
-        while self.chat.is_alive():
-            for c in self.chat.get().sync_items():
-                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()
+        if len(self.messages) > 20:
+            self.messages = [self.system_message]
+            self.tv_crew.clear_context()
 
-                user_message = f"{c.author.name}]- {c.message}"
-                # print(user_message)
-                # print("Checking message safety...")
-                is_safe, _ = self.guardian.moderate_message(user_message)
+        # print(user_message)
+        # print("Checking message safety...")
+        is_safe, _ = self.guardian.moderate_message(user_message)
 
-                if not is_safe:
-                    # print("Message flagged by Guardian, skipping...")
-                    continue
+        if not is_safe:
+            # print("Message flagged by Guardian, skipping...")
+            return
 
-                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.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.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}")
+            return
 
-                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
+        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:
+            return
 
-                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}/tts",
-                    json=payload,
-                    headers=self.headers,
-                )
+        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}/tts",
+            json=payload,
+            headers=self.headers,
+        )