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 urllib.request
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"
        self.headers = {
            'Authorization': 'f35fc9d7ce0549a88f6cdc15ec860b6e',
            'X-User-ID': '96tPb0H2cXbobV9u8iLVGyJPUPc2',
            'Content-Type': 'application/json'
        }
        
    def test(self, text, language='en', **kwargs):
        payload = json.dumps({
            "title": "Testing public api convertion",
            "voice": "en-US-MichelleNeural",
            "content": [text],
        })
        get_url = self.url+f'/convert'
        response = requests.request(
            "POST", 
            get_url, 
            headers=self.headers, 
            data=payload)
        
        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,
                )
                
            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(0.5)
                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)
        
        # 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