From f6c40deec6c424f2f42a8db9d445913056ac624f Mon Sep 17 00:00:00 2001 From: michael tse Date: Sun, 25 May 2025 17:04:56 +0800 Subject: [PATCH] feat: add custom audio file support --- app/models/schema.py | 3 +- app/services/task.py | 74 ++++++++++++++++++++++++++++++++----------- app/services/voice.py | 32 ++++++++++++++++++- 3 files changed, 88 insertions(+), 21 deletions(-) diff --git a/app/models/schema.py b/app/models/schema.py index 3696fa3..167bac0 100644 --- a/app/models/schema.py +++ b/app/models/schema.py @@ -81,7 +81,8 @@ class VideoParams(BaseModel): video_materials: Optional[List[MaterialInfo]] = ( None # Materials used to generate the video ) - + + custom_audio_file: Optional[str] = None # Custom audio file path, will ignore video_script and disable subtitle video_language: Optional[str] = "" # auto detect voice_name: Optional[str] = "" diff --git a/app/services/task.py b/app/services/task.py index 77ca908..33c35cc 100644 --- a/app/services/task.py +++ b/app/services/task.py @@ -71,30 +71,66 @@ def save_script_data(task_id, video_script, video_terms, params): def generate_audio(task_id, params, video_script): + ''' + Generate audio for the video script. + If a custom audio file is provided, it will be used directly. + There will be no subtitle maker object returned in this case. + Otherwise, TTS will be used to generate the audio. + Returns: + - audio_file: path to the generated or provided audio file + - audio_duration: duration of the audio in seconds + - sub_maker: subtitle maker object if TTS is used, None otherwise + ''' logger.info("\n\n## generating audio") - audio_file = path.join(utils.task_dir(task_id), "audio.mp3") - sub_maker = voice.tts( - text=video_script, - voice_name=voice.parse_voice_name(params.voice_name), - voice_rate=params.voice_rate, - voice_file=audio_file, - ) - if sub_maker is None: - sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) - logger.error( - """failed to generate audio: + custom_audio_file = params.custom_audio_file + if not custom_audio_file or not os.path.exists(custom_audio_file): + if custom_audio_file: + logger.warning( + f"custom audio file not found: {custom_audio_file}, using TTS to generate audio." + ) + else: + logger.info("no custom audio file provided, using TTS to generate audio.") + audio_file = path.join(utils.task_dir(task_id), "audio.mp3") + sub_maker = voice.tts( + text=video_script, + voice_name=voice.parse_voice_name(params.voice_name), + voice_rate=params.voice_rate, + voice_file=audio_file, + ) + if sub_maker is None: + sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) + logger.error( + """failed to generate audio: 1. check if the language of the voice matches the language of the video script. 2. check if the network is available. If you are in China, it is recommended to use a VPN and enable the global traffic mode. - """.strip() - ) - return None, None, None - - audio_duration = math.ceil(voice.get_audio_duration(sub_maker)) - return audio_file, audio_duration, sub_maker - + """.strip() + ) + return None, None, None + audio_duration = math.ceil(voice.get_audio_duration(sub_maker)) + if audio_duration == 0: + sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) + logger.error("failed to get audio duration.") + return None, None, None + return audio_file, audio_duration, sub_maker + else: + logger.info(f"using custom audio file: {custom_audio_file}") + audio_duration = voice.get_audio_duration(custom_audio_file) + if audio_duration == 0: + sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) + logger.error("failed to get audio duration from custom audio file.") + return None, None, None + return custom_audio_file, audio_duration, None def generate_subtitle(task_id, params, video_script, sub_maker, audio_file): - if not params.subtitle_enabled: + ''' + Generate subtitle for the video script. + If subtitle generation is disabled or no subtitle maker is provided, it will return an empty string. + Otherwise, it will generate the subtitle using the specified provider. + Returns: + - subtitle_path: path to the generated subtitle file + ''' + logger.info("\n\n## generating subtitle") + if not params.subtitle_enabled or sub_maker is None: return "" subtitle_path = path.join(utils.task_dir(task_id), "subtitle.srt") diff --git a/app/services/voice.py b/app/services/voice.py index e6b4d59..3151c20 100644 --- a/app/services/voice.py +++ b/app/services/voice.py @@ -11,6 +11,7 @@ from edge_tts import SubMaker, submaker from edge_tts.submaker import mktimestamp from loguru import logger from moviepy.video.tools import subtitles +from moviepy.audio.io.AudioFileClip import AudioFileClip from app.config import config from app.utils import utils @@ -1486,7 +1487,7 @@ def create_subtitle(sub_maker: submaker.SubMaker, text: str, subtitle_file: str) logger.error(f"failed, error: {str(e)}") -def get_audio_duration(sub_maker: submaker.SubMaker): +def _get_audio_duration_from_submaker(sub_maker: submaker.SubMaker): """ 获取音频时长 """ @@ -1494,6 +1495,35 @@ def get_audio_duration(sub_maker: submaker.SubMaker): return 0.0 return sub_maker.offset[-1][1] / 10000000 +def _get_audio_duration_from_mp3(mp3_file: str) -> float: + """ + 获取MP3音频时长 + """ + if not os.path.exists(mp3_file): + logger.error(f"MP3 file does not exist: {mp3_file}") + return 0.0 + + try: + # Use moviepy to get the duration of the MP3 file + with AudioFileClip(mp3_file) as audio: + return audio.duration # Duration in seconds + except Exception as e: + logger.error(f"Failed to get audio duration from MP3: {str(e)}") + return 0.0 + +def get_audio_duration( target: Union[str, submaker.SubMaker]) -> float: + """ + 获取音频时长 + 如果是SubMaker对象，则从SubMaker中获取时长 + 如果是MP3文件，则从MP3文件中获取时长 + """ + if isinstance(target, submaker.SubMaker): + return _get_audio_duration_from_submaker(target) + elif isinstance(target, str) and target.endswith(".mp3"): + return _get_audio_duration_from_mp3(target) + else: + logger.error(f"Invalid target type: {type(target)}") + return 0.0 if __name__ == "__main__": voice_name = "zh-CN-XiaoxiaoMultilingualNeural-V2-Female"