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

webhook 添加了backoff retry机制

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-05-29 13:58:27 +08:00
parent a8dafdafb1
commit 73bcd6de68
10 changed files with 206 additions and 423 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@@ -1,6 +1,6 @@
[project]
name = "BowongModalFunctions"
version = "0.1.5"
version = "0.1.6"
authors = [
{ name = "Yudi Xiao", email = "xyd@bowong.ai" },
]

2
src/BowongModalFunctions/api.py
View File

@@ -12,7 +12,7 @@ from .config import WorkerConfig
config = WorkerConfig()
web_app = FastAPI(title="Modal worker API",
version="0.1.5",
version="0.1.6",
summary="Modal Worker的API, 包括缓存视频, 发起生产任务等",
servers=[
{

2
src/BowongModalFunctions/middleware/authorization.py
View File

@@ -2,7 +2,7 @@ from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from starlette import status
bearer_scheme = HTTPBearer()
bearer_scheme = HTTPBearer(scheme_name="bearer", description="固定API Token的鉴权方式", auto_error=True)
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme)):

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

@@ -51,10 +51,15 @@ class ModalTaskResponse(BaseModel):
class WebhookNotify(BaseModel):
endpoint: HttpUrl = Field(description="Webhook回调端点", examples=["https://webhook.example.com"])
method: WebhookMethodEnum = Field(description="Webhook回调请求方法", examples=["get", "post"])
endpoint: HttpUrl = Field(description="Webhook回调端点", examples=["https://webhook.example.com?query=123"])
method: WebhookMethodEnum = Field(
description="Webhook回调请求方法, 使用POST方法时body与查询ffmpeg任务所获得的json body格式一致")
headers: Optional[Dict[str, str]] = Field(description="Webhook回调附带的Headers", default=None)
model_config = ConfigDict(json_schema_extra={
"description": "Webhook返回值与查询ffmpeg任务所获得的json格式一致"
})
class BaseFFMPEGTaskRequest(BaseModel):
webhook: Optional[WebhookNotify] = Field(description="Task webhook", default=None)
@@ -65,6 +70,7 @@ class BaseFFMPEGTaskStatusResponse(BaseModel):
status: TaskStatus = Field(description="任务运行状态")
error: Optional[str] = Field(description="任务错误原因", default=None)
code: Optional[int] = Field(description="任务错误原因代码", default=None)
task_type: str = Field(description="任务类型", default="unknown")
results: Optional[List[Any]] = Field(description="任务运行结果", default=None)
model_config = ConfigDict(extra='ignore')
@@ -354,6 +360,7 @@ class FFMPEGExtractFrameRequest(BaseFFMPEGTaskRequest):
else:
raise TypeError(v)
class FFMPEGExtractFrameStatusResponse(BaseFFMPEGTaskStatusResponse):
result: Optional[str] = Field(default=None, description="生成结果的URN")

120
src/BowongModalFunctions/router/comfyui.py
View File

@@ -1,4 +1,4 @@
from typing import Annotated
from typing import Annotated, cast
import modal
from fastapi import APIRouter, Depends, Header
@@ -11,7 +11,7 @@ from ..models.media_model import MediaSource
from ..utils.ModalUtils import ModalUtils
from ..models.web_model import (SentryTransactionHeader,
ModalTaskResponse, SentryTransactionInfo,
ComfyTaskRequest, ComfyTaskStatusResponse)
ComfyTaskRequest, ComfyTaskStatusResponse, TaskStatus)
config = WorkerConfig()
@@ -32,28 +32,56 @@ sentry_header_schema = {
}
}
@router.post("/v1", tags=["ComfyUI"], summary="发起ComfyUIBase任务", description="发起ComfyUIBase任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v1(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]):
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)
cls = modal.Cls.from_name(config.modal_app_name, "ComfyUI", environment_name=config.modal_environment)
fn_call = cls().api.spawn(item.video_path.path.replace(config.comfyui_s3_input+"/",""), item.start_time, item.filename_prefix, item.tts_text1,
fn_call = cls().api.spawn(item.video_path.path.replace(config.comfyui_s3_input + "/", ""), item.start_time,
item.filename_prefix, item.tts_text1,
item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace,
item.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/v1/{task_id}", tags=["ComfyUI"], summary="查询ComfyUIBase任务",
description="查询ComfyUIBase任务",
description="查询ComfyUIBase任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v1_status(task_id: str, response: Response):
task_info = await ModalUtils.get_modal_task_status(task_id)
media = None
if task_info.transaction:
response.headers["x-trace-id"] = task_info.transaction.x_trace_id
response.headers["x-baggage"] = task_info.transaction.x_baggage
if task_info.status == TaskStatus.success:
task_status = task_info.results[0]["status"]
if task_status != "success":
reason = task_info.results[0]["msg"]
else:
media = MediaSource.from_str(task_info.results[0]["file_name"])
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=cast(int, task_info.error_code.value),
error=reason, result=media.urn if media else None)
@router.post("/v2", tags=["ComfyUI"], summary="发起ComfyUILatentSync1.5任务",
description="发起ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
@@ -62,66 +90,40 @@ async def comfyui_v1(item: ComfyTaskRequest, headers: Annotated[SentryTransactio
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v1_status(task_id: str, response: Response):
task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
media = None
if transaction:
response.headers["x-trace-id"] = transaction.x_trace_id
response.headers["x-baggage"] = transaction.x_baggage
if task_status == "success":
task_status = result["status"]
if task_status != "success":
reason = result["msg"]
else:
media = MediaSource.from_str(result["file_name"])
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason, result=media.urn if media else None)
@router.post("/v2", tags=["ComfyUI"], summary="发起ComfyUILatentSync1.5任务",
description="发起ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v2(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]):
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)
cls = modal.Cls.from_name(config.modal_app_name, "ComfyUILatentSync15", environment_name=config.modal_environment)
fn_call = cls().api.spawn(item.video_path.path.replace(config.comfyui_s3_input+"/",""), item.start_time, item.filename_prefix, item.tts_text1,
fn_call = cls().api.spawn(item.video_path.path.replace(config.comfyui_s3_input + "/", ""), item.start_time,
item.filename_prefix, item.tts_text1,
item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace,
item.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/v2/{task_id}", tags=["ComfyUI"], summary="查询ComfyUILatentSync1.5任务",
description="查询ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
description="查询ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v2_status(task_id: str, response: Response):
task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
task_info = await ModalUtils.get_modal_task_status(task_id)
media = None
if transaction:
response.headers["x-trace-id"] = transaction.x_trace_id
response.headers["x-baggage"] = transaction.x_baggage
if task_status == "success":
task_status = result["status"]
if task_info.transaction:
response.headers["x-trace-id"] = task_info.transaction.x_trace_id
response.headers["x-baggage"] = task_info.transaction.x_baggage
if task_info.status == "success":
task_status = task_info.results[0]["status"]
if task_status != "success":
reason = result["msg"]
reason = task_info.results[0]["msg"]
else:
media = MediaSource.from_str(result["file_name"])
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason,
result=media.urn if media else "")
media = MediaSource.from_str(task_info.results[0]["file_name"])
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=cast(int, task_info.error_code.value),
error=reason, result=media.urn if media else "")

278
src/BowongModalFunctions/router/ffmpeg.py
View File

@@ -1,4 +1,4 @@
from typing import Annotated
from typing import Annotated, cast
import modal
from fastapi import APIRouter, Depends, Header
@@ -30,11 +30,11 @@ from ..models.web_model import (FFMPEGSliceRequest, SentryTransactionHeader,
FFMPEGSubtitleTaskStatusResponse,
FFMPEGZoomLoopTaskStatusResponse,
FFMPEGVideoLoopFillAudioResponse,
FFMPEGMixBgmWithNoiseReduceStatusResponse)
FFMPEGMixBgmWithNoiseReduceStatusResponse, BaseFFMPEGTaskStatusResponse)
config = WorkerConfig()
router = APIRouter(prefix="/ffmpeg")
router = APIRouter(prefix="/ffmpeg", tags=["FFMPEG任务"], dependencies=[Depends(verify_token)])
sentry_header_schema = {
"x-trace-id": {
@@ -52,11 +52,24 @@ sentry_header_schema = {
}
@router.post("/slice",
tags=["发起任务"],
summary="发起切割任务",
description="依据打点信息切出多个片段",
dependencies=[Depends(verify_token)])
@router.get("/task/{task_id}", summary="查询ffmpeg任务状态/结果", responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
)
async def get_task_status(task_id: str, response: Response) -> BaseFFMPEGTaskStatusResponse:
task_info = await ModalUtils.get_modal_task_status(task_id)
if task_info.transaction:
response.headers["x-trace-id"] = task_info.transaction.x_trace_id
response.headers["x-baggage"] = task_info.transaction.x_baggage
return BaseFFMPEGTaskStatusResponse(taskId=task_id, status=task_info.status,
code=cast(int, task_info.error_code.value),
error=task_info.error_reason, results=task_info.results)
@router.post("/slice", summary="发起切割任务", )
async def slice_media(request: FFMPEGSliceRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_slice_media",
@@ -68,31 +81,7 @@ async def slice_media(request: FFMPEGSliceRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/slice/{task_id}",
tags=["查询任务"],
summary="查询切割任务状态/结果",
description="根据任务Id查询运行状态",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def slice_media(task_id: str, response: Response) -> FFMPEGSliceTaskStatusResponse:
task_status, code, reason, results, transaction = await ModalUtils.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 FFMPEGSliceTaskStatusResponse(taskId=task_id, status=task_status, code=code, error=reason,
result=results)
@router.post("/convert",
tags=["发起任务"],
summary="发起直播流转换任务",
dependencies=[Depends(verify_token)],
)
@router.post("/convert", summary="发起直播流转换任务", )
async def convert_media(request: FFMPEGConvertStreamRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_convert_stream",
@@ -104,30 +93,10 @@ async def convert_media(request: FFMPEGConvertStreamRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/convert/{task_id}",
tags=["查询任务"],
summary="获取直播流转换任务结果",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def convert_media_task_status(task_id: str, response: Response) -> FFMPEGConvertStreamResponse:
task_status, code, reason, results, transaction = await ModalUtils.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 FFMPEGConvertStreamResponse(taskId=task_id, status=task_status, code=code, error=reason,
result=results)
@router.post("/concat",
tags=["发起任务"],
summary="发起合并任务",
description="依据AI分析的结果发起合并任务",
dependencies=[Depends(verify_token)])
)
async def concat_media(body: FFMPEGConcatRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
medias = body.medias
@@ -140,33 +109,7 @@ async def concat_media(body: FFMPEGConcatRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/concat/{task_id}",
tags=["查询任务"],
summary="获取合并任务结果",
description="获取合并任务的处理状态和结果",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def concat_media_status(task_id: str, response: Response) -> FFMPEGConcatTaskStatusResponse:
task_status, code, reason, results, transaction = await ModalUtils.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 FFMPEGConcatTaskStatusResponse(taskId=task_id, status=task_status,
code=code, error=reason,
result=results)
@router.post("/extract-audio",
tags=["发起任务"],
summary="发起音频提取任务",
description="提取视频文件的音频",
dependencies=[Depends(verify_token)]
)
@router.post("/extract-audio", summary="发起音频提取任务", )
async def extract_audio(body: FFMPEGExtractAudioRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
media = body.media
@@ -179,29 +122,7 @@ async def extract_audio(body: FFMPEGExtractAudioRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/extract-audio/{task_id}",
tags=["查询任务"],
summary="查询音频提取任务状态",
description="查询音频提取任务状态",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def extract_audio_status(task_id: str, response: Response) -> FFMPEGExtractAudioTaskStatusResponse:
task_status, code, reason, results, transaction = await ModalUtils.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 FFMPEGExtractAudioTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason, result=results)
@router.post("/corner-mirror", tags=["发起任务"], summary="发起镜像小窗去重任务",
description="发起镜像小窗去重任务", dependencies=[Depends(verify_token)])
@router.post("/corner-mirror", summary="发起镜像小窗去重任务", )
async def corner_mirror(body: FFMPEGCornerMirrorRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
media = body.media
@@ -214,28 +135,7 @@ async def corner_mirror(body: FFMPEGCornerMirrorRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/corner-mirror/{task_id}",
tags=["查询任务"],
summary="查询镜像小窗去重任务状态",
description="查询镜像小窗去重任务状态",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def corner_mirror_status(task_id: str, response: Response) -> FFMPEGCornerMirrorTaskStatusResponse:
task_status, code, reason, result, transaction = await ModalUtils.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 FFMPEGCornerMirrorTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason, result=result)
@router.post("/overlay-gif", tags=["发起任务"], summary="发起叠加gif特效任务",
description="发起叠加gif特效任务", dependencies=[Depends(verify_token)])
@router.post("/overlay-gif", summary="发起叠加gif特效任务", )
async def overlay_gif(body: FFMPEGOverlayGifRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_overlay_gif",
@@ -247,26 +147,7 @@ async def overlay_gif(body: FFMPEGOverlayGifRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/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 ModalUtils.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)
@router.post("/zoom-loop", tags=["发起任务"], summary="发起放大缩小循环去重任务",
description="发起放大缩小循环去重任务", dependencies=[Depends(verify_token)])
@router.post("/zoom-loop", summary="发起放大缩小循环去重任务", )
async def zoom_loop(body: FFMPEGZoomLoopRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_zoom_loop", environment_name=config.modal_environment)
@@ -278,26 +159,7 @@ async def zoom_loop(body: FFMPEGZoomLoopRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/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 ModalUtils.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)
@router.post("/subtitle-apply", tags=["发起任务"], summary="发起渲染字幕任务",
description="发起渲染字幕任务", dependencies=[Depends(verify_token)])
@router.post("/subtitle-apply", summary="发起渲染字幕任务", )
async def subtitle_apply(body: FFMPEGSubtitleOverlayRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_subtitle_apply",
@@ -310,26 +172,7 @@ async def subtitle_apply(body: FFMPEGSubtitleOverlayRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/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 ModalUtils.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)
@router.post("/bgm-nosie-reduce", tags=["发起任务"], summary="发起混合BGM并降噪任务",
description="发起混合BGM并降噪任务", dependencies=[Depends(verify_token)])
@router.post("/bgm-nosie-reduce", summary="发起混合BGM并降噪任务", )
async def bgm_nosie_reduce(body: FFMPEGMixBgmWithNoiseReduceRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_bgm_nosie_reduce",
@@ -342,26 +185,7 @@ async def bgm_nosie_reduce(body: FFMPEGMixBgmWithNoiseReduceRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/bgm-nosie-reduce/{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 ModalUtils.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)
@router.post('/video-loop-fill', tags=["发起任务"], summary="发起视频匹配音频长度任务",
description="发起视频匹配音频长度任务", dependencies=[Depends(verify_token)])
@router.post('/video-loop-fill', summary="发起视频匹配音频长度任务")
async def video_loop_fill_audio(body: FFMPEGVideoLoopFillAudioRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_loop_fill",
@@ -373,28 +197,7 @@ async def video_loop_fill_audio(body: FFMPEGVideoLoopFillAudioRequest,
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/video-loop-fill/{task_id}", tags=["查询任务"],
summary="查询视频匹配音频长度任务状态", description="查询视频匹配音频长度任务状态",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def video_loop_fill_audio_status(task_id: str, response: Response) -> FFMPEGVideoLoopFillAudioResponse:
task_status, code, reason, result, transaction = await ModalUtils.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 FFMPEGVideoLoopFillAudioResponse(taskId=task_id, status=task_status, code=code,
error=reason, result=result)
@router.post("/extract-frame",
tags=["发起任务"],
summary="发起视频截取帧画面任务",
dependencies=[Depends(verify_token)])
@router.post("/extract-frame", summary="发起视频截取帧画面任务", )
async def video_extract_frame(body: FFMPEGExtractFrameRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_extract_frame",
@@ -404,22 +207,3 @@ async def video_extract_frame(body: FFMPEGExtractFrameRequest,
sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
fn_call = fn.spawn(media=body.video, frame_index=body.frame_index, sentry_trace=sentry_trace, webhook=body.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/extract-frame/{task_id}",
tags=["查询任务"],
summary="查询视频截取帧画面任务状态",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)])
async def video_extract_frame_status(task_id: str, response: Response) -> FFMPEGExtractFrameStatusResponse:
task_status, code, reason, result, transaction = await ModalUtils.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 FFMPEGExtractFrameStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason, result=result)

6
src/BowongModalFunctions/router/task.py
View File

@@ -5,12 +5,10 @@ from loguru import logger
from ..middleware.authorization import verify_token
from ..models.web_model import ModalTaskCancelResponse
router = APIRouter(prefix="/task")
router = APIRouter(prefix="/task", tags=["tasks"], dependencies=[Depends(verify_token)])
@router.get("/cancel/{task_id}", tags=["取消任务"],
summary="终止任务, 无论是正在排队还是真正运行",
dependencies=[Depends(verify_token)])
@router.get("/cancel/{task_id}", summary="终止任务, 无论是正在排队还是真正运行", )
async def task_cancel(task_id: str) -> ModalTaskCancelResponse:
try:
fn_call = modal.FunctionCall.from_id(task_id)

58
src/BowongModalFunctions/utils/ModalUtils.py
View File

@@ -1,46 +1,66 @@
from typing import Tuple, Optional, Any
from typing import Tuple, Optional, Any, List
import modal
from loguru import logger
from modal.call_graph import InputStatus
from pydantic import BaseModel, Field
from BowongModalFunctions.models.web_model import TaskStatus, SentryTransactionInfo, ErrorCode
class ModalTaskInfo(BaseModel):
function_name: str = Field(description="任务名字")
status: TaskStatus = Field(description="任务运行状态")
error_code: ErrorCode = Field(default=ErrorCode.SUCCESS, description="错误代码")
error_reason: Optional[str] = Field(default=None, description="错误原因")
results: Optional[List[Any]] = Field(default=None, description="任务结果")
transaction: Optional[SentryTransactionInfo] = Field(default=None, description="Sentry跟踪信息")
class ModalUtils:
@staticmethod
async def get_modal_task_status(task_id: str) -> Tuple[
TaskStatus, Optional[int], Optional[str], Optional[Any], Optional[SentryTransactionInfo]]:
async def get_modal_task_status(task_id: str) -> ModalTaskInfo:
"""
使用modal任务id查看任务运行状态和结果
:param task_id: modal 任务id
:return: (TaskStaus 任务运行状态, errorCode 错误代码, errorReason 错误原因, results 任务结果, sentryTransactionInfo Sentry跟踪信息)
:return: (TaskStaus 任务运行状态, function_name 任务名字, errorCode 错误代码, errorReason 错误原因, results 任务结果, sentryTransactionInfo Sentry跟踪信息)
"""
task = None
try:
fn_task = modal.FunctionCall.from_id(task_id)
call_graph = fn_task.get_call_graph()
root = call_graph[0]
if len(root.children) == 0:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
return ModalTaskInfo(function_name="unknown", error_code=ErrorCode.NOT_FOUND,
status=TaskStatus.expired, error_reason="NOT_FOUND")
task = root.children[0]
if not task.function_call_id == task_id:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
logger.info(task)
logger.info(f"Task function call status: {task.status}")
return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.expired,
error_code=ErrorCode.NOT_FOUND, error_reason="NOT_FOUND")
logger.info(f"Task {task.function_name} function call status: {task.status}")
match task.status:
case InputStatus.PENDING:
return TaskStatus.running, None, None, None, None
return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.running)
case InputStatus.SUCCESS:
try:
results, sentry_trace = fn_task.get(timeout=2)
return TaskStatus.success, ErrorCode.SUCCESS.value, None, results, sentry_trace
result, sentry_trace = fn_task.get(timeout=2)
if isinstance(result, List):
results = result
else:
results = [result]
return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.success,
error_code=ErrorCode.SUCCESS, results=results,
transaction=sentry_trace)
except modal.exception.OutputExpiredError:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "EXPIRED", None, None
return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.expired,
error_code=ErrorCode.NOT_FOUND, error_reason="EXPIRED")
except TimeoutError:
return TaskStatus.running, None, None, None, None
return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.running)
except Exception as e:
logger.exception(e)
return TaskStatus.failed, ErrorCode.SYSTEM_ERROR.value, 'FAILURE', None, None
return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.running,
error_code=ErrorCode.SYSTEM_ERROR, error_reason="FAILURE")
case _:
error = 'FAILURE'
match task.status:
@@ -50,16 +70,18 @@ class ModalUtils:
error = 'TERMINATED'
case InputStatus.TIMEOUT:
error = 'TIMEOUT'
return TaskStatus.failed, ErrorCode.SYSTEM_ERROR.value, error, None, None
return ModalTaskInfo(function_name=task.function_name, status=TaskStatus.failed,
error_code=ErrorCode.SYSTEM_ERROR, error_reason=error)
except Exception as e:
logger.exception(e)
return TaskStatus.failed, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
return ModalTaskInfo(function_name=task.function_name if task else "unknown", status=TaskStatus.failed,
error_code=ErrorCode.NOT_FOUND, error_reason="NOT_FOUND")
@staticmethod
async def cancel_modal_task(task_id : str):
async def cancel_modal_task(task_id: str):
try:
fn_task = modal.FunctionCall.from_id(task_id)
fn_task.cancel()
except Exception as e:
logger.exception(e)
return TaskStatus.failed, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
return TaskStatus.failed, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None

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

@@ -1,5 +1,7 @@
import os
from typing import Optional, Tuple, Any, List
import backoff
import httpx
import psutil
import sentry_sdk
@@ -64,6 +66,23 @@ class SentryUtils:
@staticmethod
def webhook_handler(webhook: WebhookNotify, func_id: str):
def decorator(func):
@backoff.on_exception(exception=Exception, wait_gen=backoff.constant,
max_time=15, max_tries=5, raise_on_giveup=True)
def webhook_with_retry(webhook: WebhookNotify, body: BaseFFMPEGTaskStatusResponse):
if webhook.method == WebhookMethodEnum.POST:
response = httpx.post(url=webhook.endpoint.__str__(),
headers=webhook.headers,
json=body.model_dump(),
timeout=3)
else:
response = httpx.get(url=webhook.endpoint.__str__(),
headers=webhook.headers,
params=body.model_dump(),
timeout=3)
response.raise_for_status()
logger.info(f"[{webhook.method}] response = HTTP {response.status_code} {response.text}")
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
try:
@@ -92,26 +111,11 @@ class SentryUtils:
results = result
else:
results = [result]
with httpx.Client() as client:
match webhook.method:
case WebhookMethodEnum.POST:
response = client.post(url=webhook.endpoint.__str__(),
headers=webhook.headers,
json=BaseFFMPEGTaskStatusResponse(
taskId=func_id, status=status,
error=error, code=code,
results=results,
).model_dump())
case WebhookMethodEnum.GET:
response = client.get(url=webhook.endpoint.__str__(),
headers=webhook.headers,
params=BaseFFMPEGTaskStatusResponse(
taskId=func_id, status=status, error=error,
code=code,
results=results,
).model_dump())
logger.info(f"webhook response = {response}")
response.raise_for_status()
webhook_with_retry(webhook=webhook, body=BaseFFMPEGTaskStatusResponse(
taskId=func_id, status=status,
error=error, code=code,
results=results,
))
return result
return async_wrapper

106
src/cluster/ffmpeg_app.py
View File

@@ -44,45 +44,7 @@ with ffmpeg_worker_image.imports():
)
def ffmpeg_init(use_ffprobe: bool = False) -> FFmpeg:
if use_ffprobe:
ffmpeg_cmd = FFmpeg('ffprobe')
else:
ffmpeg_cmd = FFmpeg().option('y').option('hide_banner')
@ffmpeg_cmd.on("start")
def on_start(arguments: list[str]):
try:
filter_index = arguments.index("-filter_complex")
filter_content = arguments[filter_index + 1]
arguments[filter_index + 1] = f'"{filter_content}"'
args = " ".join(arguments)
logger.info(f"FFmpeg command:{args}")
arguments[filter_index + 1] = filter_content
except ValueError:
args = " ".join(arguments)
logger.info(f"FFmpeg command:{args}")
@ffmpeg_cmd.on("progress")
def on_progress(progress):
logger.info(f"处理进度: {progress}")
@ffmpeg_cmd.on("completed")
def on_completed():
logger.info(f"FFMpeg task completed.")
@ffmpeg_cmd.on("stderr")
def on_stderr(line: str):
if line.startswith('Error'):
logger.error(line)
raise RuntimeError(line)
else:
logger.warning(line)
return ffmpeg_cmd
@backoff.on_exception(backoff.constant, exception=Exception, max_tries=5, max_time=60, raise_on_giveup=True)
@backoff.on_exception(wait_gen=backoff.constant, exception=Exception, max_tries=5, max_time=60, raise_on_giveup=True)
def local_copy_to_s3(local_outputs: List[str]) -> List[str]:
s3_outputs = []
for output in local_outputs:
@@ -112,7 +74,7 @@ with ffmpeg_worker_image.imports():
async def ffmpeg_concat_medias(medias: MediaSources,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None) -> Tuple[
str, Optional[SentryTransactionInfo]]:
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@@ -129,11 +91,11 @@ with ffmpeg_worker_image.imports():
content_length=os.path.getsize(local_output_path), )
output_path = f"{output_path_prefix}/{config.modal_environment}/concat/outputs/{fn_id}/output.mp4"
s3_output = await ffmpeg_process(media_sources=medias, output_filepath=output_path)
result = await ffmpeg_process(media_sources=medias, output_filepath=output_path)
if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
return s3_output.urn, sentry_trace
return result, sentry_trace
@app.function(
@@ -153,7 +115,7 @@ with ffmpeg_worker_image.imports():
markers: List[FFMpegSliceSegment],
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None) -> Tuple[
List[str], Optional[SentryTransactionInfo]]:
List[FFMPEGResult], Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频文件切割任务", op="ffmpeg.slice.media", fn_id=fn_id,
@@ -196,7 +158,7 @@ with ffmpeg_worker_image.imports():
media_markers=markers,
fn_id=fn_id)
return [result.urn for result in outputs], sentry_trace
return [result for result in outputs], sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -213,7 +175,7 @@ with ffmpeg_worker_image.imports():
async def ffmpeg_extract_audio(media_source: MediaSource,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None) -> Tuple[
Optional[str], Optional[SentryTransactionInfo]]:
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@@ -228,12 +190,8 @@ with ffmpeg_worker_image.imports():
s3_outputs = local_copy_to_s3([output_path])
return FFMPEGResult(urn=s3_outputs[0], metadata=metadata, content_length=os.path.getsize(output_path), )
match media_source.protocol:
case MediaProtocol.hls:
return None, sentry_trace
case _:
output = await ffmpeg_process(media_source, fn_id=fn_id)
return output.urn, sentry_trace
result = await ffmpeg_process(media_source, fn_id=fn_id)
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -251,7 +209,8 @@ with ffmpeg_worker_image.imports():
mirror_from_right: bool = True,
mirror_position: tuple[float, float] = (40, 40),
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None) -> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频镜像去重", op="ffmpeg.mirror", fn_id=fn_id,
@@ -278,7 +237,7 @@ with ffmpeg_worker_image.imports():
if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
return result.urn, sentry_trace
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -293,7 +252,8 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1)
async def ffmpeg_overlay_gif(media: MediaSource, gif: MediaSource,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None) -> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频叠加GIF特效", op="ffmpeg.overlay", fn_id=fn_id,
@@ -315,7 +275,7 @@ with ffmpeg_worker_image.imports():
if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
return result.urn, sentry_trace
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -330,7 +290,8 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1)
async def ffmpeg_zoom_loop(media: MediaSource, duration: int = 6, zoom: float = 0.1,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None) -> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频放大缩小去重", op="ffmpeg.zoom.loop", fn_id=fn_id,
@@ -353,7 +314,7 @@ with ffmpeg_worker_image.imports():
if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
return result.urn, sentry_trace
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -369,7 +330,8 @@ with ffmpeg_worker_image.imports():
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,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None)-> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频混合BGM降噪", op="ffmpeg.bgm.nosie.reduce", fn_id=fn_id,
@@ -404,7 +366,7 @@ with ffmpeg_worker_image.imports():
if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
return result.urn, sentry_trace
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -419,7 +381,8 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1)
async def ffmpeg_subtitle_apply(media: MediaSource, subtitle: MediaSource,
fonts: List[MediaSource], sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None)-> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频叠加字幕", op="ffmpeg.subtitle.apply", fn_id=fn_id,
@@ -447,7 +410,7 @@ with ffmpeg_worker_image.imports():
if not sentry_trace:
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
return result.urn, sentry_trace
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -462,7 +425,8 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1)
async def ffmpeg_loop_fill(media: MediaSource, audio: MediaSource,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None)-> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="视频匹配音频长度", op="ffmpeg.loop.fill", fn_id=fn_id,
@@ -479,13 +443,13 @@ with ffmpeg_worker_image.imports():
s3_outputs = local_copy_to_s3([local_output])
return FFMPEGResult(urn=s3_outputs[0], metadata=metadata, content_length=os.path.getsize(local_output), )
output = await ffmpeg_process(media=media, audio=audio, func_id=fn_id)
result = await ffmpeg_process(media=media, audio=audio, 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 output.urn, sentry_trace
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -500,7 +464,8 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1)
async def ffmpeg_convert_stream(media: MediaSource,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None)-> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="处理直播流转换为MP4视频任务", op="ffmpeg.convert.stream", fn_id=fn_id,
@@ -517,13 +482,13 @@ with ffmpeg_worker_image.imports():
return FFMPEGResult(urn=s3_outputs[0], metadata=metadata, content_length=os.path.getsize(local_output), )
output = await ffmpeg_process(media_stream=media, func_id=fn_id)
result = await ffmpeg_process(media_stream=media, 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 output.urn, sentry_trace
return result, sentry_trace
@app.function(timeout=600, cloud="aws",
@@ -538,7 +503,8 @@ with ffmpeg_worker_image.imports():
@modal.concurrent(max_inputs=1)
async def ffmpeg_extract_frame(media: MediaSource, frame_index: int,
sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: Optional[WebhookNotify] = None):
webhook: Optional[WebhookNotify] = None)-> Tuple[
FFMPEGResult, Optional[SentryTransactionInfo]]:
fn_id = current_function_call_id()
@SentryUtils.sentry_tracker(name="获取指定的视频帧画面输出为图片", op="ffmpeg.extract.frame", fn_id=fn_id,
@@ -554,8 +520,8 @@ with ffmpeg_worker_image.imports():
s3_outputs = local_copy_to_s3([local_output])
return FFMPEGResult(urn=s3_outputs[0], metadata=metadata, content_length=os.path.getsize(local_output), )
output = await ffmpeg_process(media=media, frame_index=frame_index, func_id=fn_id)
result = await ffmpeg_process(media=media, frame_index=frame_index, 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 output.urn, sentry_trace
return result, sentry_trace