Files
modalDeploy/src/cluster/ffmpeg_app.py
肖宇迪 225a003d45 新增video-loop-fill接口
* FIX : 修复ffmpeg_concat_medias没有添加inputs的bug
* FIX: ffmpeg_slice_media和ffmpeg_slice_stream_media新增marker时间范围检查,如切割点不在视频时长范围内会报错
* 新增video-loop-fill接口

---------

Merge request URL: https://g-ldyi2063.coding.net/p/dev/d/modalDeploy/git/merge/4740
Co-authored-by: shuohigh@gmail.com
2025-05-23 14:27:38 +08:00

476 lines
26 KiB
Python

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, os, backoff, sentry_sdk
from typing import List, Optional, Tuple
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,
)
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(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],
fn_id: str) -> List[str]:
hls_m3u8_url = media_source.path
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)
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"
output_path = 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[0]
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
@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