diff --git a/pyproject.toml b/pyproject.toml index 63c431a..3dec490 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "BowongModalFunctions" -dynamic = ["version"] +version = "0.1.4" authors = [ { name = "Yudi Xiao", email = "xyd@bowong.ai" }, ] @@ -31,6 +31,7 @@ dependencies = [ "modal>=0.76.3", "python-dotenv>=1.1.0", "python-multipart>=0.0.20", + "m3u8>=6.0.0", ] classifiers = [ "Programming Language :: Python :: 3", diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py index 658062f..b599893 100644 --- a/src/BowongModalFunctions/models/web_model.py +++ b/src/BowongModalFunctions/models/web_model.py @@ -309,6 +309,15 @@ class FFMPEGMixBgmWithNoiseReduceStatusResponse(BaseFFMPEGTaskStatusResponse): result: Optional[str] = Field(default=None, description="生成结果的URN") +class FFMPEGVideoLoopFillAudioRequest(BaseFFMPEGTaskRequest): + video: MediaSource = Field(description="用来填充的视频素材") + audio: MediaSource = Field(description="被填充的音频素材") + + +class FFMPEGVideoLoopFillAudioResponse(BaseFFMPEGTaskStatusResponse): + result: Optional[str] = Field(default=None, description="生成结果的URN") + + class ComfyTaskStatusResponse(BaseModel): taskId: str = Field(description="任务Id") status: TaskStatus = Field(description="任务运行状态") diff --git a/src/BowongModalFunctions/router/ffmpeg.py b/src/BowongModalFunctions/router/ffmpeg.py index da84f40..79ca1b1 100644 --- a/src/BowongModalFunctions/router/ffmpeg.py +++ b/src/BowongModalFunctions/router/ffmpeg.py @@ -17,6 +17,7 @@ from ..models.web_model import (FFMPEGSliceRequest, SentryTransactionHeader, FFMPEGZoomLoopRequest, FFMPEGSubtitleOverlayRequest, FFMPEGMixBgmWithNoiseReduceRequest, + FFMPEGVideoLoopFillAudioRequest, FFMPEGSliceTaskStatusResponse, FFMPEGConcatTaskStatusResponse, FFMPEGExtractAudioTaskStatusResponse, @@ -24,6 +25,7 @@ from ..models.web_model import (FFMPEGSliceRequest, SentryTransactionHeader, FFMPEGOverlayGifTaskStatusResponse, FFMPEGSubtitleTaskStatusResponse, FFMPEGZoomLoopTaskStatusResponse, + FFMPEGVideoLoopFillAudioResponse, FFMPEGMixBgmWithNoiseReduceStatusResponse) config = WorkerConfig() @@ -317,3 +319,34 @@ async def bgm_nosie_reduce_status(task_id: str, response: Response) -> FFMPEGMix response.headers["x-baggage"] = transaction.x_baggage return FFMPEGMixBgmWithNoiseReduceStatusResponse(taskId=task_id, status=task_status, code=code, error=reason, result=result) + + +@router.post('/video-loop-fill', tags=["发起任务"], summary="发起视频匹配音频长度任务", + description="发起视频匹配音频长度任务", dependencies=[Depends(verify_token)]) +async def video_loop_fill_audio(body: FFMPEGVideoLoopFillAudioRequest, + headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse: + fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_loop_fill", + environment_name=config.modal_environment) + sentry_trace = None + if headers.x_trace_id and headers.x_baggage: + sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage) + fn_call = fn.spawn(media=body.video, audio=body.audio, sentry_trace=sentry_trace, webhook=body.webhook) + return ModalTaskResponse(success=True, taskId=fn_call.object_id) + + +@router.get("/video-loop-fill/{task_id}", tags=["查询任务"], + summary="查询视频匹配音频长度任务状态", description="查询视频匹配音频长度任务状态", + responses={ + status.HTTP_200_OK: { + "description": "", + "headers": sentry_header_schema + }, + }, + dependencies=[Depends(verify_token)]) +async def video_loop_fill_audio_status(task_id: str, response: Response) -> FFMPEGVideoLoopFillAudioResponse: + task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id) + if transaction: + response.headers["x-trace-id"] = transaction.x_trace_id + response.headers["x-baggage"] = transaction.x_baggage + return FFMPEGVideoLoopFillAudioResponse(taskId=task_id, status=task_status, code=code, + error=reason, result=result) diff --git a/src/BowongModalFunctions/utils/VideoUtils.py b/src/BowongModalFunctions/utils/VideoUtils.py index 42ff966..b45096c 100644 --- a/src/BowongModalFunctions/utils/VideoUtils.py +++ b/src/BowongModalFunctions/utils/VideoUtils.py @@ -23,6 +23,7 @@ from pedalboard import ( from pedalboard.io import AudioFile from loguru import logger from .PathUtils import FileUtils +from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment class MediaStream(BaseModel): @@ -69,6 +70,9 @@ class VideoMetadata(BaseModel): class VideoUtils: + """ + python-ffmpeg package docs : https://python-ffmpeg.readthedocs.io/en/stable/ + """ @staticmethod def ffprobe_video_format(media_path: str) -> VideoStream: @@ -270,6 +274,120 @@ class VideoUtils: return ffmpeg_cmd + @staticmethod + async def ffmpeg_slice_media(media_path: str, media_markers: List[FFMpegSliceSegment], + output_path: Optional[str] = None) -> List[str]: + """ + 使用本地视频文件按时间段切割出分段视频 + :param media_path: 本地视频路径 + :param media_markers: 分段起始结束时间标记 + :param output_path: 最终输出文件路径, 片段会根据指定路径附加_1.mp4， _2.mp4等片段编号 + :return: 输出片段的本地路径 + """ + + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(media_path) + filter_complex: List[str] = [] + outputs: List[str] = [] + + if not output_path: + output_path = FileUtils.file_path_extend(media_path, "slice") + os.makedirs(os.path.dirname(output_path), exist_ok=True) + video_metadata = VideoUtils.ffprobe_video_format(media_path=media_path) + + for index, marker in enumerate(media_markers): + filter_complex.extend( + [ + f"[v:0]trim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},setpts=PTS-STARTPTS[cut{index}]", + f"[a:0]atrim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},asetpts=PTS-STARTPTS[acut{index}]", + ] + ) + ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex)) + + for i, marker in enumerate(media_markers): + if marker.start.total_seconds() > video_metadata.duration or marker.start.total_seconds() < 0: + raise ValueError( + f"第{i}个切割点起始点{marker.start.total_seconds()}s超出视频时长[0-{video_metadata.duration}s]范围") + if marker.end.total_seconds() > video_metadata.duration or marker.end.total_seconds() < 0: + raise ValueError( + f"第{i}个切割点结束点{marker.end.total_seconds()}s超出视频时长[0-{video_metadata.duration}s]范围") + segment_output_path = FileUtils.file_path_extend(output_path, str(i)) + outputs.append(segment_output_path) + ffmpeg_cmd.output(segment_output_path, + map=[f"[cut{i}]", f"[acut{i}]"], + reset_timestamps="1", + sc_threshold="0", + g="1", + force_key_frames="expr:gte(t,n_forced*1)", + vcodec="libx264", + acodec="aac", + crf=16, + r=30, + ) + + await ffmpeg_cmd.execute() + return outputs + + @staticmethod + async def ffmpeg_slice_stream_media(media_path: str, + media_markers: List[FFMpegSliceSegment], + output_path: Optional[str] = None) -> List[str]: + """ + 按时间分段切割HLS视频流 + :param media_path: hls manifest URL + :param media_markers: 分段起始结束时间标记 + :param output_path: 最终输出文件路径, 片段会根据指定路径附加_1.mp4， _2.mp4等片段编号 + :return: 输出片段的本地路径 + """ + import m3u8 + playlist = m3u8.load(media_path) + stream_total_duration: float = sum(segment.duration for segment in playlist.segments) + + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(media_path, + protocol_whitelist="file,http,https,tcp,tls", + reconnect="1", # 自动重连 + reconnect_streamed="1", + reconnect_delay_max="5") + filter_complex: List[str] = [] + outputs: List[str] = [] + + if not output_path: + output_path = FileUtils.file_path_extend(media_path, "slice") + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + for index, marker in enumerate(media_markers): + if marker.start.total_seconds() > stream_total_duration or marker.start.total_seconds() < 0: + raise ValueError( + f"第{index}个切割点起始点{marker.start.total_seconds()}s超出视频时长[0-{stream_total_duration}s]范围") + if marker.end.total_seconds() > stream_total_duration or marker.end.total_seconds() < 0: + raise ValueError( + f"第{index}个切割点结束点{marker.end.total_seconds()}s超出视频时长[0-{stream_total_duration}s]范围") + + filter_complex.extend( + [ + f"[v:0]trim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},setpts=PTS-STARTPTS[cut{index}]", + f"[a:0]atrim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},asetpts=PTS-STARTPTS[acut{index}]", + ] + ) + ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex)) + for i, marker in enumerate(media_markers): + output_filepath = FileUtils.file_path_extend(output_path, str(i)) + outputs.append(output_filepath) + ffmpeg_cmd.output(output_filepath, + map=[f"[cut{i}]", f"[acut{i}]"], + reset_timestamps="1", + sc_threshold="0", + g="1", + force_key_frames="expr:gte(t,n_forced*1)", + vcodec="libx264", + acodec="aac", + crf=16, + r=30, ) + + await ffmpeg_cmd.execute() + return outputs + @staticmethod async def ffmpeg_concat_medias(media_paths: List[str], target_width: int = 1080, @@ -292,7 +410,8 @@ class VideoUtils: os.makedirs(os.path.dirname(output_path), exist_ok=True) ffmpeg_cmd = VideoUtils.async_ffmpeg_init() filter_complex = [] - + for input_path in media_paths: + ffmpeg_cmd.input(input_path) # 2. 统一所有视频的格式、分辨率和帧率 for i in range(total_videos): filter_complex.extend( @@ -335,7 +454,7 @@ class VideoUtils: return output_path @staticmethod - async def ffmpeg_extract_audio_async(media_path: str, output_path: Optional[str] = None): + async def ffmpeg_extract_audio_async(media_path: str, output_path: Optional[str] = None) -> str: """ 提取源视频的音频 :param media_path: 待处理的源视频 @@ -365,6 +484,7 @@ class VideoUtils: ar=44100, ac=1) await ffmpeg_cmd.execute() + return output_path @staticmethod async def ffmpeg_mix_bgm(origin_audio_path: str, bgm_audio_path: str, video_volume: float = 1.4, @@ -594,3 +714,36 @@ class VideoUtils: ) await ffmpeg_cmd.execute() return output_path + + @staticmethod + async def ffmpeg_fill_longest(video_path: str, audio_path: str, output_path: Optional[str] = None) -> str: + """ + 用视频循环对齐音频时长，如果短于音频时长则循环填满音频时长，如短于音频时长则裁剪结尾 + :param video_path: 使用的视频文件路径 + :param audio_path: 匹配的音频文件路径 + :param output_path: 指定输出文件地址 + :return: 最终输出的文件路径 + """ + video_metadata = VideoUtils.ffprobe_video_format(video_path) + audio_duration = VideoUtils.ffprobe_audio_duration(audio_path) + loop_times = 0 if video_metadata.duration > audio_duration.total_seconds() else int( + math.ceil(audio_duration.total_seconds() / video_metadata.duration)) + logger.info( + f"视频长度 = {video_metadata.duration}, 音频长度 = {audio_duration.total_seconds()}, 重复 = {loop_times}") + + if not output_path: + output_path = FileUtils.file_path_extend(video_path, 'fill') + os.makedirs(os.path.dirname(output_path), exist_ok=True) + + ffmpeg_cmd = VideoUtils.async_ffmpeg_init() + ffmpeg_cmd.input(video_path, stream_loop=str(loop_times)) + ffmpeg_cmd.input(audio_path) + + ffmpeg_cmd.output(output_path, + map=["0:v", "1:a"], + vcodec="copy", + acodec="aac", + shortest=None, + ) + await ffmpeg_cmd.execute() + return output_path diff --git a/src/cluster/ffmpeg_app.py b/src/cluster/ffmpeg_app.py index e9cfd9e..f106e34 100644 --- a/src/cluster/ffmpeg_app.py +++ b/src/cluster/ffmpeg_app.py @@ -162,99 +162,37 @@ with ffmpeg_worker_image.imports(): async def ffmpeg_slice_process(media_source: MediaSource, media_markers: List[FFMpegSliceSegment], fn_id: str) -> List[str]: cache_filepath = f"{s3_mount}/{media_source.cache_filepath}" - logger.info(cache_filepath) - outputs: List[str] = [] - ffmpeg_cmd = VideoUtils.async_ffmpeg_init() - ffmpeg_cmd.input(cache_filepath) - filter_complex: List[str] = [] - for index, marker in enumerate(media_markers): - filter_complex.extend( - [ - f"[v:0]trim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},setpts=PTS-STARTPTS[cut{index}]", - f"[a:0]atrim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},asetpts=PTS-STARTPTS[acut{index}]", - ] - ) - ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex)) - - for i, marker in enumerate(media_markers): - filename = FileUtils.file_path_extend(os.path.basename(cache_filepath), str(i)) - output_filepath = f"{output_path_prefix}/slice/outputs/{fn_id}/{filename}" - output_filepath = output_filepath.replace('//', '/') - workdir = os.path.dirname(output_filepath) - os.makedirs(workdir, exist_ok=True) - outputs.append(output_filepath) - ffmpeg_cmd.output(output_filepath, - map=[f"[cut{i}]", f"[acut{i}]"], - reset_timestamps="1", - sc_threshold="0", - g="1", - force_key_frames="expr:gte(t,n_forced*1)", - vcodec="libx264", - acodec="aac", - crf=16, - r=30, ) - - await ffmpeg_cmd.execute() - - s3_outputs = local_copy_to_s3(outputs) - + logger.info(f"从{media_source.urn}切割") + for marker in media_markers: + logger.info(f"{marker.start.toFormatStr()} --> {marker.end.toFormatStr()}") + segments = await VideoUtils.ffmpeg_slice_media(media_path=cache_filepath, + media_markers=media_markers, + output_path=f"{output_path_prefix}/{config.modal_environment}/slice/outputs/{fn_id}/output.mp4") + s3_outputs = local_copy_to_s3(segments) return s3_outputs @SentryUtils.sentry_tracker(name="视频切割任务", op="ffmpeg.slice", fn_id=fn_id, sentry_trace_id=sentry_trace.x_trace_id if sentry_trace else None, sentry_baggage=sentry_trace.x_baggage if sentry_trace else None) - async def ffmpeg_hls_slice_process(media_source: MediaSource, media_markers: List[FFMpegSliceSegment], + async def ffmpeg_hls_slice_process(media_source: MediaSource, + media_markers: List[FFMpegSliceSegment], fn_id: str) -> List[str]: hls_m3u8_url = media_source.path - logger.info(hls_m3u8_url) - outputs: List[str] = [] - ffmpeg_cmd = VideoUtils.async_ffmpeg_init() - ffmpeg_cmd.input(hls_m3u8_url, - protocol_whitelist="file,http,https,tcp,tls", - reconnect="1", # 自动重连 - reconnect_streamed="1", - reconnect_delay_max="5") - filter_complex: List[str] = [] - for index, marker in enumerate(media_markers): - filter_complex.extend( - [ - f"[v:0]trim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},setpts=PTS-STARTPTS[cut{index}]", - f"[a:0]atrim=start={marker.start.total_seconds()}:end={marker.end.total_seconds()},asetpts=PTS-STARTPTS[acut{index}]", - ] - ) - ffmpeg_cmd.option('filter_complex', ';'.join(filter_complex)) - for i, marker in enumerate(media_markers): - paths = hls_m3u8_url.split(".")[:-1] - paths.append('mp4') - output_filepath = '.'.join(paths) - filename = FileUtils.file_path_extend(os.path.basename(output_filepath), str(i)) - output_filepath = f"{output_path_prefix}/slice/outputs/{fn_id}/{filename}" - output_filepath = output_filepath.replace('//', '/') - workdir = os.path.dirname(output_filepath) - os.makedirs(workdir, exist_ok=True) - outputs.append(output_filepath) - ffmpeg_cmd.output(output_filepath, - map=[f"[cut{i}]", f"[acut{i}]"], - reset_timestamps="1", - sc_threshold="0", - g="1", - force_key_frames="expr:gte(t,n_forced*1)", - vcodec="libx264", - acodec="aac", - crf=16, - r=30, ) - - await ffmpeg_cmd.execute() - - s3_outputs = local_copy_to_s3(outputs) - + segments = await VideoUtils.ffmpeg_slice_stream_media(media_path=hls_m3u8_url, + media_markers=media_markers, + output_path=f"{output_path_prefix}/{config.modal_environment}/slice/outputs/{fn_id}/output.mp4") + s3_outputs = local_copy_to_s3(segments) return s3_outputs match media.protocol: case MediaProtocol.hls: - outputs = await ffmpeg_hls_slice_process(media_source=media, media_markers=markers, fn_id=fn_id) + outputs = await ffmpeg_hls_slice_process(media_source=media, + media_markers=markers, + fn_id=fn_id) case _: - outputs = await ffmpeg_slice_process(media_source=media, media_markers=markers, fn_id=fn_id) + outputs = await ffmpeg_slice_process(media_source=media, + media_markers=markers, + fn_id=fn_id) return outputs, sentry_trace @@ -277,14 +215,14 @@ with ffmpeg_worker_image.imports(): fn_id = current_function_call_id() - @SentryUtils.sentry_tracker(name="视频切割任务", op="ffmpeg.extract.audio", fn_id=fn_id, + @SentryUtils.sentry_tracker(name="音频提取任务", op="ffmpeg.extract.audio", fn_id=fn_id, sentry_trace_id=sentry_trace.x_trace_id if sentry_trace else None, sentry_baggage=sentry_trace.x_baggage if sentry_trace else None) @SentryUtils.webhook_handler(webhook=webhook, func_id=fn_id) async def ffmpeg_process(media: MediaSource, fn_id: str) -> str: cache_filepath = f"{s3_mount}/{media.cache_filepath}" output_path = f"{output_path_prefix}/{config.modal_environment}/extract_audio/outputs/{fn_id}/output.wav" - await VideoUtils.ffmpeg_extract_audio_async(cache_filepath, output_path) + output_path = await VideoUtils.ffmpeg_extract_audio_async(cache_filepath, output_path) s3_outputs = local_copy_to_s3([output_path]) return s3_outputs[0] @@ -497,3 +435,41 @@ with ffmpeg_worker_image.imports(): sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(), x_baggage=sentry_sdk.get_baggage()) return result, sentry_trace + + + @app.function(timeout=600, cloud="aws", + max_containers=config.ffmpeg_worker_concurrency, + volumes={ + s3_mount: modal.CloudBucketMount( + bucket_name=config.S3_bucket_name, + secret=modal.Secret.from_name("aws-s3-secret", + environment_name=config.modal_environment), + ), + }, ) + @modal.concurrent(max_inputs=1) + async def ffmpeg_loop_fill(media: MediaSource, audio: MediaSource, + sentry_trace: Optional[SentryTransactionInfo] = None, + webhook: Optional[WebhookNotify] = None): + fn_id = current_function_call_id() + + @SentryUtils.sentry_tracker(name="视频匹配音频长度", op="ffmpeg.loop.fill", fn_id=fn_id, + sentry_trace_id=sentry_trace.x_trace_id if sentry_trace else None, + sentry_baggage=sentry_trace.x_baggage if sentry_trace else None) + @SentryUtils.webhook_handler(webhook=webhook, func_id=fn_id) + async def ffmpeg_process(media: MediaSource, audio: MediaSource, func_id: str) -> str: + video_path = f"{s3_mount}/{media.cache_filepath}" + audio_path = f"{s3_mount}/{audio.cache_filepath}" + + local_output = await VideoUtils.ffmpeg_fill_longest(video_path=video_path, + audio_path=audio_path, + output_path=f"{output_path_prefix}/{config.modal_environment}/loop_fill/{func_id}/output.mp4") + s3_outputs = local_copy_to_s3([local_output]) + return s3_outputs[0] + + output = await ffmpeg_process(media=media, audio=audio, func_id=fn_id) + + if not sentry_trace: + sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(), + x_baggage=sentry_sdk.get_baggage()) + + return output, sentry_trace diff --git a/uv.lock b/uv.lock index d893b3c..7ca6022 100644 --- a/uv.lock +++ b/uv.lock @@ -168,6 +168,7 @@ wheels = [ [[package]] name = "bowongmodalfunctions" +version = "0.1.4" source = { editable = "." } dependencies = [ { name = "backoff" }, @@ -177,6 +178,7 @@ dependencies = [ { name = "fastapi", extra = ["standard"] }, { name = "httpx" }, { name = "loguru" }, + { name = "m3u8" }, { name = "modal" }, { name = "noisereduce" }, { name = "pedalboard" }, @@ -212,6 +214,7 @@ requires-dist = [ { name = "gunicorn", marker = "extra == 'cli'", specifier = ">=23.0.0" }, { name = "httpx", specifier = ">=0.28.1" }, { name = "loguru", specifier = ">=0.7.3" }, + { name = "m3u8", specifier = ">=6.0.0" }, { name = "modal", specifier = ">=0.76.3" }, { name = "noisereduce", specifier = ">=3.0.3" }, { name = "pedalboard", specifier = "==0.9.2" }, @@ -988,6 +991,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0c/29/0348de65b8cc732daa3e33e67806420b2ae89bdce2b04af740289c5c6c8c/loguru-0.7.3-py3-none-any.whl", hash = "sha256:31a33c10c8e1e10422bfd431aeb5d351c7cf7fa671e3c4df004162264b28220c", size = 61595 }, ] +[[package]] +name = "m3u8" +version = "6.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9b/a5/73697aaa99bb32b610adc1f11d46a0c0c370351292e9b271755084a145e6/m3u8-6.0.0.tar.gz", hash = "sha256:7ade990a1667d7a653bcaf9413b16c3eb5cd618982ff46aaff57fe6d9fa9c0fd", size = 42720 } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f8/31/50f3c38b38ff28635ff9c4a4afefddccc5f1b57457b539bdbdf75ce18669/m3u8-6.0.0-py3-none-any.whl", hash = "sha256:566d0748739c552dad10f8c87150078de6a0ec25071fa48e6968e96fc6dcba5d", size = 24133 }, +] + [[package]] name = "markdown-it-py" version = "3.0.0"