ai_qback/tools/audio.py

# -*- coding: utf-8 -*-

from pathlib import Path
from threading import Lock
import io
import os 
import re
import json
from cachetools import TTLCache
from hashlib import md5

from core.respone_format import *
from tools.new_mysql import MySQLUploader
from tools.loglog import logger
from tools.thread_pool_manager import pool_executor

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
import oss2
import nls
from oss2.credentials import EnvironmentVariableCredentialsProvider

Path("data/speech_data").mkdir(parents=True, exist_ok=True)


class TestTts:
    def __init__(self, token):
       
        self.appkey = "EwztMPbSeu5hTrss"
        self.token = token
        self.url = "wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"

    def start(self, text, local_file_path):
        self.__text = text
        self.__f = open(local_file_path, "wb")
        self.__test_run()
       
       

    def test_on_metainfo(self, message, *args):
        print("on_metainfo message=>{}".format(message))

    def test_on_error(self, message, *args):
        print("on_error args=>{}".format(args))

    def test_on_close(self, *args):
        try:
            self.__f.close()
        except Exception as e:
            print("close file failed since:", e)

    def test_on_data(self, data, *args):
        try:
            self.__f.write(data)
        except Exception as e:
            print("write data failed:", e)

    def test_on_completed(self, message, *args):
        print("on_completed:args=>{} message=>{}".format(args, message))

    def __test_run(self):
        tts = nls.NlsSpeechSynthesizer(url=self.url,
                                       token=self.token,
                                       appkey=self.appkey,
                                      
                                       on_data=self.test_on_data,
                                      
                                       on_error=self.test_on_error,
                                       on_close=self.test_on_close,
                                       )
        tts.start(self.__text, voice="Eva", aformat="mp3", speech_rate=-400)


class GetAudio:
    def __init__(self):
        self.m = MySQLUploader()
        self.auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
        self.bucket_name = 'public-qingti-data'
        self.bucket = oss2.Bucket(self.auth, 'oss-cn-hangzhou.aliyuncs.com', self.bucket_name)

        self.token_cache = TTLCache(maxsize=10, ttl=28800) 
        self.token = None
        self.lock = Lock()

        self.re_compile = re.compile("[.!?;*\"]")

   
    def upload_file_to_oss(self, word_or_hash_name, oss_file_name, local_file_path):
        for _ in range(2):
            try:
                r = self.bucket.put_object_from_file(oss_file_name, local_file_path)
            except Exception as e:
                logger.error(f"上传文件错误{type(e).__name__} {e}")
                continue
            s = r.resp.status
            if s == 200:
                print(f"音频上传oss成功 {word_or_hash_name}")
                os.remove(local_file_path) 
                return True
        else:
            logger.critical(f"2次上传oss错误,音频:{word_or_hash_name}")

   
    def insert_mysql(self, word_or_hash_name,oss_file_name,original_text):
        s = "insert into tts (word_or_hash_name,audio,long_tts_text) values (%s,%s,%s)"
        self.m.execute_(s, (word_or_hash_name,oss_file_name,original_text))

   
    def query_mysql(self, word):
        s = "select audio from tts where word_or_hash_name=%s"
        r = self.m.query_data(s, (word,))
        if not r:
            return False
        return r[0][0]

    @staticmethod
    def __create_token():
       
        client = AcsClient(
            os.getenv('OSS_ACCESS_KEY_ID'),
            os.getenv('OSS_ACCESS_KEY_SECRET'),
            "cn-shanghai"
        )
       
        request = CommonRequest()
        request.set_method('POST')
        request.set_domain('nls-meta.cn-shanghai.aliyuncs.com')
        request.set_version('2019-02-28')
        request.set_action_name('CreateToken')

        try:
            response = client.do_action_with_exception(request)

            jss = json.loads(response)
            if 'Token' in jss and 'Id' in jss['Token']:
                token = jss['Token']['Id']
               
               
               
                logger.info(f"生成token成功。{token}")
                return token

        except Exception as e:
            logger.error(f"token生成错误：{type(e).__name__} {e}")

   
    def get_audio(self, word, local_file_path):
        self.lock.acquire()
        try:
            if "token" not in self.token_cache:
                self.token = self.__create_token()
                self.token_cache["token"] = self.token
        except Exception as e:
            logger.error(f"{type(e).__name__} {e}")
        finally:
            self.lock.release()

        t = TestTts(self.token)
        t.start(word, local_file_path)

   
    def submit_task(self, word_or_phrase:str,resp_type:int):
        f = pool_executor.submit(self.run_task, word_or_phrase,resp_type)
        return f

   
    def __preprocess_data(self,text: str) -> tuple:
        """
        预处理数据。超过50个字符的所有名字都用哈希值。
        :param text: 预处理文本内容
        :return:
        """
        text = self.re_compile.sub("",text.strip()) 
        if len(text) > 30 or " " in text: 
            hash_str = md5(text.encode()).hexdigest()
            return text, hash_str
        return text,text

   
    def __resp_convert(self,oss_file_name,resp_type):
        """
        :param oss_file_name: OSS全路径；"baseData/audio/{word_or_hash_name}.mp3"
        :param resp_type: 回复格式设计：0返回oss路径，1 二进制文件，2 url三种；
        :return:
        """
        if resp_type == 0:
            logger.success(f"返回成功:oss路径{oss_file_name}")
            return oss_file_name
        elif resp_type == 1:
            try:
               
                obj = self.bucket.get_object(oss_file_name)
               
                content = io.BytesIO(obj.read())
                logger.success(f"返回成功:二进制数据{oss_file_name}")
                return content
            except Exception as e:
                raise resp_500(message=f"错误:转换二进制文件 {e}")
        elif resp_type == 2:
           

            mp3_file_url = f"https://{self.bucket_name}.oss-cn-hangzhou.aliyuncs.com/{oss_file_name}"
            logger.success(f"返回成功:https的url {mp3_file_url}")
            return mp3_file_url


   
    def run_task(self, word_or_phrase,resp_type):
        """
        :param word_or_phrase: 单词或短语文本，去生成tts的文本
        :param resp_type: 回复设计：0返回oss路径，1 二进制文件，2 url三种；
        :return:
        """
       
        original_text,word_or_hash_name = self.__preprocess_data(word_or_phrase)
        oss_file_name = f"baseData/audio/{word_or_hash_name}.mp3"
        local_file_path = f"data/speech_data/{word_or_hash_name}.mp3"

       
        query_word_mp3_path = self.query_mysql(word_or_hash_name)
        if query_word_mp3_path:
            resp_convertt = self.__resp_convert(oss_file_name=query_word_mp3_path,resp_type=resp_type)
            return resp_convertt

       
        self.get_audio(original_text, local_file_path)

       
        upload_result = self.upload_file_to_oss(word_or_hash_name, oss_file_name, local_file_path)
        if upload_result:
            self.insert_mysql(word_or_hash_name,oss_file_name,original_text)
            resp_convertt = self.__resp_convert(oss_file_name=oss_file_name, resp_type=resp_type)
            return resp_convertt

       
        return False


if __name__ == '__main__':
   
   

    import os

    os.chdir('..')
   
   
   
   

    g = GetAudio()
    article="""The Marches were a happy family. Poverty, hard work, and even the fact that Father March was away with the Union armies could not down the spirits of Meg, Jo, Amy, and Marmee, as the March girls called their mother. The March sisters tried to be good but had their share of faults. Pretty Meg was often displeased with the schoolchildren she taught; boyish Jo was easy to become angry; golden-haired schoolgirl Amy liked to show up; but Beth, who kept the house, was loving and gentle always. The happy days passed and darkness came when a telegram arrived for Mrs. March. "Your husband is very ill," it said, "come at once." The girl tried to be brave when their mother left for the front. They waited and prayed. Little Beth got scarlet fever when she was taking care of the sick neighbor. She became very ill but began to recover by the time Marmee was back. When Father came home from the front and at that joyful Christmas dinner they were once more all together. Three years later, the March girls had grown into young womanhood. Meg became Mrs. Brooke, and after a few family troubles got used to her new state happily. Jo had found pleasure in her literary efforts. Amy had grown into a young lady with a talent for design and an even greater one for society. But Beth had never fully regained her health, and her family watched her with love and anxiety. Amy was asked to go and stay in Europe with a relative of the Marches. Jo went to New York and became successful in her writing and had the satisfaction of seeing her work published there. But at home the bitterest blow was yet to fall. Beth had known for some time that she couldn't live much longer to be with the family and in the spring time she died. News came from Europe that Amy and Laurie, the grandson of a wealthy neighbor, had planned to be married soon. Now Jo became ever more successful in her writing and got married to Professor Bhaer and soon afterwards founded a school for boys. And so the little women had grown up and lived happily with their children, enjoying the harvest of love and goodness that they had devoted all their lives to."""
    g.get_audio(article, "1.mp3")