ToolsAE.py

# -*- coding: utf-8 -*-
import os, stat, time, json, hashlib
import urllib3.request, requests
import subprocess

class IamToken(object):
        def __init__(self):
                self.filename = '/var/lib/asterisk/agi-bin/iam.token'
                self.expireTime = 60*60*11

        def get(self):
                self.__iamUpdate()
                with open(self.filename) as f:
                        key = f.read()
                        return key.rstrip()

        def __iamUpdate(self):
                modTimeIam = int(os.path.getmtime(self.filename))
                curTime = int(time.time())
                if modTimeIam < curTime-self.expireTime:
                        iamCmd = subprocess.Popen(['/usr/local/bin/yc','iam', 'create-token'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                        out, err = iamCmd.communicate()
                        with open(self.filename, "w") as f:
                                f.write(out.rstrip())

class YandexSpeechSTT(object):
        def __init__(self, folderId, iamToken):
                self.folderId = folderId
                self.iamToken = iamToken

        def __convertAudio(self, fileSrc, fileDst):
                convertCmd = subprocess.Popen(['/usr/bin/opusenc', fileSrc, fileDst], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                output, error = convertCmd.communicate()
                if convertCmd.returncode != 0:
                        raise Exception(error.strip())
                os.remove(fileSrc)

        def convertVoiceToText(self, fileSrc, topic):
                import urllib2
                fileDst = '/tmp/' + os.path.splitext(os.path.basename(fileSrc))[0] + '.ogg'
                self.__convertAudio(fileSrc, fileDst)

                with open(fileDst, "rb") as f:
                        data = f.read()

                params = "&".join([
                        "topic=%s" % topic,
                        "folderId=%s" % self.folderId,
                        "lang=ru-RU"
                ])

                url = urllib2.Request("https://stt.api.cloud.yandex.net/speech/v1/stt:recognize?%s" % params, data=data)
                url.add_header("Authorization", "Bearer %s" % self.iamToken)

                responseData = urllib2.urlopen(url).read().decode('UTF-8')
                decodedData = json.loads(responseData)

                if decodedData.get("error_code") is None:
                        os.remove(fileDst)
                        return decodedData.get("result")
                else:
                        raise Exception("Error_code: %s, message: %s" % (decodedData.get("error_code"), decodedData.get("error_message")))

class YandexSpeechTTS(object):
        def __init__(self, folderId, iamToken):
                self.folderId = folderId
                self.iamToken = iamToken
                self.dstPath = '/var/lib/asterisk/sounds/ru/ysc'
                self.dstMd5Path = '/var/lib/asterisk/sounds/ru/ysc/md5'

        def __synthesize(self, text):
                url = 'https://tts.api.cloud.yandex.net/speech/v1/tts:synthesize'
                headers = {'Authorization': 'Bearer ' + self.iamToken,}

                data = {
                        'text': text,
                        'lang': 'ru-RU',
                        'folderId': self.folderId,
                        'format': 'lpcm',
                        'emotion': 'neutral',
                        'speed': '1.0',
                        'voice': 'jane',
                        'sampleRateHertz': 48000,
                }

                resp = requests.post(url, headers=headers, data=data, stream=True, verify=False)
                if resp.status_code != 200:
                        raise Exception("Invalid response received: code: %d, message: %s" % (resp.status_code, resp.text))
                for chunk in resp.iter_content(chunk_size=None):
                        yield chunk

        def __textToHashFname(self, hash, text):
                with open(self.dstMd5Path + "/" + hash, "w") as f:
                        f.write(text)

        def __convertAudio(self, typ, fileSrc, fileDst, text):
                if typ == 'ast':
                        convertCmd = subprocess.Popen(['/usr/bin/sox', fileSrc, '-e', 'a-law', '-t', 'RAW', '-r', '8000', '-c', '1', fileDst], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                if typ == 'raw':
                        convertCmd = subprocess.Popen(['/usr/bin/sox', '-r', '48000', '-b', '16', '-e', 'signed-integer', '-c', '1', fileSrc, fileDst], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                        self.__textToHashFname(textToHash(text),text)
                output, error = convertCmd.communicate()
                if convertCmd.returncode != 0:
                        raise Exception(error.strip())

        def ToFile(self, filename, text):
                fileSrc = self.dstPath + "/" + filename + '.raw'
                fileDst = self.dstPath + "/" + filename + '.wav'
                fileAster = self.dstPath + "/" + filename + '.alaw'
                exPath = self.dstMd5Path + "/" + textToHash(text)

                if os.path.isfile(exPath) and os.path.isfile(fileAster):
                        return

                with open(fileSrc, "wb") as f:
                        for audio_content in self.__synthesize(text):
                                f.write(audio_content)

                self.__convertAudio('raw', fileSrc, fileDst, text)
                os.remove(fileSrc)
                self.__convertAudio('ast', fileDst, fileAster, text)
                os.remove(fileDst)

def textToHash(text):
        return hashlib.md5(text).hexdigest()