# -*- encoding: utf-8 -*- ''' @Email : liaoxiju@inspur.com ''' import edge_tts import os import re import sys sys.path.append("../") sys.path.append("../..") import datetime from datetime import timedelta import logging import yaml import uvicorn from common import HAIRUO_ENV from common import HairuoEnv from fastapi import FastAPI, Request from agent_common_utils.logger import get_logger from agent_common_utils.function_monitor import log_function_call from model_utils import audio_driven_video import tempfile logger = get_logger("digithuman") dh_audio_driven_video = log_function_call(logger)(audio_driven_video) #dh_audio_driven_video = audio_driven_video app = FastAPI() def upload_to_s3(filepath): os.system(f's3cmd put --expiry-days 10 {filepath} -r s3://ihp/tmp/liao/service/edge-tts/result/') def text_to_wav(text): save_path = "result" if not os.path.exists(save_path): os.system(f"mkdir -p {save_path}") if text is None or type(text) != str or len(text) == 0: raise "input text is Error" with tempfile.TemporaryDirectory(dir=save_path) as tmp_dir: output_stream = edge_tts.Communicate(text.strip(), "zh-CN-XiaoyiNeural") wav_save_path = f"{tmp_dir}/text2audio.wav" output_stream.save_sync(wav_save_path) #return save_path, tmp_dir upload_to_s3(tmp_dir) return f"https://ihp.oss.cn-north-4.inspurcloudoss.com/tmp/liao/service/edge-tts/{wav_save_path}" @app.post("/hairuo/digithuman") async def digithuman(request: Request, req: dict): ''' 对输入文本文本描述，生成wav音频，然后调用dh-webui生成视频 ''' logger.info(" Verification authorization.") body_data = await request.body() body_data = body_data.decode("utf-8") logger.info(" {}".format(body_data)) logger.info(" input text.") try: text = req.get("text") ##edge tts生成音频，上传oss，获取音频url wav_url = text_to_wav(text) code, video_url = dh_audio_driven_video(wav_url) if code != 0: logger.info(" dh model error") resp = { "code": "1", "message": "dh model error", "result": '' } return resp resp = { "code": "0", "message": "success", "result": {"video_url":video_url} } return resp except: logger.info(" model error") resp = { "code": "1", "message": "dh model call error", "result": "" } return resp if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=14040)