File size: 3,709 Bytes
1c49e44 ab1d89e 1c49e44 6ab5444 1c49e44 6ab5444 1c49e44 6ab5444 1c49e44 ab1d89e |
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 |
import aiohttp
from pydantic import BaseModel, Field, validator
from typing import List, Optional
import uuid
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 HeygenAPI:
def __init__(self, account, password, token):
self.base_url = "https://api2.heygen.com/v1"
self.account = account
self.password = password
self.token = token
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,
"token": self.token,
}
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()
async def main():
data = {
"account": "mebaxo5916@tospage.com",
"password": "HBBHN4RPs_rA$%R",
"token":'+DTX2g=='
}
async with HeygenAPI(**data) as heygen:
req = HeyGenTTSRequest(
text="Hello, this is a test",
# voice_id="1",
)
response = await heygen.tts_request(req)
print(response)
if __name__ == "__main__":
import asyncio
asyncio.run(main()) |