diff --git a/.gitignore b/.gitignore
index 18ba43b..d7949a5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1 +1,2 @@
-*/**/.pypirc
\ No newline at end of file
+.pypirc
+.env
\ No newline at end of file
diff --git a/.idea/runConfigurations/modal_deploy.xml b/.idea/runConfigurations/modal_deploy.xml
index efb908e..120d67e 100644
--- a/.idea/runConfigurations/modal_deploy.xml
+++ b/.idea/runConfigurations/modal_deploy.xml
@@ -14,7 +14,7 @@
-
+
diff --git a/README.md b/README.md
index cbb4a96..90e973e 100644
--- a/README.md
+++ b/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
```
\ No newline at end of file
diff --git a/pyproject.toml b/pyproject.toml
index 63eb6c5..970529b 100644
--- a/pyproject.toml
+++ b/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"]
diff --git a/src/BowongModalFunctions/__init__.py b/src/BowongModalFunctions/__init__.py
index 5745e00..e5c39f8 100644
--- a/src/BowongModalFunctions/__init__.py
+++ b/src/BowongModalFunctions/__init__.py
@@ -1 +1 @@
-__version__ = "0.1.1-pre"
\ No newline at end of file
+__version__ = "0.1.1-pre3"
\ No newline at end of file
diff --git a/src/BowongModalFunctions/api.py b/src/BowongModalFunctions/api.py
index ef44a84..f44da34 100644
--- a/src/BowongModalFunctions/api.py
+++ b/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="",
+ 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)
diff --git a/src/BowongModalFunctions/models/media_model.py b/src/BowongModalFunctions/models/media_model.py
index d415dbe..0f562b6 100644
--- a/src/BowongModalFunctions/models/media_model.py
+++ b/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:
diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py
index d4e3c78..d1eaa90 100644
--- a/src/BowongModalFunctions/models/web_model.py
+++ b/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")
diff --git a/src/BowongModalFunctions/utils/PathUtils.py b/src/BowongModalFunctions/utils/PathUtils.py
index 6975eb9..a1609ef 100644
--- a/src/BowongModalFunctions/utils/PathUtils.py
+++ b/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)
diff --git a/src/BowongModalFunctions/utils/VideoUtils.py b/src/BowongModalFunctions/utils/VideoUtils.py
index 1a605e6..5c2de6a 100644
--- a/src/BowongModalFunctions/utils/VideoUtils.py
+++ b/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 src.cluster.ffmpeg_worker.Utils.PathUtils import FileUtils
+from .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):
- import json, os
+ async def ffmpeg_concat_medias(media_paths: List[str],
+ 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)
+ 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
diff --git a/src/cluster/app.py b/src/cluster/app.py
index 91e4366..d0d79f2 100644
--- a/src/cluster/app.py
+++ b/src/cluster/app.py
@@ -1,8 +1,8 @@
import modal
-from config import WorkerConfig
-from video import media_app
-from web import fastapi_app
-from ffmpeg_app import ffmpeg_app
+from BowongModalFunctions.config import WorkerConfig
+from .video import app as media_app
+from .web import app as web_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)
diff --git a/src/cluster/config.py b/src/cluster/config.py
deleted file mode 100644
index e9d5825..0000000
--- a/src/cluster/config.py
+++ /dev/null
@@ -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()
diff --git a/src/cluster/deploy.py b/src/cluster/deploy.py
deleted file mode 100644
index a22462b..0000000
--- a/src/cluster/deploy.py
+++ /dev/null
@@ -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)
diff --git a/src/cluster/ffmpeg_app.py b/src/cluster/ffmpeg_app.py
index ee4145e..81c3500 100644
--- a/src/cluster/ffmpeg_app.py
+++ b/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(
- 'BowongModalFunctions==0.1.1-pre',
- extra_index_url="https://packages-1747117713372:a8865382010036f1e8ad5ca8d0cdab520b78a8ff@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple",
- )
+ .pip_install_from_pyproject("pyproject.toml")
+ .add_local_python_source('cluster')
+ .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 = []
- for media_source in media_sources.inputs:
- cache_filepath = f"{s3_mount}/{media_source.cache_filepath}"
- logger.info(cache_filepath)
- ffmpeg_cmd.input(cache_filepath)
- input_videos.append(cache_filepath)
- target_width = 1080
- target_height = 1920
- # video_info = await print_media_streams_info(media_paths=input_videos)
-
- filter_complex = []
- total_videos = len(input_videos)
-
- # 2. 统一所有视频的格式、分辨率和帧率
- for i in range(total_videos):
- filter_complex.extend(
- [
- # 先缩放到统一分辨率，然后设置帧率和格式
- f"[{i}:v]scale={target_width}:{target_height}:force_original_aspect_ratio=decrease,"
- f"pad={target_width}:{target_height}:(ow-iw)/2:(oh-ih)/2,"
- f"setsar=1:1," # 新增强制设置SAR
- f"fps=30,format=yuv420p[v{i}]",
- f"[{i}:a]aformat=sample_fmts=fltp:sample_rates=44100:channel_layouts=stereo[a{i}]",
- ]
- )
-
- # 3. 准备处理后的视频流和音频流的连接字符串
- video_streams = "".join(f"[v{i}]" for i in range(total_videos))
- audio_streams = "".join(f"[a{i}]" for i in range(total_videos))
-
- # 4. 使用concat过滤器合并视频和音频
- filter_complex.extend(
- [
- f"{video_streams}concat=n={total_videos}:v=1:a=0[vconcated]",
- f"{audio_streams}concat=n={total_videos}:v=0:a=1[aconcated]",
- ]
- )
-
- ffmpeg_cmd.output(
- output_filepath,
- {
- "filter_complex": ";".join(filter_complex),
- "map": ["[vconcated]", "[aconcated]"],
- "vcodec": "libx264",
- "crf": 16,
- "r": 30,
- "acodec": "aac",
- "ar": 44100,
- "ac": 2,
- "ab": "192k",
- },
- )
-
- await ffmpeg_cmd.execute()
- s3_outputs = local_copy_to_s3([output_filepath])
- return s3_outputs[0]
+ 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"
- os.makedirs(os.path.dirname(output_path), exist_ok=True)
- s3_output = await ffmpeg_process(medias, output_filepath=output_path)
- return s3_output, sentry_trace
+ 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())
+ 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,18 +271,16 @@ with ffmpeg_worker_image.imports():
return outputs, sentry_trace
- @ffmpeg_app.function(
- timeout=600,
- cloud="aws",
- max_containers=config.ffmpeg_worker_concurrency,
- volumes={
- f"/mntS3": modal.CloudBucketMount(
- bucket_name=config.s3_bucket_name,
- secret=modal.Secret.from_name("aws-s3-secret",
- environment_name=config.environment),
- ),
- },
- )
+ @app.function(timeout=600, cloud="aws",
+ max_containers=config.ffmpeg_worker_concurrency,
+ volumes={
+ f"/mntS3": modal.CloudBucketMount(
+ bucket_name=config.s3_bucket_name,
+ secret=modal.Secret.from_name("aws-s3-secret",
+ environment_name=config.environment),
+ ),
+ },
+ )
@modal.concurrent(max_inputs=1)
async def ffmpeg_extract_audio(media_source: MediaSource,
webhook: Optional[str], # todo handle webhook callback
@@ -376,21 +307,15 @@ with ffmpeg_worker_image.imports():
return output, sentry_trace
- async def ffmpeg_overlay_gif(media: MediaSource, gif: MediaSource):
- raise NotImplementedError
-
-
- @ffmpeg_app.function(
- timeout=600,
- cloud="aws",
- max_containers=config.ffmpeg_worker_concurrency,
- volumes={
- f"/mntS3": modal.CloudBucketMount(
- bucket_name=config.s3_bucket_name,
- secret=modal.Secret.from_name("aws-s3-secret",
- environment_name=config.environment),
- ),
- }, )
+ @app.function(timeout=600, cloud="aws",
+ max_containers=config.ffmpeg_worker_concurrency,
+ volumes={
+ f"/mntS3": modal.CloudBucketMount(
+ bucket_name=config.s3_bucket_name,
+ secret=modal.Secret.from_name("aws-s3-secret",
+ environment_name=config.environment),
+ ),
+ }, )
@modal.concurrent(max_inputs=1)
async def ffmpeg_corner_mirror(media: MediaSource,
mirror_scale_down_size: int = 6,
@@ -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)
- ffmpeg_cmd = VideoUtils.ffmpeg_init_async()
- ffmpeg_cmd.input(media_filepath).output(output_path,
- options={"filter_complex": ";".join(filter_complex)},
- map=["[video]", "0:a"],
- vcodec="libx264",
- crf=16,
- b=video_metadata.video_bitrate, # 视频码率
- r=video_metadata.video_frame_rate # 帧率
- )
- await ffmpeg_cmd.execute()
- s3_outputs = local_copy_to_s3([output_path])
+ local_output_filepath = await VideoUtils.ffmpeg_corner_mirror(media_path=media_filepath,
+ output_path=output_path,
+ mirror_from_right=mirror_from_right,
+ mirror_position=mirror_position,
+ mirror_scale_down_size=mirror_scale_down_size)
+
+ s3_outputs = local_copy_to_s3([local_output_filepath])
return s3_outputs[0]
result = await ffmpeg_process(media=media, mirror_scale_down_size=mirror_scale_down_size, func_id=fn_id,
@@ -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):
- raise NotImplementedError
-
-
- async def ffmpeg_bgm_nosie_reduce(media: MediaSource, bgm: MediaSource, noise_sample: Optional[MediaSource] = None):
+ @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_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):
- local_input_filepath = f"{s3_mount}/{media.cache_filepath}"
- audio_filepath = FileUtils.file_path_extend(local_input_filepath, "audio").replace('mp4', 'wav')
- output_audio_filepath = f"{s3_mount}/{audio_filepath}"
- await VideoUtils.ffmpeg_extract_audio_async(local_input_filepath, output_audio_filepath)
- # VideoUtils.noise_reduce()
- raise NotImplementedError
+ @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)
+ 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"
+ 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):
- raise NotImplementedError
+ @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_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
diff --git a/src/cluster/video.py b/src/cluster/video.py
index bc58db1..5ceaf36 100644
--- a/src/cluster/video.py
+++ b/src/cluster/video.py
@@ -3,17 +3,18 @@ import modal
downloader_image = (
modal.Image
.debian_slim(python_version="3.11")
- .pip_install(
- 'BowongModalFunctions==0.1.1-pre',
- extra_index_url="https://packages-1747117713372:a8865382010036f1e8ad5ca8d0cdab520b78a8ff@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple",
- )
+ .pip_install_from_pyproject("pyproject.toml")
+ .add_local_python_source('cluster')
+ .add_local_python_source('BowongModalFunctions')
)
-media_app = modal.App(image=downloader_image,
- include_source=False,
- secrets=[
- modal.Secret.from_name("cf-kv-secret", environment_name='dev'),
- ])
+app = modal.App(
+ name="media_app",
+ image=downloader_image,
+ include_source=False,
+ secrets=[
+ modal.Secret.from_name("cf-kv-secret", environment_name='dev'),
+ ])
with downloader_image.imports():
import os, httpx, crcmod
@@ -110,17 +111,17 @@ with downloader_image.imports():
raise e
- @media_app.function(cpu=1, timeout=1800,
- cloud="aws",
- # region='ap-northeast',
- max_containers=config.video_downloader_concurrency,
- volumes={
- "/mntS3": modal.CloudBucketMount(
- bucket_name=config.s3_bucket_name,
- secret=modal.Secret.from_name("aws-s3-secret", environment_name=config.environment),
- ),
- },
- secrets=[modal.Secret.from_name("tencent-cloud-secret", environment_name=config.environment)])
+ @app.function(cpu=1, timeout=1800,
+ cloud="aws",
+ # region='ap-northeast',
+ max_containers=config.video_downloader_concurrency,
+ volumes={
+ "/mntS3": modal.CloudBucketMount(
+ bucket_name=config.s3_bucket_name,
+ secret=modal.Secret.from_name("aws-s3-secret", environment_name=config.environment),
+ ),
+ },
+ secrets=[modal.Secret.from_name("tencent-cloud-secret", environment_name=config.environment)])
@modal.concurrent(max_inputs=10)
async def cache_submit(media: MediaSource, sentry_trace: SentryTransactionInfo) -> MediaSource:
def vod_init():
@@ -305,14 +306,14 @@ with downloader_image.imports():
return media
- @media_app.function(cpu=1, timeout=300,
- max_containers=config.video_downloader_concurrency,
- volumes={
- "/mntS3": modal.CloudBucketMount(
- bucket_name=config.s3_bucket_name,
- secret=modal.Secret.from_name("aws-s3-secret", environment_name=config.environment),
- ),
- })
+ @app.function(cpu=1, timeout=300,
+ max_containers=config.video_downloader_concurrency,
+ volumes={
+ "/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=10)
async def cache_delete(cache: MediaSource) -> MediaSource:
if os.path.exists(cache.cache_filepath):
diff --git a/src/cluster/web.py b/src/cluster/web.py
index c2f7950..c4c9efc 100644
--- a/src/cluster/web.py
+++ b/src/cluster/web.py
@@ -3,24 +3,24 @@ import modal
fastapi_image = (
modal.Image
.debian_slim(python_version="3.11")
- # .pip_install("fastapi[standard]", "sentry-sdk[fastapi]",
- # 'loguru', 'pydantic', 'pydantic_settings', 'scalar-fastapi', 'psutil')
- .pip_install(
- 'BowongModalFunctions==0.1.1-pre',
- extra_index_url="https://packages-1747117713372:a8865382010036f1e8ad5ca8d0cdab520b78a8ff@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple",
- )
+ .pip_install_from_pyproject("pyproject.toml")
+ .add_local_python_source('cluster')
+ .add_local_python_source('BowongModalFunctions')
)
-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,
- secrets=[
- modal.Secret.from_name("cf-kv-secret", environment_name='dev'),
- ])
+ @app.function(scaledown_window=60,
+ secrets=[
+ modal.Secret.from_name("cf-kv-secret", environment_name='dev'),
+ ])
@modal.concurrent(max_inputs=100)
# @modal.asgi_app(custom_domains=["modal-dev.bowong.cc"])
@modal.asgi_app()