comi-logic/modalDeploy
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP： 分离出逻辑pip包

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-05-13 16:21:23 +08:00
parent 586122e3db
commit 0f2b9ddb16
32 changed files with 1153 additions and 699 deletions
Download Patch File Download Diff File Expand all files Collapse all files

466
src/cluster/ffmpeg_app.py Normal file
View File

@@ -0,0 +1,466 @@
import modal
ffmpeg_worker_image = (
modal.Image.debian_slim(python_version="3.11")
.apt_install('ffmpeg')
.pip_install(
'BowongModalFunctions==0.1.1-pre',
extra_index_url="https://packages-1747117713372:a8865382010036f1e8ad5ca8d0cdab520b78a8ff@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple",
)
)
ffmpeg_app = modal.App(image=ffmpeg_worker_image, include_source=False)
with ffmpeg_worker_image.imports():
import shutil, psutil, os, backoff, json, sentry_sdk
from typing import List, Optional, Tuple, Dict, 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
from BowongModalFunctions.config import WorkerConfig
config = WorkerConfig(
app_name="bowong-ai-video-test",
video_downloader_concurrency=10,
ffmpeg_worker_concurrency=10,
modal_kv_name="media-cache",
s3_region='ap-northeast-2',
s3_bucket_name="modal-media-cache",
environment="dev"
)
s3_mount = config.s3_mount_dir
output_path_prefix = "/mnt/outputs"
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.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
@ffmpeg_app.function(
timeout=1800,
cloud="aws",
# region='ap-northeast',
max_containers=config.ffmpeg_worker_concurrency,
volumes={
f"/mntS3": modal.CloudBucketMount(
bucket_name=config.s3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.environment),
),
}, )
@modal.concurrent(max_inputs=1)
async def ffmpeg_concat_medias(medias: MediaSources,
sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[
str, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@sentry_sdk.trace
async def print_media_streams_info(media_paths: List[str]) -> List[Dict[str, Any]]:
"""
打印多个媒体文件的视频和音频流详细信息
:param media_paths: 媒体文件路径列表
"""
res = []
for path in media_paths:
try:
# 使用ffprobe探测媒体文件信息
probe_info_byte = await FFmpeg(executable="ffprobe").input(path, print_format="json",
show_streams=None).execute()
probe_info = json.loads(probe_info_byte)
res.append(probe_info)
except Exception as e:
logger.exception(e)
continue
return res
@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)
async def ffmpeg_process(media_sources: MediaSources, output_filepath: str) -> str:
ffmpeg_cmd = VideoUtils.ffmpeg_init_async()
input_videos = []
for media_source in media_sources.inputs:
cache_filepath = f"{s3_mount}/{media_source.cache_filepath}"
logger.info(cache_filepath)
ffmpeg_cmd.input(cache_filepath)
input_videos.append(cache_filepath)
target_width = 1080
target_height = 1920
# video_info = await print_media_streams_info(media_paths=input_videos)
filter_complex = []
total_videos = len(input_videos)
# 2. 统一所有视频的格式、分辨率和帧率
for i in range(total_videos):
filter_complex.extend(
[
# 先缩放到统一分辨率，然后设置帧率和格式
f"[{i}:v]scale={target_width}:{target_height}:force_original_aspect_ratio=decrease,"
f"pad={target_width}:{target_height}:(ow-iw)/2:(oh-ih)/2,"
f"setsar=1:1," # 新增强制设置SAR
f"fps=30,format=yuv420p[v{i}]",
f"[{i}:a]aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo[a{i}]",
]
)
# 3. 准备处理后的视频流和音频流的连接字符串
video_streams = "".join(f"[v{i}]" for i in range(total_videos))
audio_streams = "".join(f"[a{i}]" for i in range(total_videos))
# 4. 使用concat过滤器合并视频和音频
filter_complex.extend(
[
f"{video_streams}concat=n={total_videos}:v=1:a=0[vconcated]",
f"{audio_streams}concat=n={total_videos}:v=0:a=1[aconcated]",
]
)
ffmpeg_cmd.output(
output_filepath,
{
"filter_complex": ";".join(filter_complex),
"map": ["[vconcated]", "[aconcated]"],
"vcodec": "libx264",
"crf": 16,
"r": 30,
"acodec": "aac",
"ar": 44100,
"ac": 2,
"ab": "192k",
},
)
await ffmpeg_cmd.execute()
s3_outputs = local_copy_to_s3([output_filepath])
return s3_outputs[0]
output_path = f"{output_path_prefix}/concat/outputs/{fn_id}/output.mp4"
os.makedirs(os.path.dirname(output_path), exist_ok=True)
s3_output = await ffmpeg_process(medias, output_filepath=output_path)
return s3_output, sentry_trace
@ffmpeg_app.function(
timeout=900,
cloud="aws",
max_containers=config.ffmpeg_worker_concurrency,
volumes={
f"/mntS3": modal.CloudBucketMount(
bucket_name=config.s3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.environment),
),
},
)
@modal.concurrent(max_inputs=1)
async def ffmpeg_slice_media(media: MediaSource,
markers: List[FFMpegSliceSegment],
webhook: Optional[str], # todo : handle webhook callback
sentry_trace: Optional[SentryTransactionInfo] = 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)
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.ffmpeg_init_async()
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.ffmpeg_init_async()
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
@ffmpeg_app.function(
timeout=600,
cloud="aws",
max_containers=config.ffmpeg_worker_concurrency,
volumes={
f"/mntS3": modal.CloudBucketMount(
bucket_name=config.s3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.environment),
),
},
)
@modal.concurrent(max_inputs=1)
async def ffmpeg_extract_audio(media_source: MediaSource,
webhook: Optional[str], # todo handle webhook callback
sentry_trace: Optional[SentryTransactionInfo] = 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)
async def ffmpeg_process(media: MediaSource, fn_id: str) -> str:
cache_filepath = f"{s3_mount}/{media.cache_filepath}"
output_path = f"{output_path_prefix}/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
async def ffmpeg_overlay_gif(media: MediaSource, gif: MediaSource):
raise NotImplementedError
@ffmpeg_app.function(
timeout=600,
cloud="aws",
max_containers=config.ffmpeg_worker_concurrency,
volumes={
f"/mntS3": modal.CloudBucketMount(
bucket_name=config.s3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.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):
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)
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}"
video_metadata = VideoUtils.ffprobe_video_format(media_filepath)
mirror_x = (
f"main_w-overlay_w-{mirror_position[0]}"
if mirror_from_right
else str(mirror_position[0])
)
filter_complex = [
"[0:v]split[original][mirror]",
f"[mirror]hflip,scale=iw/{mirror_scale_down_size}:-1,format=rgba[flipped]",
"[flipped]split[fm1][fm2]",
f"[fm2]format=gray,geq=lum='255*(1-pow(min(1,2*sqrt(pow(X/W-0.5,2)+pow(Y/H-0.5,2))),1.5))':a='if(lt(pow(X/W-0.5,2)+pow(Y/H-0.5,2),0.15),(1-pow(2*sqrt(pow(X/W-0.5,2)+pow(Y/H-0.5,2)),1.5))*255,0)'[fm2Blur]",
"[fm1][fm2Blur]alphamerge[flipped_blured]",
f"[original][flipped_blured]overlay=x={mirror_x}:y=main_h-overlay_h-{mirror_position[1]}[video]",
]
output_path = f"{output_path_prefix}/corner_mirror/outputs/{func_id}/output.mp4"
os.makedirs(os.path.dirname(output_path), exist_ok=True)
ffmpeg_cmd = VideoUtils.ffmpeg_init_async()
ffmpeg_cmd.input(media_filepath).output(output_path,
options={"filter_complex": ";".join(filter_complex)},
map=["[video]", "0:a"],
vcodec="libx264",
crf=16,
b=video_metadata.video_bitrate, # 视频码率
r=video_metadata.video_frame_rate # 帧率
)
await ffmpeg_cmd.execute()
s3_outputs = local_copy_to_s3([output_path])
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
async def ffmpeg_zoom_loop(media: MediaSource, duration: int = 6, zoom: float = 0.1):
raise NotImplementedError
async def ffmpeg_bgm_nosie_reduce(media: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None):
fn_id = current_function_call_id()
async def ffmpeg_process(video: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None):
local_input_filepath = f"{s3_mount}/{media.cache_filepath}"
audio_filepath = FileUtils.file_path_extend(local_input_filepath, "audio").replace('mp4', 'wav')
output_audio_filepath = f"{s3_mount}/{audio_filepath}"
await VideoUtils.ffmpeg_extract_audio_async(local_input_filepath, output_audio_filepath)
# VideoUtils.noise_reduce()
raise NotImplementedError
raise NotImplementedError
async def ffmpeg_subtitle_apply(media: MediaSource, subtitle: MediaSource):
raise NotImplementedError