App/TTS/utils/CharacterAi.py

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())