# -*- encoding: utf-8 -*- ''' @Email : liaoxiju@inspur.com ''' 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 logger = get_logger("digithuman") dh_audio_driven_video = log_function_call(logger)(audio_driven_video) app = FastAPI() @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: app_id = req.get("app_id") prompt = req.get("text") ##edge tts生成音频，上传oss，获取音频url。 TODO:待实现 wav_url = 'xxx' video_url, code = dh_audio_driven_video(wav_url) if not code: logger.info(" dh model error") resp = { "code": "1", "message": "dh model error", "result": '' } return resp resp = { "app_id": app_id, "code": "0", "message": "success", "result": 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)