src/utils/text2speech.py

import os
import tempfile
from TTS.api import TTS



class TTSTalker():
    def __init__(self) -> None:
        model_name = TTS.list_models()[0]
        self.tts = TTS(model_name)

    def test(self, text, language='en'):

        tempf  = tempfile.NamedTemporaryFile(
                delete = False,
                suffix = ('.'+'wav'),
            )

        self.tts.tts_to_file(text, speaker=self.tts.speakers[0], language=language, file_path=tempf.name)

        return tempf.name
    
import tempfile
import requests
import json
import time


class TTSTalkerPlayHT():
    def __init__(self) -> None:

        if 0:
            from easydict import EasyDict
            self = EasyDict()
            text = 'hello world'

        self.url = "https://play.ht/api/v1"

        
    def test(self, text, 
             ht_user_id = '96tPb0H2cXbobV9u8iLVGyJPUPc2', 
             ht_auth_key = 'f35fc9d7ce0549a88f6cdc15ec860b6e',  
             voice = 'en-US-MichelleNeural',
             **kwargs):

        print(f'ht_user_id {ht_user_id}')
        print(f'ht_auth_key {ht_auth_key}')
        
        self.headers = {
            'Authorization': ht_auth_key,
            'X-User-ID': ht_user_id,
            'Content-Type': 'application/json'
        }
        
        payload = json.dumps({
            "title": "Testing public api convertion",
            "voice": voice,
            "content": [text],
        })
        get_url = self.url+f'/convert'
        response = requests.request(
            "POST", 
            get_url, 
            headers=self.headers, 
            data=payload)
        
        print(f'convert response.status_code {response.status_code}')
        
        if response.status_code == 404:
            print('404')
            return
        
        # transcriptionId 如果成功是马上返回的
        data = json.loads(response.text)
        transcriptionId = data['transcriptionId']
        
        
        
        s_time = time.time()
        while time.time() - s_time < 10:

            if 0:
                get_url = self.url+f'/articleStatus?transcriptionId={transcriptionId}'
                response = requests.get(
                    get_url,
                    headers=self.headers,
                )
            else:
                get_url = self.url+f'/articleStatus'
                response = requests.get(
                    get_url,
                    params={
                        'transcriptionId': transcriptionId
                    },
                    headers=self.headers,
                )
                
            print(f'articleStatus response.status_code {response.status_code}')
                
            if response.status_code == 404:
                print(response.text)
                print('404')
                return
            
            # articleStatus返回的不一定马上就有audioUrl
            data = json.loads(response.text)
            converted = data['converted']
            if converted != True:
                time.sleep(1.0)
                continue
            
        # articleStatus 表示转换完成
        audioUrl = data['audioUrl']
            
        tempf  = tempfile.NamedTemporaryFile(
                delete = False,
                suffix = ('.'+'mp3'),
            )


        def download_dropbox_url(url, filepath, chunk_size=1024):

            import requests
            headers = {'user-agent': 'Wget/1.16 (linux-gnu)'}
            r = requests.get(url, stream=True, headers=headers)
            with open(filepath, 'wb') as f:
                for chunk in r.iter_content(chunk_size=chunk_size):
                    if chunk:
                        f.write(chunk)
            return filepath


        download_dropbox_url(audioUrl, tempf.name)
        
        # import urllib.request
        # urllib.request.urlretrieve(audioUrl, tempf.name)
        
        # response = requests.get(audioUrl)
        # with open(tempf.name, "wb") as f:
        #     f.write(response.content)
        
        # import subprocess
        # cmd = f'wget -O {tempf.name} {audioUrl}'
        # # ['wget', audioUrl, '-O', tempf.name]
        # subprocess.call(cmd)
        
        return tempf.name