added HeyGen

App/TTS/Schemas.py

@@ -1,23 +1,43 @@
-from pydantic import BaseModel,Field
-from typing import List,Optional
+from pydantic import BaseModel, Field, validator
+from typing import List, Optional
 import uuid
 
+
 class Speak(BaseModel):
     paragraphId: str = Field(default_factory=lambda: str(uuid.uuid4()))
     speaker: str
     text: str
-    voiceId: str = Field(default="c60166365edf46589657770d", alias="speaker") # Default speaker value
+    voiceId: str = Field(
+        default="c60166365edf46589657770d", alias="speaker"
+    )  # Default speaker value
 
     def __init__(self, **data):
-        data["text"] = data.get('text') if  '<speak>' in data.get('text') else f"<speak>{data.get('text')}</speak>"
+        data["text"] = (
+            data.get("text")
+            if "<speak>" in data.get("text")
+            else f"<speak>{data.get('text')}</speak>"
+        )
         super().__init__(**data)
 
 
+class HeyGenTTSRequest(BaseModel):
+    voice_id: str = Field(default="d7bbcdd6964c47bdaae26decade4a933")
+    rate: str = Field(default="1")
+    pitch: str = Field(default="-3%")
+    text: str = "Sample"
+
+    @validator("text")
+    def validate_age(cls, value, values):
+        if not "speak" in value:
+            return f'<speak> <voice name="{values.get("voice_id")}"><prosody rate="{values.get("rate")}" pitch="{values.get("pitch")}">{value}</prosody></voice></speak>'
+        else:
+            return value
+
 
 class TTSGenerateRequest(BaseModel):
     paragraphs: List[Speak]
     requestId: str = Field(default_factory=lambda: str(uuid.uuid4()))
-    workspaceId: str =Field(default_factory=lambda: str(uuid.uuid4()))
+    workspaceId: str = Field(default_factory=lambda: str(uuid.uuid4()))
 
 
 class StatusRequest(BaseModel):

App/TTS/TTSRoutes.py

@@ -1,27 +1,33 @@
 from fastapi import APIRouter
 
 
-from .Schemas import StatusRequest, TTSGenerateRequest
+from .Schemas import StatusRequest, TTSGenerateRequest, HeyGenTTSRequest
 from .utils.Podcastle import PodcastleAPI
+from .utils.HeyGen import HeygenAPI
 import os
 
 tts_router = APIRouter(tags=["TTS"])
 data = {"username": os.environ.get("USERNAME"), "password": os.environ.get("PASSWORD")}
 tts = PodcastleAPI(**data)
+data = {
+    "account": os.environ.get("HEYGEN_USERNAME"),
+    "password": os.environ.get("HEYGEN_PASSWORD"),
+}
+heyGentts = HeygenAPI(**data)
 
 
-#
 @tts_router.post("/generate_tts")
 async def generate_voice(req: TTSGenerateRequest):
     print("here --entered!")
     return await tts.make_request(req)
 
 
+@tts_router.post("/heygen_tts")
+async def generate_heygen_voice(req: HeyGenTTSRequest):
+    print("hey gen here")
+    return await heyGentts.tts_request(req)
+
+
 @tts_router.post("/status")
 async def search_id(req: StatusRequest):
     return await tts.check_status(req)
-
-
-# @tts_router.post("/search_text")
-# async def search_text(req: SearchRequest):
-#     return TextSearch(query=req.query)

App/TTS/utils/HeyGen.py

@@ -0,0 +1,82 @@
+import aiohttp
+from App.TTS.Schemas import HeyGenTTSRequest
+
+
+class HeygenAPI:
+    def __init__(self, account, password):
+        self.base_url = "https://api2.heygen.com/v1"
+        self.account = account
+        self.password = password
+        self.session = None
+        self.session_token = None
+
+    async def create_session(self):
+        self.session = aiohttp.ClientSession()
+
+    async def close_session(self):
+        if self.session:
+            await self.session.close()
+
+    async def login(self):
+        url = f"{self.base_url}/pacific/login"
+        payload = {
+            "login_type": "email",
+            "account": self.account,
+            "password": self.password,
+        }
+
+        if not self.session:
+            await self.create_session()
+
+        async with self.session.post(url, json=payload) as response:
+            response_data = await response.json()
+            self.session_token = response_data.get("data", {}).get("session_token")
+            return response_data
+
+    async def relogin(self):
+        # Function to relogin and update the session token
+        login_result = await self.login()
+        if login_result.get("code") == 100:
+            self.session_token = login_result["data"]["session_token"]
+            return True
+        return False
+
+    async def tts_request(self, req: HeyGenTTSRequest):
+        if not self.session_token or not self.session_token:
+            await self.login()
+
+        url = f"{self.base_url}/online/text_to_speech.generate"
+        headers = {
+            "content-type": "application/json",
+            "x-session-token": self.session_token,
+        }
+
+        tts_payload = {
+            "text_type": "ssml",
+            "output_format": "wav",
+            "text": req.text,
+            "voice_id": req.voice_id,
+            "settings": {},
+        }
+
+        async with self.session.post(
+            url, json=tts_payload, headers=headers
+        ) as response:
+            if response.status == 401:
+                # If a 401 error is encountered, relogin and retry the request
+                if await self.relogin():
+                    headers["x-session-token"] = self.session_token
+                    response = await self.session.post(
+                        url, json=tts_payload, headers=headers
+                    )
+
+            response_data = await response.json()
+            return response_data
+
+    async def __aenter__(self):
+        if not self.session:
+            await self.create_session()
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        await self.close_session()