File size: 5,318 Bytes
c4ed5be |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 |
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())
|