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

添加webhook回调处理

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-05-16 11:21:02 +08:00
parent 9f25ec0ced
commit 281dff7966
9 changed files with 118 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@@ -3,7 +3,7 @@
## 部署 ## 部署
```bash ```bash
modal deploy -e dev ./src/cluster/app.py modal deploy -m cluster.app
``` ```
# Pip # Pip

2
.idea/runConfigurations/modal_deploy.xml → cmd/modal deploy.run.xml
View File

@@ -1,7 +1,7 @@
<component name="ProjectRunConfigurationManager"> <component name="ProjectRunConfigurationManager">
<configuration default="false" name="modal deploy" type="PythonConfigurationType" factoryName="Python"> <configuration default="false" name="modal deploy" type="PythonConfigurationType" factoryName="Python">
<module name="modalDeploy" /> <module name="modalDeploy" />
<option name="ENV_FILES" value="$PROJECT_DIR$/.env" /> <option name="ENV_FILES" value="$PROJECT_DIR$/.runtime.env" />
<option name="INTERPRETER_OPTIONS" value="" /> <option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" /> <option name="PARENT_ENVS" value="true" />
<envs> <envs>

5
src/BowongModalFunctions/api.py
View File

@@ -180,7 +180,6 @@ async def get_modal_task_status(task_id: str) -> Tuple[
task = root.children[0] task = root.children[0]
if not task.function_call_id == task_id: if not task.function_call_id == task_id:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
logger.info(task) logger.info(task)
logger.info(f"Task function call status: {task.status}") logger.info(f"Task function call status: {task.status}")
match task.status: match task.status:
@@ -593,7 +592,7 @@ async def subtitle_apply(body: FFMPEGSubtitleOverlayRequest,
sentry_trace = None sentry_trace = None
if headers.x_trace_id and headers.x_baggage: if headers.x_trace_id and headers.x_baggage:
sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage) sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
fn_call = fn.spawn(media=body.media, subtitle=body.subtitle, sentry_trace=sentry_trace) fn_call = fn.spawn(media=body.media, subtitle=body.subtitle, fonts=body.fonts, sentry_trace=sentry_trace)
return ModalTaskResponse(success=True, taskId=fn_call.object_id) return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@@ -629,7 +628,7 @@ async def bgm_nosie_reduce(body: FFMPEGMixBgmWithNoiseReduceRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id) return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/subtitle-apply/{task_id}", tags=["查询任务"], @web_app.get("/ffmpeg/bgm-nosie-reduce/{task_id}", tags=["查询任务"],
summary="查询混合BGM并降噪任务状态", description="查询混合BGM并降噪任务状态", summary="查询混合BGM并降噪任务状态", description="查询混合BGM并降噪任务状态",
responses={ responses={
status.HTTP_200_OK: { status.HTTP_200_OK: {

1
src/BowongModalFunctions/config.py
View File

@@ -6,6 +6,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class WorkerConfig(BaseSettings): class WorkerConfig(BaseSettings):
video_downloader_concurrency: int = Field(default=10, description="处理缓存任务的并行数") video_downloader_concurrency: int = Field(default=10, description="处理缓存任务的并行数")
ffmpeg_worker_concurrency: int = Field(default=10, description="处理视频合成任务的并行数") ffmpeg_worker_concurrency: int = Field(default=10, description="处理视频合成任务的并行数")
local_output_mount: str = Field(default="/mnt/outputs", description="本地结果挂载路径前缀")
S3_region: str = Field(default='ap-northeast-2', description="S3挂载桶的地域") S3_region: str = Field(default='ap-northeast-2', description="S3挂载桶的地域")
S3_bucket_name: str = Field(default='modal-media-cache', description="集群挂载的S3存储桶") S3_bucket_name: str = Field(default='modal-media-cache', description="集群挂载的S3存储桶")

8
src/BowongModalFunctions/models/media_model.py
View File

@@ -7,10 +7,12 @@ from urllib.parse import urlparse
from pydantic import (BaseModel, Field, field_validator, ValidationError, from pydantic import (BaseModel, Field, field_validator, ValidationError,
field_serializer, SerializationInfo, computed_field) field_serializer, SerializationInfo, computed_field)
from pydantic.json_schema import JsonSchemaValue from pydantic.json_schema import JsonSchemaValue
from ..config import WorkerConfig
# todo: 临时解决方案，需要想个办法获取本地挂载点配置 config = WorkerConfig()
s3_region = 'ap-northeast-2'
s3_bucket_name = 'modal-media-cache' s3_region = config.S3_region
s3_bucket_name = config.S3_bucket_name
class MediaProtocol(str, Enum): class MediaProtocol(str, Enum):

18
src/BowongModalFunctions/models/web_model.py
View File

@@ -55,6 +55,7 @@ class WebhookNotify(BaseModel):
examples=["Bearer 123456"]) examples=["Bearer 123456"])
class BaseFFMPEGTaskRequest(BaseModel): class BaseFFMPEGTaskRequest(BaseModel):
webhook: Optional[WebhookNotify] = Field(description="Task webhook", default=None) webhook: Optional[WebhookNotify] = Field(description="Task webhook", default=None)
@@ -64,6 +65,7 @@ class BaseFFMPEGTaskStatusResponse(BaseModel):
status: TaskStatus = Field(description="任务运行状态") status: TaskStatus = Field(description="任务运行状态")
error: Optional[str] = Field(description="任务错误原因", default=None) error: Optional[str] = Field(description="任务错误原因", default=None)
code: Optional[int] = Field(description="任务错误原因代码", default=None) code: Optional[int] = Field(description="任务错误原因代码", default=None)
results: Optional[List[str]] = Field(description="任务运行结果", default=None)
model_config = ConfigDict(extra='ignore') model_config = ConfigDict(extra='ignore')
@@ -217,6 +219,7 @@ class FFMPEGOverlayGifTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest): class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
media: MediaSource = Field(description="需要处理的媒体源") media: MediaSource = Field(description="需要处理的媒体源")
subtitle: MediaSource = Field(description="需要叠加的字幕文件") subtitle: MediaSource = Field(description="需要叠加的字幕文件")
fonts: List[MediaSource] = Field(description="字幕文件内使用到的字体文件")
@field_validator('media', mode='before') @field_validator('media', mode='before')
@classmethod @classmethod
@@ -241,6 +244,21 @@ class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
else: else:
raise TypeError(v) raise TypeError(v)
@field_validator('fonts', mode='before')
@classmethod
def parse_inputs(cls, v: Union[str, MediaSource]) -> List[MediaSource]:
if not v:
return v
result = []
for item in v:
if isinstance(item, str):
result.append(MediaSource.from_str(item))
elif isinstance(item, MediaSource):
result.append(item)
else:
raise pydantic.ValidationError("fonts元素类型错误: 必须是字符串")
return result
class FFMPEGSubtitleTaskStatusResponse(BaseFFMPEGTaskStatusResponse): class FFMPEGSubtitleTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
result: Optional[str] = Field(default=None, description="生成结果的URN") result: Optional[str] = Field(default=None, description="生成结果的URN")

37
src/BowongModalFunctions/utils/SentryUtils.py
View File

@@ -1,6 +1,11 @@
from typing import Optional, Tuple, Any from typing import Optional, Tuple, Any
import httpx
import psutil import psutil
import sentry_sdk import sentry_sdk
from loguru import logger
from BowongModalFunctions.models.web_model import WebhookNotify, WebhookMethodEnum, BaseFFMPEGTaskStatusResponse, \
TaskStatus, ErrorCode
class SentryUtils: class SentryUtils:
@@ -38,3 +43,35 @@ class SentryUtils:
return decorator return decorator
@staticmethod
def webhook_handler(webhook: WebhookNotify, func_id: str):
def decorator(func):
def wrapper(*args, **kwargs):
try:
result = func(*args, **kwargs)
status = TaskStatus.success
error = None
code = ErrorCode.SUCCESS.value
except Exception as e:
logger.exception(e)
result = None
status = TaskStatus.failed
error = e.message if hasattr(e, 'message') else str(e)
code = ErrorCode.SYSTEM_ERROR.value
with httpx.Client() as client:
match webhook.method:
case WebhookMethodEnum.POST:
response = client.post(url=webhook.endpoint.__str__(),
json=BaseFFMPEGTaskStatusResponse(
taskId=func_id, status=status, error=error,
code=code,
results=[result] if isinstance(result, str) else result,
).model_dump())
case WebhookMethodEnum.GET:
response = client.post(url=webhook.endpoint.__str__())
response.raise_for_status()
return result
return wrapper
return decorator

86
src/cluster/ffmpeg_app.py
View File

@@ -19,6 +19,7 @@ with ffmpeg_worker_image.imports():
import shutil, psutil, os, backoff, json, sentry_sdk import shutil, psutil, os, backoff, json, sentry_sdk
from typing import List, Optional, Tuple, Dict, Any from typing import List, Optional, Tuple, Dict, Any
from loguru import logger from loguru import logger
import httpx
from modal import current_function_call_id from modal import current_function_call_id
from ffmpeg.asyncio import FFmpeg from ffmpeg.asyncio import FFmpeg
from BowongModalFunctions.utils.SentryUtils import SentryUtils from BowongModalFunctions.utils.SentryUtils import SentryUtils
@@ -26,13 +27,14 @@ with ffmpeg_worker_image.imports():
from BowongModalFunctions.utils.VideoUtils import VideoUtils from BowongModalFunctions.utils.VideoUtils import VideoUtils
from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment
from BowongModalFunctions.models.media_model import MediaSources, MediaSource, MediaProtocol from BowongModalFunctions.models.media_model import MediaSources, MediaSource, MediaProtocol
from BowongModalFunctions.models.web_model import SentryTransactionInfo from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, WebhookMethodEnum, \
FFMPEGConcatTaskStatusResponse, TaskStatus, ErrorCode
from BowongModalFunctions.config import WorkerConfig from BowongModalFunctions.config import WorkerConfig
config = WorkerConfig() config = WorkerConfig()
s3_mount = config.S3_mount_dir s3_mount = config.S3_mount_dir
output_path_prefix = "/mnt/outputs" output_path_prefix = config.local_output_mount
sentry_sdk.init( sentry_sdk.init(
dsn="https://75cca970bfcc3d45d24361e7c0f1833c@sentry.bowongai.com/4", dsn="https://75cca970bfcc3d45d24361e7c0f1833c@sentry.bowongai.com/4",
@@ -111,7 +113,7 @@ with ffmpeg_worker_image.imports():
# region='ap-northeast', # region='ap-northeast',
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -119,7 +121,8 @@ with ffmpeg_worker_image.imports():
}, ) }, )
@modal.concurrent(max_inputs=1) @modal.concurrent(max_inputs=1)
async def ffmpeg_concat_medias(medias: MediaSources, async def ffmpeg_concat_medias(medias: MediaSources,
sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[ sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None) -> Tuple[
str, Optional[SentryTransactionInfo]]: str, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id() fn_id = current_function_call_id()
@@ -127,15 +130,17 @@ with ffmpeg_worker_image.imports():
@SentryUtils.sentry_tracker(name="视频合并任务", op="ffmpeg.concat", fn_id=fn_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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage 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: 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] 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, local_output_path = await VideoUtils.ffmpeg_concat_medias(media_paths=input_videos,
output_path=output_filepath) output_path=output_filepath)
return local_output_path return local_output_path
output_path = f"{output_path_prefix}/concat/outputs/{fn_id}/output.mp4" output_path = f"{output_path_prefix}/{config.modal_environment}/concat/outputs/{fn_id}/output.mp4"
local_output = await ffmpeg_process(media_sources=medias, output_filepath=output_path) local_output = await ffmpeg_process(media_sources=medias, output_filepath=output_path)
s3_outputs = local_copy_to_s3([local_output]) s3_outputs = local_copy_to_s3([local_output])
if not sentry_trace: if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(), sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage()) x_baggage=sentry_sdk.get_baggage())
@@ -147,7 +152,7 @@ with ffmpeg_worker_image.imports():
cloud="aws", cloud="aws",
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -157,14 +162,15 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1) @modal.concurrent(max_inputs=1)
async def ffmpeg_slice_media(media: MediaSource, async def ffmpeg_slice_media(media: MediaSource,
markers: List[FFMpegSliceSegment], markers: List[FFMpegSliceSegment],
webhook: Optional[str], # todo : handle webhook callback sentry_trace: Optional[SentryTransactionInfo] = None,
sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[ webhook: Optional[WebhookNotify] = None) -> Tuple[
List[str], Optional[SentryTransactionInfo]]: List[str], Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id() fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频切割任务", op="ffmpeg.slice", fn_id=fn_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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage 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], async def ffmpeg_slice_process(media_source: MediaSource, media_markers: List[FFMpegSliceSegment],
fn_id: str) -> List[str]: fn_id: str) -> List[str]:
cache_filepath = f"{s3_mount}/{media_source.cache_filepath}" cache_filepath = f"{s3_mount}/{media_source.cache_filepath}"
@@ -268,7 +274,7 @@ with ffmpeg_worker_image.imports():
@app.function(timeout=600, cloud="aws", @app.function(timeout=600, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -277,8 +283,8 @@ with ffmpeg_worker_image.imports():
) )
@modal.concurrent(max_inputs=1) @modal.concurrent(max_inputs=1)
async def ffmpeg_extract_audio(media_source: MediaSource, async def ffmpeg_extract_audio(media_source: MediaSource,
webhook: Optional[str], # todo handle webhook callback sentry_trace: Optional[SentryTransactionInfo] = None,
sentry_trace: Optional[SentryTransactionInfo] = None) -> Tuple[ webhook: Optional[WebhookNotify] = None) -> Tuple[
Optional[str], Optional[SentryTransactionInfo]]: Optional[str], Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id() fn_id = current_function_call_id()
@@ -286,9 +292,10 @@ with ffmpeg_worker_image.imports():
@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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage 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: async def ffmpeg_process(media: MediaSource, fn_id: str) -> str:
cache_filepath = f"{s3_mount}/{media.cache_filepath}" cache_filepath = f"{s3_mount}/{media.cache_filepath}"
output_path = f"{output_path_prefix}/extract_audio/outputs/{fn_id}/output.wav" 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) await VideoUtils.ffmpeg_extract_audio_async(cache_filepath, output_path)
s3_outputs = local_copy_to_s3([output_path]) s3_outputs = local_copy_to_s3([output_path])
return s3_outputs[0] return s3_outputs[0]
@@ -304,7 +311,7 @@ with ffmpeg_worker_image.imports():
@app.function(timeout=600, cloud="aws", @app.function(timeout=600, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -315,17 +322,19 @@ with ffmpeg_worker_image.imports():
mirror_scale_down_size: int = 6, mirror_scale_down_size: int = 6,
mirror_from_right: bool = True, mirror_from_right: bool = True,
mirror_position: tuple[float, float] = (40, 40), mirror_position: tuple[float, float] = (40, 40),
sentry_trace: Optional[SentryTransactionInfo] = None): sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
fn_id = current_function_call_id() fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频镜像去重", op="ffmpeg.mirror", fn_id=fn_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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage 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, async def ffmpeg_process(media: MediaSource, func_id: str, mirror_scale_down_size: int = 6,
mirror_from_right: bool = True, mirror_from_right: bool = True,
mirror_position: tuple[float, float] = (40, 40)) -> str: mirror_position: tuple[float, float] = (40, 40)) -> str:
media_filepath = f"{s3_mount}/{media.cache_filepath}" media_filepath = f"{s3_mount}/{media.cache_filepath}"
output_path = f"{output_path_prefix}/corner_mirror/outputs/{func_id}/output.mp4" 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, local_output_filepath = await VideoUtils.ffmpeg_corner_mirror(media_path=media_filepath,
output_path=output_path, output_path=output_path,
mirror_from_right=mirror_from_right, mirror_from_right=mirror_from_right,
@@ -346,7 +355,7 @@ with ffmpeg_worker_image.imports():
@app.function(timeout=600, cloud="aws", @app.function(timeout=600, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -354,16 +363,18 @@ with ffmpeg_worker_image.imports():
}, ) }, )
@modal.concurrent(max_inputs=1) @modal.concurrent(max_inputs=1)
async def ffmpeg_overlay_gif(media: MediaSource, gif: MediaSource, async def ffmpeg_overlay_gif(media: MediaSource, gif: MediaSource,
sentry_trace: Optional[SentryTransactionInfo] = None): sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
fn_id = current_function_call_id() fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频叠加GIF特效", op="ffmpeg.overlay", fn_id=fn_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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage 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: async def ffmpeg_process(media: MediaSource, func_id: str, gif: MediaSource) -> str:
media_filepath = f"{s3_mount}/{media.cache_filepath}" media_filepath = f"{s3_mount}/{media.cache_filepath}"
gif_filepath = f"{s3_mount}/{gif.cache_filepath}" gif_filepath = f"{s3_mount}/{gif.cache_filepath}"
output_path = f"{output_path_prefix}/overlay/outputs/{func_id}/output.mp4" 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, local_output_filepath = await VideoUtils.ffmpeg_overlay_gif(media_path=media_filepath,
output_path=output_path, output_path=output_path,
overlay_gif_path=gif_filepath) overlay_gif_path=gif_filepath)
@@ -380,7 +391,7 @@ with ffmpeg_worker_image.imports():
@app.function(timeout=600, cloud="aws", @app.function(timeout=600, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -388,15 +399,17 @@ with ffmpeg_worker_image.imports():
}, ) }, )
@modal.concurrent(max_inputs=1) @modal.concurrent(max_inputs=1)
async def ffmpeg_zoom_loop(media: MediaSource, duration: int = 6, zoom: float = 0.1, async def ffmpeg_zoom_loop(media: MediaSource, duration: int = 6, zoom: float = 0.1,
sentry_trace: Optional[SentryTransactionInfo] = None): sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
fn_id = current_function_call_id() fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频放大缩小去重", op="ffmpeg.zoom.loop", fn_id=fn_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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage 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: 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}" media_filepath = f"{s3_mount}/{media.cache_filepath}"
output_path = f"{output_path_prefix}/zoom_loop/outputs/{func_id}/output.mp4" 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, local_output_filepath = await VideoUtils.ffmpeg_zoom_loop(media_path=media_filepath,
output_path=output_path, output_path=output_path,
duration=duration, duration=duration,
@@ -414,7 +427,7 @@ with ffmpeg_worker_image.imports():
@app.function(timeout=600, cloud="aws", @app.function(timeout=600, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -423,18 +436,20 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1) @modal.concurrent(max_inputs=1)
async def ffmpeg_bgm_nosie_reduce(media: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None, 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, video_volume: float = 1.4, music_volume: float = 0.1,
sentry_trace: Optional[SentryTransactionInfo] = None): sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
fn_id = current_function_call_id() fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频混合BGM降噪", op="ffmpeg.bgm.nosie.reduce", fn_id=fn_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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage 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, async def ffmpeg_process(video: MediaSource, bgm: MediaSource, func_id: str,
noise_sample: Optional[MediaSource] = None) -> str: noise_sample: Optional[MediaSource] = None) -> str:
local_input_filepath = f"{s3_mount}/{video.cache_filepath}" local_input_filepath = f"{s3_mount}/{video.cache_filepath}"
bgm_filepath = f"{s3_mount}/{bgm.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 noise_sample_path = f"{s3_mount}/{noise_sample.cache_filepath}" if noise_sample else None
output_path = f"{output_path_prefix}/bgm_nosie_reduce/outputs/{func_id}/output.mp4" 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, local_output_filepath = await VideoUtils.ffmpeg_mix_bgm_with_noise_reduce(media_path=local_input_filepath,
bgm_audio_path=bgm_filepath, bgm_audio_path=bgm_filepath,
video_volume=video_volume, video_volume=video_volume,
@@ -457,7 +472,7 @@ with ffmpeg_worker_image.imports():
@app.function(timeout=600, cloud="aws", @app.function(timeout=600, cloud="aws",
max_containers=config.ffmpeg_worker_concurrency, max_containers=config.ffmpeg_worker_concurrency,
volumes={ volumes={
f"/mntS3": modal.CloudBucketMount( s3_mount: modal.CloudBucketMount(
bucket_name=config.S3_bucket_name, bucket_name=config.S3_bucket_name,
secret=modal.Secret.from_name("aws-s3-secret", secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment), environment_name=config.modal_environment),
@@ -465,26 +480,31 @@ with ffmpeg_worker_image.imports():
}, ) }, )
@modal.concurrent(max_inputs=1) @modal.concurrent(max_inputs=1)
async def ffmpeg_subtitle_apply(media: MediaSource, subtitle: MediaSource, async def ffmpeg_subtitle_apply(media: MediaSource, subtitle: MediaSource,
sentry_trace: Optional[SentryTransactionInfo] = None): fonts: List[MediaSource], sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
fn_id = current_function_call_id() fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频叠加字幕", op="ffmpeg.subtitle.apply", fn_id=fn_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_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
sentry_baggage=sentry_trace.x_baggage if sentry_trace else None) sentry_baggage=sentry_trace.x_baggage if sentry_trace else None)
async def ffmpeg_process(video: MediaSource, subtitle: MediaSource, func_id: str, font_dir: str) -> str: @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}" media_path = f"{s3_mount}/{video.cache_filepath}"
subtitle_path = f"{s3_mount}/{subtitle.cache_filepath}" subtitle_path = f"{s3_mount}/{subtitle.cache_filepath}"
output_path = f"{output_path_prefix}/subtitle_apply/{func_id}/output.mp4" 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, local_output = await VideoUtils.ffmpeg_subtitle_apply(media_path=media_path, subtitle_path=subtitle_path,
font_dir=font_dir, output_path=output_path) font_dir=font_dir, output_path=output_path)
s3_outputs = local_copy_to_s3([local_output]) s3_outputs = local_copy_to_s3([local_output])
return s3_outputs[0] return s3_outputs[0]
# todo : check if subtitle use fonts exist in fonts dir result = await ffmpeg_process(video=media, subtitle=subtitle, fonts=fonts, func_id=fn_id)
font_dir = f"{s3_mount}/fonts"
result = await ffmpeg_process(video=media, subtitle=subtitle, func_id=fn_id, font_dir=font_dir)
if not sentry_trace: if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(), sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage()) x_baggage=sentry_sdk.get_baggage())

9
src/cluster/video.py
View File

@@ -38,15 +38,6 @@ with downloader_image.imports():
from BowongModalFunctions.models.web_model import SentryTransactionInfo from BowongModalFunctions.models.web_model import SentryTransactionInfo
config = WorkerConfig() config = WorkerConfig()
# config = WorkerConfig(
# modal_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",
# modal_environment="dev"
# )
sentry_sdk.init(dsn="https://85632fdcd62f699c2f88af6ca489e9ec@sentry.bowongai.com/3", sentry_sdk.init(dsn="https://85632fdcd62f699c2f88af6ca489e9ec@sentry.bowongai.com/3",
send_default_pii=True, send_default_pii=True,