新增API

- /ffmpeg/subtitle-apply
- /ffmpeg/bgm-nosie-reduce
- /ffmpeg/zoom-loop
- /ffmpeg/overlay-gif

.gitignore

@@ -1 +1,2 @@
-*/**/.pypirc
+.pypirc
+.env

.idea/runConfigurations/modal_deploy.xml

@@ -14,7 +14,7 @@
     <option name="ADD_SOURCE_ROOTS" value="true" />
     <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
     <option name="SCRIPT_NAME" value="modal" />
-    <option name="PARAMETERS" value="deploy ./src/cluster/app.py" />
+    <option name="PARAMETERS" value="deploy -m cluster.app" />
     <option name="SHOW_COMMAND_LINE" value="false" />
     <option name="EMULATE_TERMINAL" value="false" />
     <option name="MODULE_MODE" value="true" />

README.md

@@ -28,5 +28,5 @@ py -m twine upload --config-file=.pypirc --repository=coding-pypi dist/*
 ## 安装
 
 ```bash
- uv add bowong-modal-functions --index-url=https://<用户名>:<密钥>@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple
+ uv pip install bowong-modal-functions --index-url=https://<用户名>:<密钥>@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple
 ```

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "BowongModalFunctions"
-version = "0.1.1-pre"
+dynamic = ["version"]
 authors = [
     { name = "Yudi Xiao", email = "xyd@bowong.ai" },
 ]
@@ -27,7 +27,6 @@ dependencies = [
     "cos-python-sdk-v5>=1.9.36",
     "boto3>=1.37.37",
     "psutil>=7.0.0",
-    "twine>=6.1.0",
     "scalar-fastapi>=1.0.3",
     "modal>=0.76.3",
 ]
@@ -39,10 +38,16 @@ classifiers = [
 license = "MIT"
 license-files = ["LICEN[CS]E*"]
 
+[project.optional-dependencies]
+cli = ["twine>=6.1.0"]
+
 [build-system]
 requires = ["hatchling >= 1.26"]
 build-backend = "hatchling.build"
 
+[tool.hatch.version]
+path = "src/BowongModalFunctions/__init__.py"
+
 [tool.hatch.build.targets.wheel]
 packages = ["src/BowongModalFunctions"]
 sources = ["src"]

src/BowongModalFunctions/__init__.py

@@ -1 +1 @@
-__version__ = "0.1.1-pre"
+__version__ = "0.1.1-pre3"

src/BowongModalFunctions/api.py

@@ -30,10 +30,19 @@ from .models.web_model import (TaskStatus, ErrorCode,
                                FFMPEGConcatRequest,
                                FFMPEGExtractAudioRequest,
                                FFMPEGCornerMirrorRequest,
+                               FFMPEGOverlayGifRequest,
+                               FFMPEGZoomLoopRequest,
+                               FFMPEGSubtitleOverlayRequest,
+                               FFMPEGMixBgmWithNoiseReduceRequest,
                                FFMPEGSliceTaskStatusResponse,
                                FFMPEGConcatTaskStatusResponse,
                                FFMPEGExtractAudioTaskStatusResponse,
-                               FFMPEGCornerMirrorTaskStatusResponse)
+                               FFMPEGCornerMirrorTaskStatusResponse,
+                               FFMPEGOverlayGifTaskStatusResponse,
+                               FFMPEGSubtitleTaskStatusResponse,
+                               FFMPEGZoomLoopTaskStatusResponse,
+                               FFMPEGMixBgmWithNoiseReduceStatusResponse
+                               )
 from .config import WorkerConfig
 
 bearer_scheme = HTTPBearer()
@@ -499,7 +508,7 @@ async def corner_mirror(body: FFMPEGCornerMirrorRequest,
 @web_app.get("/ffmpeg/corner-mirror/{task_id}",
              tags=["查询任务"],
              summary="查询镜像小窗去重任务状态",
-             description="<UNK>",
+             description="查询镜像小窗去重任务状态",
              responses={
                  status.HTTP_200_OK: {
                      "description": "",
@@ -514,3 +523,124 @@ async def corner_mirror_status(task_id: str, response: Response) -> FFMPEGCorner
         response.headers["x-baggage"] = transaction.x_baggage
     return FFMPEGCornerMirrorTaskStatusResponse(taskId=task_id, status=task_status, code=code,
                                                 error=reason, result=result)
+
+
+@web_app.post("/ffmpeg/overlay-gif", tags=["发起任务"], summary="发起叠加gif特效任务",
+              description="发起叠加gif特效任务", dependencies=[Depends(verify_token)])
+async def overlay_gif(body: FFMPEGOverlayGifRequest,
+                      headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
+    fn = modal.Function.from_name(config.app_name, "ffmpeg_overlay_gif", environment_name=config.environment)
+    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, gif=body.gif, sentry_trace=sentry_trace)
+    return ModalTaskResponse(success=True, taskId=fn_call.object_id)
+
+
+@web_app.get("/ffmpeg/overlay-gif/{task_id}", tags=["查询任务"],
+             summary="查询叠加gif特效任务状态", description="查询叠加gif特效任务状态",
+             responses={
+                 status.HTTP_200_OK: {
+                     "description": "",
+                     "headers": sentry_header_schema
+                 },
+             },
+             dependencies=[Depends(verify_token)])
+async def overlay_gif_status(task_id: str, response: Response) -> FFMPEGOverlayGifTaskStatusResponse:
+    task_status, code, reason, result, transaction = await get_modal_task_status(task_id)
+    if transaction:
+        response.headers["x-trace-id"] = transaction.x_trace_id
+        response.headers["x-baggage"] = transaction.x_baggage
+    return FFMPEGOverlayGifTaskStatusResponse(taskId=task_id, status=task_status, code=code,
+                                              error=reason, result=result)
+
+
+@web_app.post("/ffmpeg/zoom-loop", tags=["发起任务"], summary="发起放大缩小循环去重任务",
+              description="发起放大缩小循环去重任务", dependencies=[Depends(verify_token)])
+async def zoom_loop(body: FFMPEGZoomLoopRequest,
+                    headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
+    fn = modal.Function.from_name(config.app_name, "ffmpeg_zoom_loop", environment_name=config.environment)
+    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, duration=body.duration, zoom=body.zoom, sentry_trace=sentry_trace)
+    return ModalTaskResponse(success=True, taskId=fn_call.object_id)
+
+
+@web_app.get("/ffmpeg/zoom-loop/{task_id}", tags=["查询任务"],
+             summary="查询放大缩小循环去重任务状态", description="查询放大缩小循环去重任务状态",
+             responses={
+                 status.HTTP_200_OK: {
+                     "description": "",
+                     "headers": sentry_header_schema
+                 },
+             },
+             dependencies=[Depends(verify_token)])
+async def zoom_loop_status(task_id: str, response: Response) -> FFMPEGZoomLoopTaskStatusResponse:
+    task_status, code, reason, result, transaction = await get_modal_task_status(task_id)
+    if transaction:
+        response.headers["x-trace-id"] = transaction.x_trace_id
+        response.headers["x-baggage"] = transaction.x_baggage
+    return FFMPEGZoomLoopTaskStatusResponse(taskId=task_id, status=task_status, code=code,
+                                            error=reason, result=result)
+
+
+@web_app.post("/ffmpeg/subtitle-apply", tags=["发起任务"], summary="发起渲染字幕任务",
+              description="发起渲染字幕任务", dependencies=[Depends(verify_token)])
+async def subtitle_apply(body: FFMPEGSubtitleOverlayRequest,
+                         headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
+    fn = modal.Function.from_name(config.app_name, "ffmpeg_subtitle_apply", environment_name=config.environment)
+    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)
+    return ModalTaskResponse(success=True, taskId=fn_call.object_id)
+
+
+@web_app.get("/ffmpeg/subtitle-apply/{task_id}", tags=["查询任务"],
+             summary="查询渲染字幕任务状态", description="查询渲染字幕任务状态",
+             responses={
+                 status.HTTP_200_OK: {
+                     "description": "",
+                     "headers": sentry_header_schema
+                 },
+             },
+             dependencies=[Depends(verify_token)])
+async def subtitle_apply_status(task_id: str, response: Response) -> FFMPEGSubtitleTaskStatusResponse:
+    task_status, code, reason, result, transaction = await get_modal_task_status(task_id)
+    if transaction:
+        response.headers["x-trace-id"] = transaction.x_trace_id
+        response.headers["x-baggage"] = transaction.x_baggage
+    return FFMPEGSubtitleTaskStatusResponse(taskId=task_id, status=task_status, code=code,
+                                            error=reason, result=result)
+
+
+@web_app.post("/ffmpeg/bgm-nosie-reduce", tags=["发起任务"], summary="发起混合BGM并降噪任务",
+              description="发起混合BGM并降噪任务", dependencies=[Depends(verify_token)])
+async def bgm_nosie_reduce(body: FFMPEGMixBgmWithNoiseReduceRequest,
+                           headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
+    fn = modal.Function.from_name(config.app_name, "ffmpeg_bgm_nosie_reduce", environment_name=config.environment)
+    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, bgm=body.bgm, video_volume=body.video_volume, music_volume=body.music_volume,
+                       noise_sample=body.noise_sample, sentry_trace=sentry_trace)
+    return ModalTaskResponse(success=True, taskId=fn_call.object_id)
+
+
+@web_app.get("/ffmpeg/subtitle-apply/{task_id}", tags=["查询任务"],
+             summary="查询混合BGM并降噪任务状态", description="查询混合BGM并降噪任务状态",
+             responses={
+                 status.HTTP_200_OK: {
+                     "description": "",
+                     "headers": sentry_header_schema
+                 },
+             },
+             dependencies=[Depends(verify_token)])
+async def bgm_nosie_reduce_status(task_id: str, response: Response) -> FFMPEGMixBgmWithNoiseReduceStatusResponse:
+    task_status, code, reason, result, transaction = await get_modal_task_status(task_id)
+    if transaction:
+        response.headers["x-trace-id"] = transaction.x_trace_id
+        response.headers["x-baggage"] = transaction.x_baggage
+    return FFMPEGMixBgmWithNoiseReduceStatusResponse(taskId=task_id, status=task_status, code=code,
+                                                     error=reason, result=result)

src/BowongModalFunctions/models/media_model.py

@@ -1,6 +1,7 @@
 import os
 from datetime import datetime
 from enum import Enum
+from functools import cached_property
 from typing import List, Union, Optional, Any, Dict
 from urllib.parse import urlparse
 from pydantic import (BaseModel, Field, field_validator, ValidationError,
@@ -116,6 +117,11 @@ class MediaSource(BaseModel):
             case _:
                 return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.path}"
 
+    @computed_field(description="文件后缀名")
+    @cached_property
+    def file_extension(self) -> Optional[str]:
+        return os.path.basename(self.urn).split('.')[-1]
+
     @field_serializer('expired_at')
     def serialize_datetime(self, value: Optional[datetime], info: SerializationInfo) -> Optional[str]:
         if value:

src/BowongModalFunctions/models/web_model.py

@@ -1,29 +1,12 @@
 from enum import Enum
 from typing import List, Union, Optional
+
+import pydantic
 from pydantic import BaseModel, Field, field_validator, ConfigDict, HttpUrl
 from .ffmpeg_worker_model import FFMpegSliceSegment
 from .media_model import MediaSource, MediaSources
 
 
-class SentryTransactionHeader(BaseModel):
-    x_trace_id: Optional[str] = Field(description="Sentry Transaction ID", default=None)
-    x_baggage: Optional[str] = Field(description="Sentry Transaction baggage", default=None)
-
-
-class SentryTransactionInfo(BaseModel):
-    x_trace_id: str = Field(description="Sentry Transaction ID")
-    x_baggage: str = Field(description="Sentry Transaction baggage")
-
-
-class FFMPEGSliceTaskStatusRequest(BaseModel):
-    taskId: str = Field(description="任务Id")
-
-
-class ModalTaskResponse(BaseModel):
-    success: bool = Field(description="任务接受成功")
-    taskId: str = Field(description="任务Id")
-
-
 class TaskStatus(str, Enum):
     running = "running"
     failed = "failed"
@@ -46,6 +29,25 @@ class WebhookMethodEnum(str, Enum):
     POST = "post"
 
 
+class SentryTransactionHeader(BaseModel):
+    x_trace_id: Optional[str] = Field(description="Sentry Transaction ID", default=None)
+    x_baggage: Optional[str] = Field(description="Sentry Transaction baggage", default=None)
+
+
+class SentryTransactionInfo(BaseModel):
+    x_trace_id: str = Field(description="Sentry Transaction ID")
+    x_baggage: str = Field(description="Sentry Transaction baggage")
+
+
+class FFMPEGSliceTaskStatusRequest(BaseModel):
+    taskId: str = Field(description="任务Id")
+
+
+class ModalTaskResponse(BaseModel):
+    success: bool = Field(description="任务接受成功")
+    taskId: str = Field(description="任务Id")
+
+
 class WebhookNotify(BaseModel):
     endpoint: HttpUrl = Field(description="Webhook回调端点", examples=["https://webhook.example.com"])
     method: WebhookMethodEnum = Field(description="Webhook回调请求方法", examples=["get", "post"])
@@ -112,7 +114,7 @@ class FFMPEGExtractAudioTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
 
 
 class FFMPEGCornerMirrorRequest(BaseFFMPEGTaskRequest):
-    media: MediaSource = Field(description="")
+    media: MediaSource = Field(description="需要处理的媒体源")
 
     @field_validator('media', mode='before')
     @classmethod
@@ -127,3 +129,163 @@ class FFMPEGCornerMirrorRequest(BaseFFMPEGTaskRequest):
 
 class FFMPEGCornerMirrorTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
     result: Optional[str] = Field(default=None, description="任务运行结果")
+
+
+class FFMPEGBgmRequest(BaseFFMPEGTaskRequest):
+    media: MediaSource = Field(description="需要处理的媒体源")
+    bgm_media: MediaSource = Field(description="添加的BGM媒体源", alias="bgmMedia")
+
+    @field_validator('media', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+    @field_validator('bgm_media', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            media = MediaSource.from_str(v)
+            if not media.file_extension in ['wav', 'mp3']:
+                raise pydantic.ValidationError("必须使用符合规范的音频格式, 如wav或者mp3")
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+
+class FFMPEGBgmTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
+    result: Optional[str] = Field(default=None, description="生成结果的URN")
+
+
+class FFMPEGZoomLoopRequest(BaseFFMPEGTaskRequest):
+    media: MediaSource = Field(description="需要处理的媒体源")
+    duration: float = Field(description="放大缩小一个循环的持续时间秒数", default=6.0)
+    zoom: float = Field(description="放大缩小系数", default=0.1)
+
+    @field_validator('media', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+
+class FFMPEGZoomLoopTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
+    result: Optional[str] = Field(default=None, description="生成结果的URN")
+
+
+class FFMPEGOverlayGifRequest(BaseFFMPEGTaskRequest):
+    media: MediaSource = Field(description="需要处理的媒体源")
+    gif: MediaSource = Field(description="叠加的特效gif")
+
+    @field_validator('media', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+    @field_validator('gif', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            media = MediaSource.from_str(v)
+            if not media.file_extension == 'gif':
+                raise pydantic.ValidationError("必须使用.gif文件")
+            return media
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+
+class FFMPEGOverlayGifTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
+    result: Optional[str] = Field(default=None, description="生成结果的URN")
+
+
+class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
+    media: MediaSource = Field(description="需要处理的媒体源")
+    subtitle: MediaSource = Field(description="需要叠加的字幕文件")
+
+    @field_validator('media', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+    @field_validator('subtitle', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            media = MediaSource.from_str(v)
+            if not media.file_extension in ['ass']:
+                raise pydantic.ValidationError("必须使用标准字幕文件, 如.ass")
+            return media
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+
+class FFMPEGSubtitleTaskStatusResponse(BaseFFMPEGTaskStatusResponse):
+    result: Optional[str] = Field(default=None, description="生成结果的URN")
+
+
+class FFMPEGMixBgmWithNoiseReduceRequest(BaseFFMPEGTaskRequest):
+    media: MediaSource = Field(description="需要处理的媒体源")
+    bgm: MediaSource = Field(description="需要添加的BGM媒体源")
+    video_volume: float = Field(description="最终输出的视频音量系数", default=1.4)
+    music_volume: float = Field(description="最终输出的BGM音量系数", default=0.1)
+    noise_sample: Optional[MediaSource] = Field(description="常考噪音样本，如不指定将使用源视频的前2秒作为样本",
+                                                default=None)
+
+    @field_validator('media', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+    @field_validator('bgm', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[str, MediaSource]):
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+    @field_validator('noise_sample', mode='before')
+    @classmethod
+    def parse_inputs(cls, v: Union[None, str, MediaSource]):
+        if not v:
+            return None
+        if isinstance(v, str):
+            return MediaSource.from_str(v)
+        elif isinstance(v, MediaSource):
+            return v
+        else:
+            raise TypeError(v)
+
+
+class FFMPEGMixBgmWithNoiseReduceStatusResponse(BaseFFMPEGTaskStatusResponse):
+    result: Optional[str] = Field(default=None, description="生成结果的URN")

src/BowongModalFunctions/utils/PathUtils.py

@@ -18,3 +18,12 @@ class FileUtils:
             raise IndexError("Depth is out of range")
         media_dir_prefix = prefix + '/'.join(media_dirs[depth:])
         return media_dir_prefix
+
+    @staticmethod
+    def file_path_change_extension(media_path: str, extension: str) -> str:
+        media_filename = os.path.basename(media_path)
+        media_dir = os.path.dirname(media_path) + "/"
+        filenames = media_filename.split(".")
+        filenames[-1] = extension
+        filename = ".".join(filenames)
+        return os.path.join(media_dir, filename)

src/BowongModalFunctions/utils/VideoUtils.py

@@ -1,6 +1,9 @@
 from typing import Union, List, Tuple, Optional
-
 import numpy as np
+import json, os
+import math
+from BowongModalFunctions.models.media_model import MediaSource
+from .TimeUtils import TimeDelta
 from pydantic import BaseModel, ConfigDict, computed_field
 from ffmpeg import FFmpeg
 from ffmpeg.asyncio import FFmpeg as AsyncFFmpeg
@@ -20,8 +23,7 @@ from pedalboard import (
 )
 from pedalboard.io import AudioFile
 from loguru import logger
-
+from .PathUtils import FileUtils
-from src.cluster.ffmpeg_worker.Utils.PathUtils import FileUtils
 
 
 class MediaStream(BaseModel):
@@ -77,6 +79,14 @@ class VideoUtils:
         video_metadata = VideoMetadata.model_validate_json(ffprobe.execute())
         return video_metadata.streams[0]
 
+    @staticmethod
+    def ffprobe_audio_duration(media_path: str) -> TimeDelta:
+        ffprobe_cmd = VideoUtils.ffmpeg_init(use_ffprobe=True)
+        ffprobe_cmd.input(media_path, print_format="json", show_streams=None)
+        metadata_json = ffprobe_cmd.execute()
+        metadata = VideoMetadata.model_validate_json(metadata_json)
+        return TimeDelta(seconds=metadata.streams[-1].duration)
+
     @staticmethod
     async def ffprobe_video_format_async(media_path: str) -> VideoStream:
         ffprobe = AsyncFFmpeg(executable="ffprobe").input(
@@ -106,7 +116,8 @@ class VideoUtils:
         return video_metadata.streams[0].width, video_metadata.streams[0].height
 
     @staticmethod
-    def noise_reduce(media_path: str, noise_sample_path: Optional[str] = None) -> str:
+    def noise_reduce(media_path: str, noise_sample_path: Optional[str] = None,
+                     output_path: Optional[str] = None) -> str:
         samplerate = 44100
         with AudioFile(media_path).resampled_to(float(samplerate)) as f:
             audio = f.read(f.frames)
@@ -119,6 +130,9 @@ class VideoUtils:
             noise_sample_length = min(int(2 * samplerate), audio.shape[0])
             noise_sample = audio[:noise_sample_length]
 
+        if not output_path:
+            output_path = FileUtils.file_path_extend(media_path, "nr")
+
         reduced_noise = nr.reduce_noise(y=audio, y_noise=noise_sample, sr=samplerate,
                                         stationary=True, prop_decrease=0.75, n_std_thresh_stationary=1.5,
                                         n_fft=2048, win_length=1024, hop_length=512, n_jobs=1)
@@ -171,18 +185,18 @@ class VideoUtils:
             if max_peak > 0.85:
                 additional_safety_factor = 0.85 / max_peak
                 normalized_audio = normalized_audio * additional_safety_factor
-        processed_audio_path = FileUtils.file_path_extend(media_path, "nr")
+
         sf.write(
-            processed_audio_path,
+            output_path,
             normalized_audio,
             samplerate,
             format="WAV",
             subtype="PCM_16",
         )
-        return processed_audio_path
+        return output_path
 
     @staticmethod
-    def ffmpeg_init_async(use_ffprobe: bool = False) -> AsyncFFmpeg:
+    def async_ffmpeg_init(use_ffprobe: bool = False) -> AsyncFFmpeg:
         if use_ffprobe:
             ffmpeg_cmd = AsyncFFmpeg('ffprobe')
         else:
@@ -258,10 +272,81 @@ class VideoUtils:
         return ffmpeg_cmd
 
     @staticmethod
-    async def ffmpeg_extract_audio_async(media_path: str, output_path: str):
+    async def ffmpeg_concat_medias(media_paths: List[str],
-        import json, os
+                                   target_width: int = 1080,
+                                   target_height: int = 1920,
+                                   output_path: Optional[str] = None) -> str:
+        """
+        将待处理的视频合并为一个视频
+        :param media_paths: 待合并的多个视频文件路径
+        :param target_width: 输出的视频分辨率宽
+        :param target_height: 输出的视频分辨率高
+        :param output_path: 指定输出视频路径
+        :return:
+        """
 
-        ffprobe_cmd = VideoUtils.ffmpeg_init_async(use_ffprobe=True)
+        total_videos = len(media_paths)
+        if total_videos == 0:
+            raise ValueError("没有可以合并的视频源")
+        if not output_path:
+            output_path = FileUtils.file_path_extend(media_paths[0], "concat")
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
+        filter_complex = []
+
+        # 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_path,
+            {
+                "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()
+        return output_path
+
+    @staticmethod
+    async def ffmpeg_extract_audio_async(media_path: str, output_path: Optional[str] = None):
+        """
+        提取源视频的音频
+        :param media_path: 待处理的源视频
+        :param output_path: 指定输出的音频文件路径（可选）
+        :return: 最终输出音频文件路径
+        """
+        if not output_path:
+            output_path = FileUtils.file_path_change_extension(output_path, 'wav')
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        ffprobe_cmd = VideoUtils.async_ffmpeg_init(use_ffprobe=True)
         ffprobe_cmd.input(media_path,
                           v="quiet",
                           print_format="json",
@@ -274,10 +359,239 @@ class VideoUtils:
             raise RuntimeError(f"Media has no audio streams.")
         # output_path = f"{output_path_prefix}/extract_audio/outputs/{fn_id}/output.wav"
         os.makedirs(os.path.dirname(output_path), exist_ok=True)
-        ffmpeg_cmd = VideoUtils.ffmpeg_init_async()
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
         ffmpeg_cmd.input(media_path).output(output_path,
                                             map="0:a",
                                             acodec="pcm_s16le",
                                             ar=44100,
                                             ac=1)
         await ffmpeg_cmd.execute()
+
+    @staticmethod
+    async def ffmpeg_mix_bgm(origin_audio_path: str, bgm_audio_path: str, video_volume: float = 1.4,
+                             music_volume: float = 0.1, output_path: Optional[str] = None) -> str:
+        """
+        给待处理视频混合BGM
+        :param origin_audio_path: 待处理的源视频
+        :param bgm_audio_path: 需要混合的BGM
+        :param video_volume: 最终输出视频的音量系数
+        :param music_volume: BGM在源视频音量内占比的音量系数
+        :param output_path: 指定最终输出的视频路径（可选）
+        :return: 最终输出视频文件路径
+        """
+        if not output_path:
+            output_path = FileUtils.file_path_extend(origin_audio_path, "bgm")
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        origin_audio_duration = VideoUtils.ffprobe_audio_duration(origin_audio_path)
+        bgm_duration = VideoUtils.ffprobe_audio_duration(bgm_audio_path)
+        loops_needed = math.ceil(origin_audio_duration.total_seconds() / bgm_duration.total_seconds())
+
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init(use_ffprobe=True)
+        ffmpeg_cmd.input(origin_audio_path)
+        ffmpeg_cmd.input(bgm_audio_path)
+        filter_complex = [
+            f"[0:a]volume={video_volume}[a1]",
+            f"[1:a]aloop=loop={loops_needed}:size={bgm_duration.total_seconds()},volume={music_volume}[a2]"
+            "[a1][a2]amix=inputs=2:duration=first[audio]"
+        ]
+        ffmpeg_cmd.output(output_path,
+                          options={"filter_complex": ";".join(filter_complex), },
+                          map="[audio]",
+                          acodec='libmp3lame',  # 音频编码器
+                          ar=48000,  # 音频采样率
+                          ab='192k',  # 音频码率
+                          ac=2,  # 音频通道数
+                          )
+        await ffmpeg_cmd.execute()
+        return output_path
+
+    @staticmethod
+    async def ffmpeg_mix_bgm_with_noise_reduce(media_path: str, bgm_audio_path: str,
+                                               video_volume: float = 1.4,
+                                               music_volume: float = 0.1,
+                                               noise_sample_path: Optional[str] = None,
+                                               output_path: Optional[str] = None) -> str:
+        """
+        先对待处理的视频音轨降噪，再将降噪后的结果添加BGM，最终输出降噪过且混合BGM的视频；
+        由于最终视频画面和音轨是同步混合+合成视频，所以处理速度会比分步降噪, 加BGM快；
+        :param media_path: 待处理的原始视频路径
+        :param bgm_audio_path: 待处理的BGM音频路径
+        :param video_volume: 最终输出的视频音量系数
+        :param music_volume: 最终输出的BGM音量系数
+        :param noise_sample_path: 降噪使用的噪音样本，如不指定将使用源视频的前2秒作为样本（可选）
+        :param output_path: 指定输出视频的路径（可选）
+        :return: 最终输出视频的路径
+        """
+
+        if not output_path:
+            output_path = FileUtils.file_path_extend(media_path, "bgm_nr")
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        nr_audio_path = VideoUtils.noise_reduce(media_path=media_path, noise_sample_path=noise_sample_path)
+        video_metadata = VideoUtils.ffprobe_video_format(media_path)
+        origin_audio_duration = VideoUtils.ffprobe_audio_duration(nr_audio_path)
+        bgm_duration = VideoUtils.ffprobe_audio_duration(bgm_audio_path)
+        loops_needed = math.ceil(origin_audio_duration.total_seconds() / bgm_duration.total_seconds())
+
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
+        ffmpeg_cmd.input(media_path)
+        ffmpeg_cmd.input(nr_audio_path)
+        ffmpeg_cmd.input(bgm_audio_path)
+        filter_complex = [
+            f"[1:a]volume={video_volume}[a1]",
+            f"[2:a]aloop=loop={loops_needed}:size={bgm_duration.total_seconds()},volume={music_volume}[a2]"
+            "[a1][a2]amix=inputs=2:duration=first[audio]"
+        ]
+        ffmpeg_cmd.output(output_path,
+                          options={"filter_complex": ";".join(filter_complex), },
+                          map=["0:v", "[audio]"],
+                          crf=16,
+                          vcodec='libx264',
+                          b=video_metadata.video_bitrate,  # 视频码率
+                          r=video_metadata.video_frame_rate,  # 帧率
+                          acodec='libmp3lame',
+                          ar=48000, ab='192k', ac=2,
+                          )
+        await ffmpeg_cmd.execute()
+        return output_path
+
+    @staticmethod
+    async def ffmpeg_overlay_gif(media_path: str, overlay_gif_path: str, output_path: Optional[str] = None) -> str:
+        """
+        将GIF特效叠加到视频上，如果视频较长则循环播放GIF
+        :param media_path: 输入视频路径
+        :param overlay_gif_path: GIF特效文件路径
+        :param output_path: 指定输出路径
+        :return: 输出视频路径
+        """
+        if not output_path:
+            output_path = FileUtils.file_path_extend(media_path, "overlay")
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        video_metadata = VideoUtils.ffprobe_video_format(media_path)
+        filter_complex = [
+            # 确保GIF正确解码并循环
+            "[1:v]fps=30,format=rgba[gif]",  # 强制设置30fps
+            # 叠加GIF到视频上，保持透明通道
+            "[0:v][gif]overlay=shortest=1:format=auto,setpts=PTS-STARTPTS[v]",
+        ]
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
+        ffmpeg_cmd.input(media_path)
+        ffmpeg_cmd.input(overlay_gif_path)
+        ffmpeg_cmd.output(output_path,
+                          options={"filter_complex": ";".join(filter_complex), },
+                          map=["[v]", "0:a"],
+                          crf=16,
+                          vcodec='libx264',
+                          b=video_metadata.video_bitrate,  # 视频码率
+                          r=video_metadata.video_frame_rate,  # 帧率
+                          )
+        await ffmpeg_cmd.execute()
+        return output_path
+
+    @staticmethod
+    async def ffmpeg_zoom_loop(media_path: str, duration: float = 6.0, zoom: float = 0.1,
+                               output_path: Optional[str] = None) -> str:
+        """
+        视频放大缩小循环特效
+        :param media_path: 待处理的视频文件路径
+        :param duration: 视频特效循环时间长度
+        :param zoom: 视频特效放大缩小系数
+        :param output_path: 指定输出视频地址（可选）
+        :return: 最终输出视频地址
+        """
+
+        if not output_path:
+            output_path = FileUtils.file_path_extend(media_path, 'zoomed')
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        video_metadata = VideoUtils.ffprobe_video_format(media_path)
+        # abs(sin())表达式会导致实际的往复频率为2倍
+        duration = duration * 2
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
+        ffmpeg_cmd.input(media_path).output(output_path,
+                                            options={
+                                                "vf": f"scale={4 * video_metadata.width}x{4 * video_metadata.height},fps=30,"
+                                                      f"zoompan=z='1+{zoom}*abs(sin(2*PI*time/{duration}))':"
+                                                      "x='trunc(iw/2*(1-1/zoom))':"
+                                                      "y='trunc(ih/2*(1-1/zoom))':"
+                                                      f"d=1:s={video_metadata.width}x{video_metadata.height}:fps=30"
+                                            },
+                                            vcodec="libx264",
+                                            acodec="copy",
+                                            crf=16,
+                                            b=video_metadata.video_bitrate,  # 视频码率
+                                            r=video_metadata.video_frame_rate,  # 帧率
+                                            )
+        await ffmpeg_cmd.execute()
+
+        return output_path
+
+    @staticmethod
+    async def ffmpeg_corner_mirror(media_path: str, mirror_scale_down_size: int = 6,
+                                   mirror_from_right: bool = True, mirror_position: tuple[float, float] = (40, 40),
+                                   output_path: Optional[str] = None) -> str:
+        """
+        对源视频添加镜像小窗特效
+        :param media_path: 待处理的源视频
+        :param mirror_scale_down_size: 源视频画面缩放系数
+        :param mirror_from_right: 小窗原点是否使用右下角
+        :param mirror_position: 小窗基于原点坐标轴的偏移量
+        :param output_path: 指定的输出视频路径（可选）
+        :return: 返回最终输出视频的路径
+        """
+        if not output_path:
+            output_path = FileUtils.file_path_extend(media_path, 'mir')
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
+
+        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]",
+        ]
+        video_metadata = VideoUtils.ffprobe_video_format(media_path)
+        ffmpeg_cmd.input(media_path).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()
+        return output_path
+
+    @staticmethod
+    async def ffmpeg_subtitle_apply(media_path: str, subtitle_path: str,
+                                    font_dir: str, output_path: Optional[str] = None) -> str:
+        """
+        给视频画面叠加字幕，需要确保字幕文件为ass字幕，并且subtitle文件内设置的字体存在与font_dir文件夹内
+        :param media_path: 待处理的源视频
+        :param subtitle_path: ass字幕文件路径
+        :param font_dir: 字体文件目录路径
+        :param output_path: 指定输出文件路径（可选）
+        :return: 返回最终输出视频路径
+        """
+        if not output_path:
+            output_path = FileUtils.file_path_extend(media_path, 'sub')
+
+        video_metadata = VideoUtils.ffprobe_video_format(media_path)
+        ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
+        ffmpeg_cmd.input(media_path)
+        ffmpeg_cmd.output(output_path,
+                          vf=f"subtitles=filename={subtitle_path}:fontsdir={font_dir}",
+                          vcodec="libx264",
+                          crf=16,
+                          b=video_metadata.video_bitrate,  # 视频码率
+                          r=video_metadata.video_frame_rate,  # 帧率
+                          acodec="copy",
+                          )
+        await ffmpeg_cmd.execute()
+        return output_path

src/cluster/app.py

@@ -1,8 +1,8 @@
 import modal
-from config import WorkerConfig
+from BowongModalFunctions.config import WorkerConfig
-from video import media_app
+from .video import app as media_app
-from web import fastapi_app
+from .web import app as web_app
-from ffmpeg_app import ffmpeg_app
+from .ffmpeg_app import app as ffmpeg_app
 
 config = WorkerConfig(
     app_name="bowong-ai-video-test",
@@ -19,6 +19,6 @@ app = modal.App('bowong-modal-ai-video-preview',
                 secrets=[modal.Secret.from_name("cf-kv-secret",
                                                 environment_name=config.environment)])
 
-app.include(fastapi_app)
 app.include(media_app)
 app.include(ffmpeg_app)
+app.include(web_app)

src/cluster/config.py

@@ -1,19 +0,0 @@
-from typing import Optional, Any
-from pydantic import Field
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class WorkerConfig(BaseSettings):
-    app_name: str = Field(default='bowong-ai-video', description="Modal App集群名称")
-    video_downloader_concurrency: Optional[int] = Field(default=10, description="处理缓存任务的并行数")
-    ffmpeg_worker_concurrency: Optional[int] = Field(default=10, description="处理视频合成任务的并行数")
-
-    modal_kv_name: Optional[str] = Field(default='media-cache', description="Modal视频缓存KV库")
-    s3_region: Optional[str] = Field(default='ap-northeast-2', description="S3挂载桶的地域")
-    s3_bucket_name: Optional[str] = Field(default='modal-media-cache', description="集群挂载的S3存储桶")
-    s3_mount_dir: str = Field(default='/mntS3', description="集群S3存储桶挂载在本地的根目录")
-    cdn_endpoint: Optional[str] = Field(default="https://d2nj71io21vkj2.cloudfront.net",
-                                        description="集群挂载S3存储桶的对应AWS Cloudfront CDN")
-    environment: Optional[str] = Field(default="dev", description="Modal worker运行环境")
-
-    modal_config: Any = SettingsConfigDict()

src/cluster/deploy.py

@@ -1,7 +0,0 @@
-import modal.cli.run
-from .config import config
-
-if __name__ == '__main__':
-    # modal.cli.run.deploy(app_ref='src.cluster.app', env=config.environment, use_module_mode=True)
-    # todo : 通过python命令部署不成功，换https://modal.com/docs/guide/continuous-deployment推荐的方式通过命令行部署
-    modal.cli.run.deploy(app_ref='src\\cluster\\app.py', env=config.environment, use_module_mode=False)

src/cluster/ffmpeg_app.py

@@ -3,13 +3,15 @@ import modal
 ffmpeg_worker_image = (
     modal.Image.debian_slim(python_version="3.11")
     .apt_install('ffmpeg')
-    .pip_install(
+    .pip_install_from_pyproject("pyproject.toml")
-        'BowongModalFunctions==0.1.1-pre',
+    .add_local_python_source('cluster')
-        extra_index_url="https://packages-1747117713372:a8865382010036f1e8ad5ca8d0cdab520b78a8ff@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple",
+    .add_local_python_source('BowongModalFunctions')
-    )
 )
 
-ffmpeg_app = modal.App(image=ffmpeg_worker_image, include_source=False)
+app = modal.App(
+    name="ffmpeg_app",
+    image=ffmpeg_worker_image,
+    include_source=False)
 
 with ffmpeg_worker_image.imports():
     import shutil, psutil, os, backoff, json, sentry_sdk
@@ -109,7 +111,7 @@ with ffmpeg_worker_image.imports():
         return s3_outputs
 
 
-    @ffmpeg_app.function(
+    @app.function(
         timeout=1800,
         cloud="aws",
         # region='ap-northeast',
@@ -128,94 +130,25 @@ with ffmpeg_worker_image.imports():
 
         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 = [f"{s3_mount}/{media_source.cache_filepath}" for media_source in media_sources.inputs]
-            input_videos = []
+            local_output_path = await VideoUtils.ffmpeg_concat_medias(media_paths=input_videos,
-            for media_source in media_sources.inputs:
+                                                                      output_path=output_filepath)
-                cache_filepath = f"{s3_mount}/{media_source.cache_filepath}"
+            return local_output_path
-                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)
+        local_output = await ffmpeg_process(media_sources=medias, output_filepath=output_path)
-        s3_output = await ffmpeg_process(medias, output_filepath=output_path)
+        s3_outputs = local_copy_to_s3([local_output])
-        return s3_output, sentry_trace
+        if not sentry_trace:
+            sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
+                                                 x_baggage=sentry_sdk.get_baggage())
+        return s3_outputs[0], sentry_trace
 
 
-    @ffmpeg_app.function(
+    @app.function(
         timeout=900,
         cloud="aws",
         max_containers=config.ffmpeg_worker_concurrency,
@@ -243,7 +176,7 @@ with ffmpeg_worker_image.imports():
             cache_filepath = f"{s3_mount}/{media_source.cache_filepath}"
             logger.info(cache_filepath)
             outputs: List[str] = []
-            ffmpeg_cmd = VideoUtils.ffmpeg_init_async()
+            ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
             ffmpeg_cmd.input(cache_filepath)
             filter_complex: List[str] = []
             for index, marker in enumerate(media_markers):
@@ -287,7 +220,7 @@ with ffmpeg_worker_image.imports():
             hls_m3u8_url = media_source.path
             logger.info(hls_m3u8_url)
             outputs: List[str] = []
-            ffmpeg_cmd = VideoUtils.ffmpeg_init_async()
+            ffmpeg_cmd = VideoUtils.async_ffmpeg_init()
             ffmpeg_cmd.input(hls_m3u8_url,
                              protocol_whitelist="file,http,https,tcp,tls",
                              reconnect="1",  # 自动重连
@@ -338,9 +271,7 @@ with ffmpeg_worker_image.imports():
         return outputs, sentry_trace
 
 
-    @ffmpeg_app.function(
+    @app.function(timeout=600, cloud="aws",
-        timeout=600,
-        cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
                       f"/mntS3": modal.CloudBucketMount(
@@ -376,13 +307,7 @@ with ffmpeg_worker_image.imports():
                 return output, sentry_trace
 
 
-    async def ffmpeg_overlay_gif(media: MediaSource, gif: MediaSource):
+    @app.function(timeout=600, cloud="aws",
-        raise NotImplementedError
-
-
-    @ffmpeg_app.function(
-        timeout=600,
-        cloud="aws",
                   max_containers=config.ffmpeg_worker_concurrency,
                   volumes={
                       f"/mntS3": modal.CloudBucketMount(
@@ -406,34 +331,14 @@ with ffmpeg_worker_image.imports():
                                  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)
+            local_output_filepath = await VideoUtils.ffmpeg_corner_mirror(media_path=media_filepath,
-            ffmpeg_cmd = VideoUtils.ffmpeg_init_async()
+                                                                          output_path=output_path,
-            ffmpeg_cmd.input(media_filepath).output(output_path,
+                                                                          mirror_from_right=mirror_from_right,
-                                                    options={"filter_complex": ";".join(filter_complex)},
+                                                                          mirror_position=mirror_position,
-                                                    map=["[video]", "0:a"],
+                                                                          mirror_scale_down_size=mirror_scale_down_size)
-                                                    vcodec="libx264",
+
-                                                    crf=16,
+            s3_outputs = local_copy_to_s3([local_output_filepath])
-                                                    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,
@@ -444,23 +349,149 @@ with ffmpeg_worker_image.imports():
         return result, sentry_trace
 
 
-    async def ffmpeg_zoom_loop(media: MediaSource, duration: int = 6, zoom: float = 0.1):
+    @app.function(timeout=600, cloud="aws",
-        raise NotImplementedError
+                  max_containers=config.ffmpeg_worker_concurrency,
-
+                  volumes={
-
+                      f"/mntS3": modal.CloudBucketMount(
-    async def ffmpeg_bgm_nosie_reduce(media: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None):
+                          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_overlay_gif(media: MediaSource, gif: MediaSource,
+                                 sentry_trace: Optional[SentryTransactionInfo] = None):
         fn_id = current_function_call_id()
 
-        async def ffmpeg_process(video: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None):
+        @SentryUtils.sentry_tracker(name="视频叠加GIF特效", op="ffmpeg.overlay", fn_id=fn_id,
-            local_input_filepath = f"{s3_mount}/{media.cache_filepath}"
+                                    sentry_trace_id=sentry_trace.x_trace_id if sentry_trace else None,
-            audio_filepath = FileUtils.file_path_extend(local_input_filepath, "audio").replace('mp4', 'wav')
+                                    sentry_baggage=sentry_trace.x_baggage if sentry_trace else None)
-            output_audio_filepath = f"{s3_mount}/{audio_filepath}"
+        async def ffmpeg_process(media: MediaSource, func_id: str, gif: MediaSource) -> str:
-            await VideoUtils.ffmpeg_extract_audio_async(local_input_filepath, output_audio_filepath)
+            media_filepath = f"{s3_mount}/{media.cache_filepath}"
-            # VideoUtils.noise_reduce()
+            gif_filepath = f"{s3_mount}/{gif.cache_filepath}"
-            raise NotImplementedError
+            output_path = f"{output_path_prefix}/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]
 
-        raise NotImplementedError
+        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
 
 
-    async def ffmpeg_subtitle_apply(media: MediaSource, subtitle: MediaSource):
+    @app.function(timeout=600, cloud="aws",
-        raise NotImplementedError
+                  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_zoom_loop(media: MediaSource, duration: int = 6, zoom: float = 0.1,
+                               sentry_trace: Optional[SentryTransactionInfo] = 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)
+        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"
+            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={
+                      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_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):
+        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)
+        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"
+            local_output_filepath = await VideoUtils.ffmpeg_mix_bgm_with_noise_reduce(media_path=local_input_filepath,
+                                                                                      bgm_audio_path=bgm_filepath,
+                                                                                      video_volume=video_volume,
+                                                                                      music_volume=music_volume,
+                                                                                      noise_sample_path=noise_sample_path,
+                                                                                      output_path=output_path)
+
+            s3_outputs = local_copy_to_s3([local_output_filepath])
+            return s3_outputs
+
+        result = await ffmpeg_process(video=media, bgm=bgm, video_volume=video_volume,
+                                      music_volume=music_volume, noise_sample=noise_sample)
+
+        if not sentry_trace:
+            sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
+                                                 x_baggage=sentry_sdk.get_baggage())
+        return result, sentry_trace
+
+
+    @app.function(timeout=600, cloud="aws",
+                  max_containers=config.ffmpeg_worker_concurrency,
+                  volumes={
+                      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_subtitle_apply(media: MediaSource, subtitle: MediaSource,
+                                    sentry_trace: Optional[SentryTransactionInfo] = 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:
+            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"
+
+            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)
+        if not sentry_trace:
+            sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
+                                                 x_baggage=sentry_sdk.get_baggage())
+        return result, sentry_trace

src/cluster/video.py

@@ -3,13 +3,14 @@ import modal
 downloader_image = (
     modal.Image
     .debian_slim(python_version="3.11")
-    .pip_install(
+    .pip_install_from_pyproject("pyproject.toml")
-        'BowongModalFunctions==0.1.1-pre',
+    .add_local_python_source('cluster')
-        extra_index_url="https://packages-1747117713372:a8865382010036f1e8ad5ca8d0cdab520b78a8ff@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple",
+    .add_local_python_source('BowongModalFunctions')
-    )
 )
 
-media_app = modal.App(image=downloader_image,
+app = modal.App(
+    name="media_app",
+    image=downloader_image,
     include_source=False,
     secrets=[
         modal.Secret.from_name("cf-kv-secret", environment_name='dev'),
@@ -110,7 +111,7 @@ with downloader_image.imports():
                 raise e
 
 
-    @media_app.function(cpu=1, timeout=1800,
+    @app.function(cpu=1, timeout=1800,
                   cloud="aws",
                   # region='ap-northeast',
                   max_containers=config.video_downloader_concurrency,
@@ -305,7 +306,7 @@ with downloader_image.imports():
         return media
 
 
-    @media_app.function(cpu=1, timeout=300,
+    @app.function(cpu=1, timeout=300,
                   max_containers=config.video_downloader_concurrency,
                   volumes={
                       "/mntS3": modal.CloudBucketMount(

src/cluster/web.py

@@ -3,21 +3,21 @@ import modal
 fastapi_image = (
     modal.Image
     .debian_slim(python_version="3.11")
-    # .pip_install("fastapi[standard]", "sentry-sdk[fastapi]",
+    .pip_install_from_pyproject("pyproject.toml")
-    #              'loguru', 'pydantic', 'pydantic_settings', 'scalar-fastapi', 'psutil')
+    .add_local_python_source('cluster')
-    .pip_install(
+    .add_local_python_source('BowongModalFunctions')
-        'BowongModalFunctions==0.1.1-pre',
-        extra_index_url="https://packages-1747117713372:a8865382010036f1e8ad5ca8d0cdab520b78a8ff@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple",
-    )
 )
 
-fastapi_app = modal.App(image=fastapi_image, include_source=False)
+app = modal.App(
+    name="web_app",
+    image=fastapi_image,
+    include_source=False)
 
 with fastapi_image.imports():
     from BowongModalFunctions.api import web_app
 
 
-    @fastapi_app.function(scaledown_window=60,
+    @app.function(scaledown_window=60,
                   secrets=[
                       modal.Secret.from_name("cf-kv-secret", environment_name='dev'),
                   ])