character ai

App/TTS/Schemas.py

@@ -30,6 +30,12 @@ class DescriptTranscript(BaseModel):
     file_extenstion: str = ".wav"
 
 
+# Define the request model
+class CharacterTTSRequest(BaseModel):
+    text: str
+    voice: Optional[str] = None
+
+
 class PiTTSRequest(BaseModel):
     text: str
     voice: Optional[str]

App/TTS/TTSRoutes.py

@@ -10,16 +10,18 @@ from .Schemas import (
     DescriptSfxRequest,
     DescriptTranscript,
     PiTTSRequest,
+    CharacterTTSRequest,
 )
 from .utils.Podcastle import PodcastleAPI
 from .utils.HeyGen import HeygenAPI
-
+from .utils.CharacterAi import CharacterTTS
 from .utils.Descript import DescriptTTS
 import os
 import asyncio
-from .utils.Pi import PiAIClient
 
-from fastapi import FastAPI, Request, HTTPException
+# from .utils.Pi import PiAIClient
+
+from fastapi import Request, HTTPException
 from fastapi.responses import StreamingResponse, FileResponse
 import os
 import aiofiles
@@ -35,7 +37,7 @@ data = {
 
 descript_tts = DescriptTTS()
 heyGentts = HeygenAPI(**data)
-pi = PiAIClient(headless=True)
+# pi = PiAIClient(headless=True)
 
 
 @tts_router.post("/generate_tts")
@@ -97,11 +99,17 @@ async def search_id(req: StatusRequest):
     return await tts.check_status(req)
 
 
-@tts_router.post("/pi_tts")
-async def pi_tts(req: PiTTSRequest):
-    if not pi.initialized:
-        await pi.setup()
-    return await pi.say(req.text, voice=req.voice)
+@tts_router.post("/cai_tts")
+async def cai_tts(req: CharacterTTSRequest):
+    cai = CharacterTTS()
+    return await cai.say(req)
+
+
+# @tts_router.post("/pi_tts")
+# async def pi_tts(req: PiTTSRequest):
+#     if not pi.initialized:
+#         await pi.setup()
+#     return await pi.say(req.text, voice=req.voice)
 
 
 @tts_router.get("/audio/{audio_name}")

App/TTS/utils/CharacterAi.py

@@ -0,0 +1,162 @@
+import asyncio
+import os
+from characterai import aiocai
+from typing import Optional
+from pydantic import BaseModel
+import aiohttp
+
+from App.TTS.Schemas import CharacterTTSRequest
+
+
+# class CharacterTTSRequest(BaseModel):
+#     text: str
+#     voice: Optional[str] = None
+
+
+# Voice ID selected by the user
+
+
+class CharacterTTS:
+    TTS_ENDPOINT = "https://neo.character.ai/multimodal/api/v1/memo/replay"
+
+    def __init__(self):
+        """
+        Initialize the CharacterTTS instance.
+
+        Args:
+            token (str): The authorization token for Character AI.
+        """
+        self.token = "110626c85c57978218fa0066f58da72807e179d2"
+        self.client = aiocai.Client(self.token)
+
+    async def say(
+        self,
+        request: CharacterTTSRequest,
+        char_id: str = "nrXc-uQ5vtunnBTqKxB39vwNyJg2TARqD9r2wHVRO4U",
+    ) -> str:
+        """
+        Send a message to the character AI and generate TTS.
+
+        Args:
+            char_id (str): The character ID to chat with.
+            request (CharacterTTSRequest): The TTS request containing text and optional voice ID.
+
+        Returns:
+            str: The replay URL for the generated TTS audio.
+        """
+        # Connect to the AI client and start a new chat
+        me = await self.client.get_me()
+        chat = await self.client.connect()
+        async with chat as chat_session:
+            new_chat, answer = await chat_session.new_chat(char_id, me.id)
+            chat_id = new_chat.chat_id
+
+            # Send the user's message
+            message = await chat_session.send_message(
+                char=char_id, chat_id=chat_id, text=request.text
+            )
+
+            # Extract necessary identifiers
+            room_id = chat_id
+            turn_id = message.turn_key.turn_id
+            primary_candidate = next(
+                (
+                    c
+                    for c in message.candidates
+                    if c.candidate_id == message.primary_candidate_id
+                ),
+                None,
+            )
+
+            if not primary_candidate:
+                raise Exception("Primary candidate not found in the message response.")
+
+            candidate_id = primary_candidate.candidate_id
+            voice_id = request.voice if request.voice else self.default_voice()
+
+            # Generate TTS and get the replay URL
+            replay_url = await self.generate_tts(
+                room_id, turn_id, candidate_id, voice_id
+            )
+
+            return replay_url
+
+    def default_voice(self) -> str:
+        """
+        Return a default voice ID if none is provided.
+
+        Returns:
+            str: The default voice ID.
+        """
+        return (
+            "0ef6c2d5-14e1-420c-a634-693c3f13fade"  # Replace with your default voice ID
+        )
+
+    async def generate_tts(
+        self, room_id: str, turn_id: str, candidate_id: str, voice_id: str
+    ) -> str:
+        """
+        Generate TTS by making a POST request to the TTS endpoint.
+
+        Args:
+            room_id (str): The chat room ID.
+            turn_id (str): The turn ID of the message.
+            candidate_id (str): The candidate ID of the message.
+            voice_id (str): The selected voice ID.
+
+        Returns:
+            str: The replay URL for the generated TTS audio.
+        """
+        payload = {
+            "roomId": room_id,
+            "turnId": turn_id,
+            "candidateId": candidate_id,
+            "voiceId": voice_id,
+        }
+
+        headers = {
+            "accept": "application/json, text/plain, */*",
+            "accept-language": "en-US,en;q=0.9",
+            "authorization": f"Token {self.token}",
+            "content-type": "application/json",
+            "origin": "https://character.ai",
+            "priority": "u=1, i",
+            "referer": "https://character.ai/",
+            "sec-ch-ua": '"Google Chrome";v="129", "Not=A?Brand";v="8", "Chromium";v="129"',
+            "sec-ch-ua-mobile": "?0",
+            "sec-ch-ua-platform": '"Windows"',
+            "sec-fetch-dest": "empty",
+            "sec-fetch-mode": "cors",
+            "sec-fetch-site": "same-site",
+            "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/129.0.0.0 Safari/537.36",
+        }
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                self.TTS_ENDPOINT, json=payload, headers=headers
+            ) as response:
+                if response.status == 200:
+                    data = await response.json()
+                    replay_url = data.get("replayUrl")
+                    if replay_url:
+                        print(f"TTS generated successfully. Replay URL: {replay_url}")
+                        return replay_url
+                    else:
+                        raise Exception("Replay URL not found in the response.")
+                else:
+                    error = await response.text()
+                    raise Exception(
+                        f"TTS generation failed with status {response.status}: {error}"
+                    )
+
+
+# async def main():
+#     cai = CharacterTTS()
+#     x = await cai.say(CharacterTTSRequest(text="Hello, how are you?"))
+#     print(x)
+
+
+# if __name__ == "__main__":
+#     asyncio.run(main())

Dockerfile

@@ -28,7 +28,6 @@ RUN apt-get update && \
 #copy requirements    
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
-RUN playwright install-deps   
 
 
 # Copy the application code
@@ -37,6 +36,6 @@ USER admin
 COPY --chown=admin . /srv
 
 # Command to run the application
-CMD playwright install && uvicorn App.app:app --host 0.0.0.0 --port 7860 --workers 1
+CMD uvicorn App.app:app --host 0.0.0.0 --port 7860 --workers 1
 # Expose the server port
 EXPOSE 7860

requirements.txt

@@ -30,7 +30,7 @@ glfw
 Pillow 
 numpy
 broken-source
-playwright
+git+https://github.com/Xtr4F/PyCharacterAI.git
 
 # git+https://github.com/snowby666/poe-api-wrapper.git@2c91207598ad930901cb9fb09734705056b7b6a9
 # git+https://github.com/Mbonea-Mjema/ballyregan.git