From 281dff79661af8e59310f2382a987833943a241c Mon Sep 17 00:00:00 2001
From: "shuohigh@gmail.com" <VStC*5Szb8@qQD>
Date: Fri, 16 May 2025 11:21:02 +0800
Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0webhook=E5=9B=9E=E8=B0=83?=
 =?UTF-8?q?=E5=A4=84=E7=90=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 README.md                                     |  2 +-
 .../modal deploy.run.xml                      |  2 +-
 src/BowongModalFunctions/api.py               |  5 +-
 src/BowongModalFunctions/config.py            |  1 +
 .../models/media_model.py                     |  8 +-
 src/BowongModalFunctions/models/web_model.py  | 18 ++++
 src/BowongModalFunctions/utils/SentryUtils.py | 37 ++++++++
 src/cluster/ffmpeg_app.py                     | 86 ++++++++++++-------
 src/cluster/video.py                          |  9 --
 9 files changed, 118 insertions(+), 50 deletions(-)
 rename .idea/runConfigurations/modal_deploy.xml => cmd/modal deploy.run.xml (93%)

diff --git a/README.md b/README.md
index 90e973e..c6c77b1 100644
--- a/README.md
+++ b/README.md
@@ -3,7 +3,7 @@
 ## 部署
 
 ```bash
-modal deploy -e dev ./src/cluster/app.py
+modal deploy -m cluster.app 
 ```
 
 # Pip
diff --git a/.idea/runConfigurations/modal_deploy.xml b/cmd/modal deploy.run.xml
similarity index 93%
rename from .idea/runConfigurations/modal_deploy.xml
rename to cmd/modal deploy.run.xml
index 120d67e..3f70d48 100644
--- a/.idea/runConfigurations/modal_deploy.xml
+++ b/cmd/modal deploy.run.xml	
@@ -1,7 +1,7 @@
 <component name="ProjectRunConfigurationManager">
   <configuration default="false" name="modal deploy" type="PythonConfigurationType" factoryName="Python">
     <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="PARENT_ENVS" value="true" />
     <envs>
diff --git a/src/BowongModalFunctions/api.py b/src/BowongModalFunctions/api.py
index 631645e..8cd8ddd 100644
--- a/src/BowongModalFunctions/api.py
+++ b/src/BowongModalFunctions/api.py
@@ -180,7 +180,6 @@ async def get_modal_task_status(task_id: str) -> Tuple[
         task = root.children[0]
         if not task.function_call_id == task_id:
             return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
-
         logger.info(task)
         logger.info(f"Task function call status: {task.status}")
         match task.status:
@@ -593,7 +592,7 @@ async def subtitle_apply(body: FFMPEGSubtitleOverlayRequest,
     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.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)
 
 
@@ -629,7 +628,7 @@ async def bgm_nosie_reduce(body: FFMPEGMixBgmWithNoiseReduceRequest,
     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并降噪任务状态",
              responses={
                  status.HTTP_200_OK: {
diff --git a/src/BowongModalFunctions/config.py b/src/BowongModalFunctions/config.py
index 9bff9c7..32383e9 100644
--- a/src/BowongModalFunctions/config.py
+++ b/src/BowongModalFunctions/config.py
@@ -6,6 +6,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 class WorkerConfig(BaseSettings):
     video_downloader_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_bucket_name: str = Field(default='modal-media-cache', description="集群挂载的S3存储桶")
diff --git a/src/BowongModalFunctions/models/media_model.py b/src/BowongModalFunctions/models/media_model.py
index 0f562b6..034be3c 100644
--- a/src/BowongModalFunctions/models/media_model.py
+++ b/src/BowongModalFunctions/models/media_model.py
@@ -7,10 +7,12 @@ from urllib.parse import urlparse
 from pydantic import (BaseModel, Field, field_validator, ValidationError,
                       field_serializer, SerializationInfo, computed_field)
 from pydantic.json_schema import JsonSchemaValue
+from ..config import WorkerConfig
 
-# todo: 临时解决方案，需要想个办法获取本地挂载点配置
-s3_region = 'ap-northeast-2'
-s3_bucket_name = 'modal-media-cache'
+config = WorkerConfig()
+
+s3_region = config.S3_region
+s3_bucket_name = config.S3_bucket_name
 
 
 class MediaProtocol(str, Enum):
diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py
index d1eaa90..661455b 100644
--- a/src/BowongModalFunctions/models/web_model.py
+++ b/src/BowongModalFunctions/models/web_model.py
@@ -55,6 +55,7 @@ class WebhookNotify(BaseModel):
                                           examples=["Bearer 123456"])
 
 
+
 class BaseFFMPEGTaskRequest(BaseModel):
     webhook: Optional[WebhookNotify] = Field(description="Task webhook", default=None)
 
@@ -64,6 +65,7 @@ class BaseFFMPEGTaskStatusResponse(BaseModel):
     status: TaskStatus = Field(description="任务运行状态")
     error: Optional[str] = Field(description="任务错误原因", default=None)
     code: Optional[int] = Field(description="任务错误原因代码", default=None)
+    results: Optional[List[str]] = Field(description="任务运行结果", default=None)
 
     model_config = ConfigDict(extra='ignore')
 
@@ -217,6 +219,7 @@ class FFMPEGOverlayGifTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
 class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
     media: MediaSource = Field(description="需要处理的媒体源")
     subtitle: MediaSource = Field(description="需要叠加的字幕文件")
+    fonts: List[MediaSource] = Field(description="字幕文件内使用到的字体文件")
 
     @field_validator('media', mode='before')
     @classmethod
@@ -241,6 +244,21 @@ class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
         else:
             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):
     result: Optional[str] = Field(default=None, description="生成结果的URN")
diff --git a/src/BowongModalFunctions/utils/SentryUtils.py b/src/BowongModalFunctions/utils/SentryUtils.py
index 5256de3..ac9962d 100644
--- a/src/BowongModalFunctions/utils/SentryUtils.py
+++ b/src/BowongModalFunctions/utils/SentryUtils.py
@@ -1,6 +1,11 @@
 from typing import Optional, Tuple, Any
+import httpx
 import psutil
 import sentry_sdk
+from loguru import logger
+
+from BowongModalFunctions.models.web_model import WebhookNotify, WebhookMethodEnum, BaseFFMPEGTaskStatusResponse, \
+    TaskStatus, ErrorCode
 
 
 class SentryUtils:
@@ -38,3 +43,35 @@ class SentryUtils:
 
         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
diff --git a/src/cluster/ffmpeg_app.py b/src/cluster/ffmpeg_app.py
index d4a473e..f93d0fb 100644
--- a/src/cluster/ffmpeg_app.py
+++ b/src/cluster/ffmpeg_app.py
@@ -19,6 +19,7 @@ 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
+    import httpx
     from modal import current_function_call_id
     from ffmpeg.asyncio import FFmpeg
     from BowongModalFunctions.utils.SentryUtils import SentryUtils
@@ -26,13 +27,14 @@ with ffmpeg_worker_image.imports():
     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.models.web_model import SentryTransactionInfo, WebhookNotify, WebhookMethodEnum, \
+        FFMPEGConcatTaskStatusResponse, TaskStatus, ErrorCode
     from BowongModalFunctions.config import WorkerConfig
 
     config = WorkerConfig()
 
     s3_mount = config.S3_mount_dir
-    output_path_prefix = "/mnt/outputs"
+    output_path_prefix = config.local_output_mount
 
     sentry_sdk.init(
         dsn="https://75cca970bfcc3d45d24361e7c0f1833c@sentry.bowongai.com/4",
@@ -111,7 +113,7 @@ with ffmpeg_worker_image.imports():
         # region='ap-northeast',
         max_containers=config.ffmpeg_worker_concurrency,
         volumes={
-            f"/mntS3": modal.CloudBucketMount(
+            s3_mount: modal.CloudBucketMount(
                 bucket_name=config.S3_bucket_name,
                 secret=modal.Secret.from_name("aws-s3-secret",
                                               environment_name=config.modal_environment),
@@ -119,7 +121,8 @@ with ffmpeg_worker_image.imports():
         }, )
     @modal.concurrent(max_inputs=1)
     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]]:
 
         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,
                                     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)
             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)
         s3_outputs = local_copy_to_s3([local_output])
+
         if not sentry_trace:
             sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
                                                  x_baggage=sentry_sdk.get_baggage())
@@ -147,7 +152,7 @@ with ffmpeg_worker_image.imports():
         cloud="aws",
         max_containers=config.ffmpeg_worker_concurrency,
         volumes={
-            f"/mntS3": modal.CloudBucketMount(
+            s3_mount: modal.CloudBucketMount(
                 bucket_name=config.S3_bucket_name,
                 secret=modal.Secret.from_name("aws-s3-secret",
                                               environment_name=config.modal_environment),
@@ -157,14 +162,15 @@ with ffmpeg_worker_image.imports():
     @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[
+                                 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}"
@@ -268,7 +274,7 @@ with ffmpeg_worker_image.imports():
     @app.function(timeout=600, cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
-                      f"/mntS3": modal.CloudBucketMount(
+                      s3_mount: modal.CloudBucketMount(
                           bucket_name=config.S3_bucket_name,
                           secret=modal.Secret.from_name("aws-s3-secret",
                                                         environment_name=config.modal_environment),
@@ -277,8 +283,8 @@ with ffmpeg_worker_image.imports():
                   )
     @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[
+                                   sentry_trace: Optional[SentryTransactionInfo] = None,
+                                   webhook: Optional[WebhookNotify] = None) -> Tuple[
         Optional[str], Optional[SentryTransactionInfo]]:
 
         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,
                                     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}/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)
             s3_outputs = local_copy_to_s3([output_path])
             return s3_outputs[0]
@@ -304,7 +311,7 @@ with ffmpeg_worker_image.imports():
     @app.function(timeout=600, cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
-                      f"/mntS3": modal.CloudBucketMount(
+                      s3_mount: modal.CloudBucketMount(
                           bucket_name=config.S3_bucket_name,
                           secret=modal.Secret.from_name("aws-s3-secret",
                                                         environment_name=config.modal_environment),
@@ -315,17 +322,19 @@ with ffmpeg_worker_image.imports():
                                    mirror_scale_down_size: int = 6,
                                    mirror_from_right: bool = True,
                                    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()
 
         @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}/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,
                                                                           output_path=output_path,
                                                                           mirror_from_right=mirror_from_right,
@@ -346,7 +355,7 @@ with ffmpeg_worker_image.imports():
     @app.function(timeout=600, cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
-                      f"/mntS3": modal.CloudBucketMount(
+                      s3_mount: modal.CloudBucketMount(
                           bucket_name=config.S3_bucket_name,
                           secret=modal.Secret.from_name("aws-s3-secret",
                                                         environment_name=config.modal_environment),
@@ -354,16 +363,18 @@ with ffmpeg_worker_image.imports():
                   }, )
     @modal.concurrent(max_inputs=1)
     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()
 
         @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}/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,
                                                                         output_path=output_path,
                                                                         overlay_gif_path=gif_filepath)
@@ -380,7 +391,7 @@ with ffmpeg_worker_image.imports():
     @app.function(timeout=600, cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
-                      f"/mntS3": modal.CloudBucketMount(
+                      s3_mount: modal.CloudBucketMount(
                           bucket_name=config.S3_bucket_name,
                           secret=modal.Secret.from_name("aws-s3-secret",
                                                         environment_name=config.modal_environment),
@@ -388,15 +399,17 @@ with ffmpeg_worker_image.imports():
                   }, )
     @modal.concurrent(max_inputs=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()
 
         @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}/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,
                                                                       output_path=output_path,
                                                                       duration=duration,
@@ -414,7 +427,7 @@ with ffmpeg_worker_image.imports():
     @app.function(timeout=600, cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
-                      f"/mntS3": modal.CloudBucketMount(
+                      s3_mount: modal.CloudBucketMount(
                           bucket_name=config.S3_bucket_name,
                           secret=modal.Secret.from_name("aws-s3-secret",
                                                         environment_name=config.modal_environment),
@@ -423,18 +436,20 @@ with ffmpeg_worker_image.imports():
     @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):
+                                      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}/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,
                                                                                       bgm_audio_path=bgm_filepath,
                                                                                       video_volume=video_volume,
@@ -457,7 +472,7 @@ with ffmpeg_worker_image.imports():
     @app.function(timeout=600, cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
-                      f"/mntS3": modal.CloudBucketMount(
+                      s3_mount: modal.CloudBucketMount(
                           bucket_name=config.S3_bucket_name,
                           secret=modal.Secret.from_name("aws-s3-secret",
                                                         environment_name=config.modal_environment),
@@ -465,26 +480,31 @@ with ffmpeg_worker_image.imports():
                   }, )
     @modal.concurrent(max_inputs=1)
     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()
 
         @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)
-        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}"
             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,
                                                                   font_dir=font_dir, output_path=output_path)
-
             s3_outputs = local_copy_to_s3([local_output])
             return s3_outputs[0]
 
-        # todo : check if subtitle use fonts exist in fonts dir
-        font_dir = f"{s3_mount}/fonts"
-        result = await ffmpeg_process(video=media, subtitle=subtitle, func_id=fn_id, font_dir=font_dir)
+        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())
diff --git a/src/cluster/video.py b/src/cluster/video.py
index 65f272a..8fa12b6 100644
--- a/src/cluster/video.py
+++ b/src/cluster/video.py
@@ -38,15 +38,6 @@ with downloader_image.imports():
     from BowongModalFunctions.models.web_model import SentryTransactionInfo
 
     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",
                     send_default_pii=True,