import modal from dotenv import dotenv_values ffmpeg_worker_image = ( modal.Image.debian_slim(python_version="3.11") .apt_install('ffmpeg') .pip_install_from_pyproject("pyproject.toml") .env(dotenv_values("../../.runtime.env")) .add_local_python_source('cluster') .add_local_python_source('BowongModalFunctions') ) app = modal.App( name="ffmpeg_app", image=ffmpeg_worker_image, include_source=False) with ffmpeg_worker_image.imports(): import shutil, psutil, os, backoff, sentry_sdk from typing import List, Optional, Tuple, Any from loguru import logger from modal import current_function_call_id from ffmpeg.asyncio import FFmpeg from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.PathUtils import FileUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment from BowongModalFunctions.models.media_model import MediaSources, MediaSource, MediaProtocol from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify from BowongModalFunctions.config import WorkerConfig config = WorkerConfig() s3_mount = config.S3_mount_dir output_path_prefix = config.local_output_mount sentry_sdk.init( dsn="https://75cca970bfcc3d45d24361e7c0f1833c@sentry.bowongai.com/4", send_default_pii=True, add_full_stack=True, shutdown_timeout=2, traces_sample_rate=1.0, environment=config.modal_environment, ) @sentry_sdk.trace def capture_hardware_info() -> Tuple[Any, int, Any]: cpu_freq = psutil.cpu_freq() cpu_count = psutil.cpu_count() mem = psutil.virtual_memory() return (cpu_freq, cpu_count, mem) def ffmpeg_init(use_ffprobe: bool = False) -> FFmpeg: if use_ffprobe: ffmpeg_cmd = FFmpeg('ffprobe') else: ffmpeg_cmd = FFmpeg().option('y').option('hide_banner') @ffmpeg_cmd.on("start") def on_start(arguments: list[str]): try: filter_index = arguments.index("-filter_complex") filter_content = arguments[filter_index + 1] arguments[filter_index + 1] = f'"{filter_content}"' args = " ".join(arguments) logger.info(f"FFmpeg command:{args}") arguments[filter_index + 1] = filter_content except ValueError: args = " ".join(arguments) logger.info(f"FFmpeg command:{args}") @ffmpeg_cmd.on("progress") def on_progress(progress): logger.info(f"处理进度: {progress}") @ffmpeg_cmd.on("completed") def on_completed(): logger.info(f"FFMpeg task completed.") @ffmpeg_cmd.on("stderr") def on_stderr(line: str): if line.startswith('Error'): logger.error(line) raise RuntimeError(line) else: logger.warning(line) return ffmpeg_cmd @backoff.on_exception(backoff.constant, exception=Exception, max_tries=5, max_time=60, raise_on_giveup=True) def local_copy_to_s3(local_outputs: List[str]) -> List[str]: s3_outputs = [] for output in local_outputs: out_s3 = output.replace(output_path_prefix, s3_mount) out_s3_dir = os.path.dirname(out_s3) logger.info(out_s3_dir) os.makedirs(out_s3_dir, exist_ok=True) shutil.copy(output, out_s3) s3_outputs.append( out_s3.replace(s3_mount + '/', f"s3://{config.S3_region}/{config.S3_bucket_name}/")) return s3_outputs @app.function( timeout=1800, cloud="aws", # region='ap-northeast', 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_concat_medias(medias: MediaSources, sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None) -> Tuple[ str, Optional[SentryTransactionInfo]]: fn_id = current_function_call_id() @SentryUtils.sentry_tracker(name="视频合并任务", op="ffmpeg.concat", 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_sources: MediaSources, output_filepath: str) -> str: input_videos = [f"{s3_mount}/{media_source.cache_filepath}" for media_source in media_sources.inputs] local_output_path = await VideoUtils.ffmpeg_concat_medias(media_paths=input_videos, output_path=output_filepath) s3_outputs = local_copy_to_s3([local_output_path]) return s3_outputs[0] output_path = f"{output_path_prefix}/{config.modal_environment}/concat/outputs/{fn_id}/output.mp4" s3_output = await ffmpeg_process(media_sources=medias, output_filepath=output_path) if not sentry_trace: sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(), x_baggage=sentry_sdk.get_baggage()) return s3_output, sentry_trace @app.function( timeout=900, 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_slice_media(media: MediaSource, markers: List[FFMpegSliceSegment], sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None) -> Tuple[ List[str], Optional[SentryTransactionInfo]]: fn_id = current_function_call_id() @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) @SentryUtils.webhook_handler(webhook=webhook, func_id=fn_id) 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) 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], 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) 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) case _: outputs = await ffmpeg_slice_process(media_source=media, media_markers=markers, fn_id=fn_id) return outputs, 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_extract_audio(media_source: MediaSource, sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None) -> Tuple[ Optional[str], Optional[SentryTransactionInfo]]: fn_id = current_function_call_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) s3_outputs = local_copy_to_s3([output_path]) return s3_outputs[0] match media_source.protocol: case MediaProtocol.hls: return None, sentry_trace case _: output = await ffmpeg_process(media_source, fn_id=fn_id) return output, 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_corner_mirror(media: MediaSource, mirror_scale_down_size: int = 6, mirror_from_right: bool = True, mirror_position: tuple[float, float] = (40, 40), sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None): fn_id = current_function_call_id() @SentryUtils.sentry_tracker(name="视频镜像去重", op="ffmpeg.mirror", 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, func_id: str, mirror_scale_down_size: int = 6, mirror_from_right: bool = True, mirror_position: tuple[float, float] = (40, 40)) -> str: media_filepath = f"{s3_mount}/{media.cache_filepath}" output_path = f"{output_path_prefix}/{config.modal_environment}/corner_mirror/outputs/{func_id}/output.mp4" local_output_filepath = await VideoUtils.ffmpeg_corner_mirror(media_path=media_filepath, output_path=output_path, mirror_from_right=mirror_from_right, mirror_position=mirror_position, mirror_scale_down_size=mirror_scale_down_size) s3_outputs = local_copy_to_s3([local_output_filepath]) return s3_outputs[0] result = await ffmpeg_process(media=media, mirror_scale_down_size=mirror_scale_down_size, func_id=fn_id, mirror_from_right=mirror_from_right, mirror_position=mirror_position) if not sentry_trace: 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_overlay_gif(media: MediaSource, gif: MediaSource, sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None): fn_id = current_function_call_id() @SentryUtils.sentry_tracker(name="视频叠加GIF特效", op="ffmpeg.overlay", 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, func_id: str, gif: MediaSource) -> str: media_filepath = f"{s3_mount}/{media.cache_filepath}" gif_filepath = f"{s3_mount}/{gif.cache_filepath}" output_path = f"{output_path_prefix}/{config.modal_environment}/overlay/outputs/{func_id}/output.mp4" local_output_filepath = await VideoUtils.ffmpeg_overlay_gif(media_path=media_filepath, output_path=output_path, overlay_gif_path=gif_filepath) s3_outputs = local_copy_to_s3([local_output_filepath]) return s3_outputs[0] result = await ffmpeg_process(media=media, func_id=fn_id, gif=gif) if not sentry_trace: 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_zoom_loop(media: MediaSource, duration: int = 6, zoom: float = 0.1, sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None): fn_id = current_function_call_id() @SentryUtils.sentry_tracker(name="视频放大缩小去重", op="ffmpeg.zoom.loop", 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, func_id: str, duration: int = 6, zoom: float = 0.1) -> str: media_filepath = f"{s3_mount}/{media.cache_filepath}" output_path = f"{output_path_prefix}/{config.modal_environment}/zoom_loop/outputs/{func_id}/output.mp4" local_output_filepath = await VideoUtils.ffmpeg_zoom_loop(media_path=media_filepath, output_path=output_path, duration=duration, zoom=zoom) s3_outputs = local_copy_to_s3([local_output_filepath]) return s3_outputs[0] result = await ffmpeg_process(media=media, duration=duration, zoom=zoom, 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 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_bgm_nosie_reduce(media: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None, video_volume: float = 1.4, music_volume: float = 0.1, sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None): fn_id = current_function_call_id() @SentryUtils.sentry_tracker(name="视频混合BGM降噪", op="ffmpeg.bgm.nosie.reduce", 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(video: MediaSource, bgm: MediaSource, func_id: str, noise_sample: Optional[MediaSource] = None) -> str: local_input_filepath = f"{s3_mount}/{video.cache_filepath}" bgm_filepath = f"{s3_mount}/{bgm.cache_filepath}" noise_sample_path = f"{s3_mount}/{noise_sample.cache_filepath}" if noise_sample else None output_path = f"{output_path_prefix}/{config.modal_environment}/bgm_nosie_reduce/outputs/{func_id}/output.mp4" local_output_filepath = await VideoUtils.ffmpeg_mix_bgm_with_noise_reduce(media_path=local_input_filepath, bgm_audio_path=bgm_filepath, video_volume=video_volume, music_volume=music_volume, noise_sample_path=noise_sample_path, output_path=output_path) s3_outputs = local_copy_to_s3([local_output_filepath]) return s3_outputs result = await ffmpeg_process(video=media, bgm=bgm, video_volume=video_volume, music_volume=music_volume, noise_sample=noise_sample) if not sentry_trace: 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_subtitle_apply(media: MediaSource, subtitle: MediaSource, fonts: List[MediaSource], sentry_trace: Optional[SentryTransactionInfo] = None, webhook: Optional[WebhookNotify] = None): fn_id = current_function_call_id() @SentryUtils.sentry_tracker(name="视频叠加字幕", op="ffmpeg.subtitle.apply", 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(video: MediaSource, subtitle: MediaSource, fonts: List[MediaSource], func_id: str) -> str: media_path = f"{s3_mount}/{video.cache_filepath}" subtitle_path = f"{s3_mount}/{subtitle.cache_filepath}" output_path = f"{output_path_prefix}/{config.modal_environment}/subtitle_apply/{func_id}/output.mp4" font_dir = f"{output_path_prefix}/{config.modal_environment}/subtitle_apply/{func_id}/fonts" os.makedirs(font_dir, exist_ok=True) local_fonts = [f"{s3_mount}/{font.cache_filepath}" for font in fonts] for font in local_fonts: font_filename = os.path.basename(font) shutil.copy(font, f"{font_dir}/{font_filename}") local_output = await VideoUtils.ffmpeg_subtitle_apply(media_path=media_path, subtitle_path=subtitle_path, font_dir=font_dir, output_path=output_path) s3_outputs = local_copy_to_s3([local_output]) return s3_outputs[0] result = await ffmpeg_process(video=media, subtitle=subtitle, fonts=fonts, 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 result, sentry_trace