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

按功能模块拆分API路由到不同文件

Browse Source
This commit is contained in:
shuohigh@gmail.com
2025-05-19 15:02:34 +08:00
parent c21e999188
commit 668a5a33d4
20 changed files with 708 additions and 629 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.runtime.env
View File

@@ -1,5 +1,5 @@
MODAL_ENVIRONMENT=dev
modal_app_name=cluster-test
modal_app_name=bowong-ai-video
S3_mount_dir=/mntS3
S3_bucket_name=modal-media-cache
S3_region=ap-northeast-2

6
README.md
View File

@@ -6,6 +6,12 @@
modal deploy -m cluster.app
```
## 本地验证fastapi部署
```bash
uvicorn BowongModalFunctions.api:web_app --host 0.0.0.0 --port 8080
```
# Pip
## 构建

6
pyproject.toml
View File

@@ -41,7 +41,11 @@ license = "MIT"
license-files = ["LICEN[CS]E*"]
[project.optional-dependencies]
cli = ["twine>=6.1.0"]
cli = [
"gunicorn>=23.0.0",
"twine>=6.1.0",
"uvicorn[standard]>=0.34.2",
]
[build-system]
requires = ["hatchling >= 1.26"]

2
src/BowongModalFunctions/__init__.py
View File

@@ -1 +1 @@
__version__ = "0.1.1-pre3"
__version__: str = "0.1.2"

616
src/BowongModalFunctions/api.py
View File

@@ -1,58 +1,29 @@
import os
import modal
import httpx
import asyncio
from typing import List, Annotated, Tuple, Any, Optional
from modal import current_function_call_id
from modal.call_graph import InputStatus
from loguru import logger
from typing import Annotated
from fastapi import FastAPI, Response, Depends, Header, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.exceptions import HTTPException
from starlette import status
from scalar_fastapi import get_scalar_api_reference
import sentry_sdk
from sentry_sdk.integrations.loguru import LoguruIntegration, LoggingLevels
from sentry_sdk.integrations.fastapi import FastApiIntegration
from fastapi.middleware.cors import CORSMiddleware
from .middleware.authorization import verify_token
from .utils.KVCache import KVCache
from .utils.SentryUtils import SentryUtils
from .models.media_model import (MediaSource,
MediaSources,
MediaCacheStatus,
CacheResult,
DownloadResult)
from .models.web_model import (TaskStatus,
ErrorCode,
SentryTransactionInfo,
from .utils.ModalUtils import ModalUtils
from .models.media_model import MediaSource
from .models.web_model import (SentryTransactionInfo,
SentryTransactionHeader,
ModalTaskResponse,
FFMPEGSliceRequest,
FFMPEGConcatRequest,
FFMPEGExtractAudioRequest,
FFMPEGCornerMirrorRequest,
FFMPEGOverlayGifRequest,
FFMPEGZoomLoopRequest,
FFMPEGSubtitleOverlayRequest,
FFMPEGMixBgmWithNoiseReduceRequest,
FFMPEGSliceTaskStatusResponse,
FFMPEGConcatTaskStatusResponse,
FFMPEGExtractAudioTaskStatusResponse,
FFMPEGCornerMirrorTaskStatusResponse,
FFMPEGOverlayGifTaskStatusResponse,
FFMPEGSubtitleTaskStatusResponse,
FFMPEGZoomLoopTaskStatusResponse,
FFMPEGMixBgmWithNoiseReduceStatusResponse,
ComfyTaskRequest,
ComfyTaskStatusResponse
)
from .router import ffmpeg, cache
from .config import WorkerConfig
bearer_scheme = HTTPBearer()
config = WorkerConfig()
web_app = FastAPI(title="Modal worker API",
version="0.1.2",
summary="Modal Worker的API, 包括缓存视频, 发起生产任务等",
servers=[
{'url': f'https://bowongai-dev--{config.modal_app_name}-fastapi-webapp.modal.run',
@@ -66,25 +37,6 @@ web_app = FastAPI(title="Modal worker API",
}
])
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme)):
token = credentials.credentials
# 在这里实现你的token验证逻辑
if not is_valid_token(token):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return token
def is_valid_token(token: str) -> bool:
# 这里实现具体的token验证逻辑
# 例如：验证JWT token，检查数据库中的token等
return token == "bowong7777"
sentry_sdk.init(dsn="https://dab7b7ae652216282c89f029a76bb10a@sentry.bowongai.com/2",
send_default_pii=True,
traces_sample_rate=1.0,
@@ -97,9 +49,6 @@ sentry_sdk.init(dsn="https://dab7b7ae652216282c89f029a76bb10a@sentry.bowongai.co
]
)
modal_kv_cache = KVCache(kv_name=config.modal_kv_name, environment=config.modal_environment)
cf_account_id = os.environ.get("CF_ACCOUNT_ID")
cf_kv_api_token = os.environ.get("CF_KV_API_TOKEN")
cf_kv_namespace_id = os.environ.get("CF_KV_NAMESPACE_ID")
sentry_header_schema = {
"x-trace-id": {
@@ -120,6 +69,14 @@ ALIAS_MAP = {
"/comfyui": "/comfyui/v2",
}
web_app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@web_app.middleware("http")
@web_app.middleware("https")
@@ -129,538 +86,13 @@ async def alias_middleware(request: Request, call_next):
return await call_next(request)
@sentry_sdk.trace
def batch_update_cloudflare_kv(caches: List[MediaSource]):
with httpx.Client() as client:
try:
response = client.put(
f"https://api.cloudflare.com/client/v4/accounts/{cf_account_id}/storage/kv/namespaces/{cf_kv_namespace_id}/bulk",
headers={"Authorization": f"Bearer {cf_kv_api_token}"},
json=[
{
"based64": False,
"key": cache.urn,
"value": cache.model_dump_json(),
}
for cache in caches
]
)
response.raise_for_status()
except httpx.RequestError as e:
logger.error(f"An error occurred while put kv to cloudflare")
raise e
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
raise e
except Exception as e:
logger.error(f"An unexpected error occurred: {str(e)}")
raise e
@sentry_sdk.trace
def batch_remove_cloudflare_kv(keys: List[str]):
with httpx.Client() as client:
try:
response = client.post(
f"https://api.cloudflare.com/client/v4/accounts/{cf_account_id}/storage/kv/namespaces/{cf_kv_namespace_id}/bulk/delete",
headers={"Authorization": f"Bearer {cf_kv_api_token}"},
json=keys
)
response.raise_for_status()
except httpx.RequestError as e:
logger.error(f"An error occurred while put kv to cloudflare")
raise e
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
raise e
except Exception as e:
logger.error(f"An unexpected error occurred: {str(e)}")
raise e
@sentry_sdk.trace
async def get_modal_task_status(task_id: str) -> Tuple[
TaskStatus, Optional[int], Optional[str], Optional[Any], Optional[SentryTransactionInfo]]:
"""
使用modal任务id查看任务运行状态和结果
:param task_id: modal 任务id
:return: (TaskStaus 任务运行状态, errorCode 错误代码, errorReason 错误原因, results 任务结果, sentryTransactionInfo Sentry跟踪信息)
"""
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
task = root.children[0]
if not task.function_call_id == task_id:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
logger.info(task)
logger.info(f"Task function call status: {task.status}")
match task.status:
case InputStatus.PENDING:
return TaskStatus.running, None, None, None, None
case InputStatus.SUCCESS:
try:
results, sentry_trace = fn_task.get(timeout=2)
return TaskStatus.success, ErrorCode.SUCCESS.value, None, results, sentry_trace
except modal.exception.OutputExpiredError:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "EXPIRED", None, None
except TimeoutError:
return TaskStatus.running, None, None, None, None
except Exception as e:
logger.exception(e)
return TaskStatus.failed, ErrorCode.SYSTEM_ERROR.value, 'FAILURE', None, None
case _:
error = 'FAILURE'
match task.status:
case InputStatus.INIT_FAILURE:
error = 'INIT_FAILURE'
case InputStatus.TERMINATED:
error = 'TERMINATED'
case InputStatus.TIMEOUT:
error = 'TIMEOUT'
return TaskStatus.failed, ErrorCode.SYSTEM_ERROR.value, error, None, None
except Exception as e:
logger.exception(e)
return TaskStatus.failed, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
@web_app.get("/scalar", include_in_schema=False)
async def scalar():
return get_scalar_api_reference(openapi_url=web_app.openapi_schema, title="Modal worker web endpoint")
@web_app.post("/cache",
tags=["缓存"],
summary="缓存视频文件",
description="异步缓存视频文件到S3存储桶和Modal Dict(KV)",
dependencies=[Depends(verify_token)])
@sentry_sdk.trace
async def cache(medias: MediaSources) -> CacheResult:
fn_id = current_function_call_id()
caches: MediaSources
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
@SentryUtils.sentry_tracker(name="同步视频缓存", op="cache.get", fn_id=fn_id,
sentry_trace_id=None, sentry_baggage=None)
async def cache_handler(media: MediaSource):
cache_span = sentry_sdk.get_current_span()
cache_span.set_data("runner_id", fn_id)
cache_span.set_data("cache.key", [media.urn])
cached_media = modal_kv_cache.get_cache(media.urn)
cache_hit: bool = False
if not cached_media:
# start new download task
with cache_span.start_child(name="视频缓存任务入队",
op="queue.publish") as queue_publish_span:
fn = modal.Function.from_name(config.modal_app_name, 'cache_submit')
fn_task = fn.spawn(media, sentry_trace)
queue_publish_span.set_data("cache.key", media.urn)
queue_publish_span.set_data("messaging.message.id", fn_task.object_id)
queue_publish_span.set_data("messaging.destination.name",
"video-downloader.cache_submit")
queue_publish_span.set_data("messaging.message.body.size", 0)
# video_cache = MediaCache(status=MediaCacheStatus.downloading,
# downloader_id=fn_task.object_id)
media.status = MediaCacheStatus.downloading
media.downloader_id = fn_task.object_id
# video_cache_status_json = video_cache.model_dump_json()
modal_kv_cache.set_cache(media)
else:
media = cached_media
# video_cache = MediaCache.model_validate_json(video_cache_status_json)
match media.status:
case MediaCacheStatus.ready:
cache_hit = True
case MediaCacheStatus.downloading: # 下载任务已经在进行
cache_hit = True
case _:
# start new download task
with cache_span.start_child(name="视频缓存任务入队",
op="queue.publish") as queue_publish_span:
fn = modal.Function.from_name(config.modal_app_name, 'cache_submit')
fn_task = fn.spawn(media, sentry_trace)
queue_publish_span.set_data("cache.key", media.urn)
queue_publish_span.set_data("messaging.message.id", fn_task.object_id)
queue_publish_span.set_data("messaging.destination.name",
"video-downloader.cache_submit")
queue_publish_span.set_data("messaging.message.body.size", 0)
media.status = MediaCacheStatus.downloading
media.downloader_id = fn_task.object_id
# video_cache = MediaCache(status=MediaCacheStatus.downloading,
# downloader_id=fn_task.object_id)
# video_cache_status_json = video_cache.model_dump_json()
modal_kv_cache.set_cache(media)
cache_hit = False
# caches[media.urn] = video_cache
# logger.info(f"Media cache hit ? {cache_hit}")
cache_span.set_data("cache.hit", cache_hit)
return media
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(cache_handler(media)) for media in medias.inputs]
cache_task_result = [task.result() for task in tasks]
batch_update_cloudflare_kv(cache_task_result)
return CacheResult(caches={media.urn: media for media in cache_task_result})
@web_app.post("/cache/download",
tags=["缓存"],
summary="批量获取下载地址",
description="获取已缓存的视频下载地址",
dependencies=[Depends(verify_token)])
@sentry_sdk.trace
async def download_caches(medias: MediaSources) -> DownloadResult:
cdn_endpoint = config.S3_cdn_endpoint
urls = []
for media in medias.inputs:
urls.append(f"{cdn_endpoint}/{media.get_cdn_url()}")
return DownloadResult(urls=urls)
@web_app.get("/cache/download",
tags=["缓存"],
summary="下载已缓存的视频",
description="通过CDN下载已缓存的视频文件")
@sentry_sdk.trace
async def download_cache(media: str) -> RedirectResponse:
cdn_endpoint = config.S3_cdn_endpoint
media = MediaSource.from_str(media)
return RedirectResponse(url=f"{cdn_endpoint}/{media.get_cdn_url()}", status_code=status.HTTP_302_FOUND)
@web_app.delete("/cache/kv",
tags=["缓存"],
summary="清除KV记录",
description="清除当前环境下KV缓存过的所有数据(S3存储桶内的文件会保留)",
dependencies=[Depends(verify_token)])
async def purge_kv_all():
parent = sentry_sdk.get_current_span()
span = parent.start_child(name="清除缓存KV", op="purge.flush")
modal_kv_cache.clear()
span.set_data("cache.success", True)
span.finish()
return JSONResponse(content={"success": True})
@web_app.post("/cache/kv",
tags=["缓存"],
summary="删除对应的KV记录",
description="删除请求中对应的视频缓存记录",
dependencies=[Depends(verify_token)])
async def purge_kv(medias: MediaSources):
try:
for media in medias.inputs:
modal_kv_cache.pop(media.urn)
keys = [media.urn for media in medias.inputs]
batch_remove_cloudflare_kv(keys)
return JSONResponse(content={"success": True, "keys": keys})
except Exception as e:
return JSONResponse(content={"success": False, "error": str(e)})
@web_app.post("/cache/media",
tags=["缓存"],
summary="清除指定的所有缓存",
description="清除指定的所有缓存(包括KV记录和S3存储文件)",
dependencies=[Depends(verify_token)])
async def purge_media(medias: MediaSources):
fn_id = current_function_call_id()
fn = modal.Function.from_name(config.modal_app_name, "cache_delete", environment_name=config.modal_environment)
@SentryUtils.sentry_tracker(name="清除媒体源缓存", op="cache.purge", fn_id=fn_id,
sentry_trace_id=None, sentry_baggage=None)
async def purge_handle(media: MediaSource):
cache_media = modal_kv_cache.pop(media.urn)
if cache_media:
deleted_cache: MediaSource = await fn.remote.aio(cache_media)
return deleted_cache.urn
return None
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(purge_handle(media)) for media in medias.inputs]
keys = [task.result() for task in tasks]
batch_remove_cloudflare_kv(keys)
return JSONResponse(content={"success": True, "keys": keys})
@web_app.post("/ffmpeg/slice",
tags=["发起任务"],
summary="发起切割任务",
description="依据打点信息切出多个片段",
dependencies=[Depends(verify_token)])
async def slice_media(request: FFMPEGSliceRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_slice_media",
environment_name=config.modal_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=request.media, markers=request.markers, sentry_trace=sentry_trace, webhook=request.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/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 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)
@web_app.post("/ffmpeg/concat",
tags=["发起任务"],
summary="发起合并任务",
description="依据AI分析的结果发起合并任务",
dependencies=[Depends(verify_token)])
async def concat_media(body: FFMPEGConcatRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
medias = body.medias
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_concat_medias",
environment_name=config.modal_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(medias, sentry_trace, webhook=body.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/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 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)
@web_app.post("/ffmpeg/extract-audio",
tags=["发起任务"],
summary="发起音频提取任务",
description="提取视频文件的音频",
dependencies=[Depends(verify_token)]
)
async def extract_audio(body: FFMPEGExtractAudioRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
media = body.media
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_extract_audio",
environment_name=config.modal_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, sentry_trace, webhook=body.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/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 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)
@web_app.post("/ffmpeg/corner-mirror", tags=["发起任务"], summary="发起镜像小窗去重任务",
description="发起镜像小窗去重任务", dependencies=[Depends(verify_token)])
async def corner_mirror(body: FFMPEGCornerMirrorRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
media = body.media
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_corner_mirror",
environment_name=config.modal_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=media, sentry_trace=sentry_trace, webhook=body.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/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 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)
@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.modal_app_name, "ffmpeg_overlay_gif",
environment_name=config.modal_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, webhook=body.webhook)
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.modal_app_name, "ffmpeg_zoom_loop", environment_name=config.modal_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,
webhook=body.webhook)
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.modal_app_name, "ffmpeg_subtitle_apply",
environment_name=config.modal_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, fonts=body.fonts, sentry_trace=sentry_trace,
webhook=body.webhook)
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.modal_app_name, "ffmpeg_bgm_nosie_reduce",
environment_name=config.modal_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, webhook=body.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/ffmpeg/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 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)
web_app.include_router(ffmpeg.router)
web_app.include_router(cache.router)
@web_app.post("/comfyui/v1", tags=["ComfyUI"], summary="发起ComfyUIBase任务", description="发起ComfyUIBase任务",
@@ -678,7 +110,8 @@ async def comfyui_v1(item: ComfyTaskRequest, headers: Annotated[SentryTransactio
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, 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)
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)
@@ -693,7 +126,7 @@ 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 get_modal_task_status(task_id)
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
@@ -725,7 +158,8 @@ async def comfyui_v2(item: ComfyTaskRequest, headers: Annotated[SentryTransactio
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, 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)
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)
@@ -740,7 +174,7 @@ async def comfyui_v2(item: ComfyTaskRequest, headers: Annotated[SentryTransactio
dependencies=[Depends(verify_token)]
)
async def comfyui_v2_status(task_id: str, response: Response):
task_status, code, reason, result, transaction = await get_modal_task_status(task_id)
task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
media = None
if transaction:
@@ -756,5 +190,3 @@ async def comfyui_v2_status(task_id: str, response: Response):
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason,
result=media.urn if media else "")

0
src/BowongModalFunctions/middleware/__init__.py Normal file
View File

23
src/BowongModalFunctions/middleware/authorization.py Normal file
View File

@@ -0,0 +1,23 @@
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from starlette import status
bearer_scheme = HTTPBearer()
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme)):
token = credentials.credentials
# 在这里实现你的token验证逻辑
if not is_valid_token(token):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return token
def is_valid_token(token: str) -> bool:
# 这里实现具体的token验证逻辑
# 例如：验证JWT token，检查数据库中的token等
return token == "bowong7777"

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

@@ -39,7 +39,7 @@ class MediaSource(BaseModel):
bucket: Optional[str] = Field(description="媒体源所使用的存储桶(s3)/SubAppId(vod)")
urn: Optional[str] = Field(description="媒体源的唯一指定标识")
status: MediaCacheStatus = Field(default=MediaCacheStatus.unknown, description="媒体源在Modal集群挂载的缓存状态")
expired_at: Optional[datetime] = Field(description="缓存过期时间点", default=None)
expired_at: Optional[datetime] = Field(description="缓存过期时间点, 为None时不会过期（过期处理WIP）", default=None)
downloader_id: Optional[str] = Field(description="正在处理下载的Downloader ID", default=None)
progress: int = Field(description="缓存进度", default=0)
@@ -165,7 +165,7 @@ class MediaSources(BaseModel):
@classmethod
def parse_inputs(cls, v: Union[str, MediaSource]) -> List[MediaSource]:
if not v:
return v
raise ValidationError("inputs为空")
result = []
for item in v:
if isinstance(item, str):

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

@@ -56,7 +56,6 @@ class WebhookNotify(BaseModel):
examples=["Bearer 123456"], default=None)
class BaseFFMPEGTaskRequest(BaseModel):
webhook: Optional[WebhookNotify] = Field(description="Task webhook", default=None)
@@ -249,7 +248,7 @@ class FFMPEGSubtitleOverlayRequest(BaseFFMPEGTaskRequest):
@classmethod
def parse_inputs(cls, v: Union[str, MediaSource]) -> List[MediaSource]:
if not v:
return v
raise pydantic.ValidationError("fonts输入为空")
result = []
for item in v:
if isinstance(item, str):
@@ -309,6 +308,7 @@ class FFMPEGMixBgmWithNoiseReduceRequest(BaseFFMPEGTaskRequest):
class FFMPEGMixBgmWithNoiseReduceStatusResponse(BaseFFMPEGTaskStatusResponse):
result: Optional[str] = Field(default=None, description="生成结果的URN")
class ComfyTaskStatusResponse(BaseModel):
taskId: str = Field(description="任务Id")
status: TaskStatus = Field(description="任务运行状态")
@@ -316,8 +316,10 @@ class ComfyTaskStatusResponse(BaseModel):
code: Optional[int] = Field(description="任务错误原因代码", default=None)
result: Optional[str] = Field(description="任务运行结果", default=None)
class ComfyTaskRequest(BaseFFMPEGTaskRequest):
video_path: MediaSource = Field(default=MediaSource.from_str("s3://modal-media-cache/comfyui-input/markers/1016.mp4"), description="视频源")
video_path: MediaSource = Field(
default=MediaSource.from_str("s3://modal-media-cache/comfyui-input/markers/1016.mp4"), description="视频源")
start_time: str = Field(default="00:00:01.600", description="开始时间")
filename_prefix: str = Field(default=str(uuid.uuid4()), description="生成文件名前缀")
tts_text1: str = Field(default="好好看，这是我们家专门为女生定制的背心，", description="tts文本1")

0
src/BowongModalFunctions/router/__init__.py Normal file
View File

180
src/BowongModalFunctions/router/cache.py Normal file
View File

@@ -0,0 +1,180 @@
import asyncio
import modal
from modal import current_function_call_id
import sentry_sdk
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse
from starlette import status
from starlette.responses import RedirectResponse
from ..config import WorkerConfig
from ..middleware.authorization import verify_token
from ..models.media_model import MediaSources, CacheResult, MediaSource, MediaCacheStatus, DownloadResult
from ..models.web_model import SentryTransactionInfo
from ..utils.KVCache import KVCache
from ..utils.SentryUtils import SentryUtils
config = WorkerConfig()
router = APIRouter(prefix="/cache")
modal_kv_cache = KVCache(kv_name=config.modal_kv_name, environment=config.modal_environment)
@router.post("/",
tags=["缓存"],
summary="缓存视频文件",
description="异步缓存视频文件到S3存储桶和Modal Dict(KV)",
dependencies=[Depends(verify_token)])
async def cache(medias: MediaSources) -> CacheResult:
fn_id = current_function_call_id()
caches: MediaSources
sentry_trace = SentryTransactionInfo(x_trace_id=sentry_sdk.get_traceparent(),
x_baggage=sentry_sdk.get_baggage())
@SentryUtils.sentry_tracker(name="同步视频缓存", op="cache.get", fn_id=fn_id,
sentry_trace_id=None, sentry_baggage=None)
async def cache_handler(media: MediaSource):
cache_span = sentry_sdk.get_current_span()
cache_span.set_data("runner_id", fn_id)
cache_span.set_data("cache.key", [media.urn])
cached_media = modal_kv_cache.get_cache(media.urn)
cache_hit: bool = False
if not cached_media:
# start new download task
with cache_span.start_child(name="视频缓存任务入队",
op="queue.publish") as queue_publish_span:
fn = modal.Function.from_name(config.modal_app_name, 'cache_submit')
fn_task = fn.spawn(media, sentry_trace)
queue_publish_span.set_data("cache.key", media.urn)
queue_publish_span.set_data("messaging.message.id", fn_task.object_id)
queue_publish_span.set_data("messaging.destination.name",
"video-downloader.cache_submit")
queue_publish_span.set_data("messaging.message.body.size", 0)
# video_cache = MediaCache(status=MediaCacheStatus.downloading,
# downloader_id=fn_task.object_id)
media.status = MediaCacheStatus.downloading
media.downloader_id = fn_task.object_id
# video_cache_status_json = video_cache.model_dump_json()
modal_kv_cache.set_cache(media)
else:
media = cached_media
# video_cache = MediaCache.model_validate_json(video_cache_status_json)
match media.status:
case MediaCacheStatus.ready:
cache_hit = True
case MediaCacheStatus.downloading: # 下载任务已经在进行
cache_hit = True
case _:
# start new download task
with cache_span.start_child(name="视频缓存任务入队",
op="queue.publish") as queue_publish_span:
fn = modal.Function.from_name(config.modal_app_name, 'cache_submit')
fn_task = fn.spawn(media, sentry_trace)
queue_publish_span.set_data("cache.key", media.urn)
queue_publish_span.set_data("messaging.message.id", fn_task.object_id)
queue_publish_span.set_data("messaging.destination.name",
"video-downloader.cache_submit")
queue_publish_span.set_data("messaging.message.body.size", 0)
media.status = MediaCacheStatus.downloading
media.downloader_id = fn_task.object_id
# video_cache = MediaCache(status=MediaCacheStatus.downloading,
# downloader_id=fn_task.object_id)
# video_cache_status_json = video_cache.model_dump_json()
modal_kv_cache.set_cache(media)
cache_hit = False
# caches[media.urn] = video_cache
# logger.info(f"Media cache hit ? {cache_hit}")
cache_span.set_data("cache.hit", cache_hit)
return media
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(cache_handler(media)) for media in medias.inputs]
cache_task_result = [task.result() for task in tasks]
KVCache.batch_update_cloudflare_kv(cache_task_result)
return CacheResult(caches={media.urn: media for media in cache_task_result})
@router.post("/download",
tags=["缓存"],
summary="批量获取下载地址",
description="获取已缓存的视频下载地址",
dependencies=[Depends(verify_token)])
@sentry_sdk.trace
async def download_caches(medias: MediaSources) -> DownloadResult:
cdn_endpoint = config.S3_cdn_endpoint
urls = []
for media in medias.inputs:
urls.append(f"{cdn_endpoint}/{media.get_cdn_url()}")
return DownloadResult(urls=urls)
@router.get("/download",
tags=["缓存"],
summary="下载已缓存的视频",
description="通过CDN下载已缓存的视频文件")
@sentry_sdk.trace
async def download_cache(media: str) -> RedirectResponse:
cdn_endpoint = config.S3_cdn_endpoint
media = MediaSource.from_str(media)
return RedirectResponse(url=f"{cdn_endpoint}/{media.get_cdn_url()}", status_code=status.HTTP_302_FOUND)
@router.delete("/kv",
tags=["缓存"],
summary="清除KV记录",
description="清除当前环境下KV缓存过的所有数据(S3存储桶内的文件会保留)",
dependencies=[Depends(verify_token)])
async def purge_kv_all():
parent = sentry_sdk.get_current_span()
span = parent.start_child(name="清除缓存KV", op="purge.flush")
modal_kv_cache.clear()
span.set_data("cache.success", True)
span.finish()
return JSONResponse(content={"success": True})
@router.post("/kv",
tags=["缓存"],
summary="删除对应的KV记录",
description="删除请求中对应的视频缓存记录",
dependencies=[Depends(verify_token)])
async def purge_kv(medias: MediaSources):
try:
for media in medias.inputs:
modal_kv_cache.pop(media.urn)
keys = [media.urn for media in medias.inputs]
KVCache.batch_remove_cloudflare_kv(keys)
return JSONResponse(content={"success": True, "keys": keys})
except Exception as e:
return JSONResponse(content={"success": False, "error": str(e)})
@router.post("/media",
tags=["缓存"],
summary="清除指定的所有缓存",
description="清除指定的所有缓存(包括KV记录和S3存储文件)",
dependencies=[Depends(verify_token)])
async def purge_media(medias: MediaSources):
fn_id = current_function_call_id()
fn = modal.Function.from_name(config.modal_app_name, "cache_delete", environment_name=config.modal_environment)
@SentryUtils.sentry_tracker(name="清除媒体源缓存", op="cache.purge", fn_id=fn_id,
sentry_trace_id=None, sentry_baggage=None)
async def purge_handle(media: MediaSource):
cache_media = modal_kv_cache.pop(media.urn)
if cache_media:
deleted_cache: MediaSource = await fn.remote.aio(cache_media)
return deleted_cache.urn
return None
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(purge_handle(media)) for media in medias.inputs]
keys = [task.result() for task in tasks]
KVCache.batch_remove_cloudflare_kv(keys)
return JSONResponse(content={"success": True, "keys": keys})

319
src/BowongModalFunctions/router/ffmpeg.py Normal file
View File

@@ -0,0 +1,319 @@
from typing import Annotated
import modal
from fastapi import APIRouter, Depends, Header
from fastapi.responses import Response
from starlette import status
from ..config import WorkerConfig
from ..middleware.authorization import verify_token
from ..utils.ModalUtils import ModalUtils
from ..models.web_model import (FFMPEGSliceRequest, SentryTransactionHeader,
ModalTaskResponse, SentryTransactionInfo,
FFMPEGConcatRequest,
FFMPEGExtractAudioRequest,
FFMPEGCornerMirrorRequest,
FFMPEGOverlayGifRequest,
FFMPEGZoomLoopRequest,
FFMPEGSubtitleOverlayRequest,
FFMPEGMixBgmWithNoiseReduceRequest,
FFMPEGSliceTaskStatusResponse,
FFMPEGConcatTaskStatusResponse,
FFMPEGExtractAudioTaskStatusResponse,
FFMPEGCornerMirrorTaskStatusResponse,
FFMPEGOverlayGifTaskStatusResponse,
FFMPEGSubtitleTaskStatusResponse,
FFMPEGZoomLoopTaskStatusResponse,
FFMPEGMixBgmWithNoiseReduceStatusResponse)
config = WorkerConfig()
router = APIRouter(prefix="/ffmpeg")
sentry_header_schema = {
"x-trace-id": {
"description": "Sentry Transaction ID",
"schema": {
"type": "string",
}
},
"x-baggage": {
"description": "Sentry Transaction baggage",
"schema": {
"type": "string",
}
}
}
@router.post("/slice",
tags=["发起任务"],
summary="发起切割任务",
description="依据打点信息切出多个片段",
dependencies=[Depends(verify_token)])
async def slice_media(request: FFMPEGSliceRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_slice_media",
environment_name=config.modal_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=request.media, markers=request.markers, sentry_trace=sentry_trace, webhook=request.webhook)
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("/concat",
tags=["发起任务"],
summary="发起合并任务",
description="依据AI分析的结果发起合并任务",
dependencies=[Depends(verify_token)])
async def concat_media(body: FFMPEGConcatRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
medias = body.medias
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_concat_medias",
environment_name=config.modal_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(medias, sentry_trace, webhook=body.webhook)
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)]
)
async def extract_audio(body: FFMPEGExtractAudioRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
media = body.media
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_extract_audio",
environment_name=config.modal_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, sentry_trace, webhook=body.webhook)
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)])
async def corner_mirror(body: FFMPEGCornerMirrorRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
media = body.media
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_corner_mirror",
environment_name=config.modal_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=media, sentry_trace=sentry_trace, webhook=body.webhook)
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)])
async def overlay_gif(body: FFMPEGOverlayGifRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_overlay_gif",
environment_name=config.modal_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, webhook=body.webhook)
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)])
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)
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,
webhook=body.webhook)
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)])
async def subtitle_apply(body: FFMPEGSubtitleOverlayRequest,
headers: Annotated[SentryTransactionHeader, Header()]) -> ModalTaskResponse:
fn = modal.Function.from_name(config.modal_app_name, "ffmpeg_subtitle_apply",
environment_name=config.modal_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, fonts=body.fonts, sentry_trace=sentry_trace,
webhook=body.webhook)
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)])
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",
environment_name=config.modal_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, webhook=body.webhook)
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)

56
src/BowongModalFunctions/utils/KVCache.py
View File

@@ -1,8 +1,13 @@
from typing import Optional
import os
from typing import Optional, List
import httpx
import modal
from loguru import logger
from ..models.media_model import MediaSource
cf_account_id = os.environ.get("CF_ACCOUNT_ID")
cf_kv_api_token = os.environ.get("CF_KV_API_TOKEN")
cf_kv_namespace_id = os.environ.get("CF_KV_NAMESPACE_ID")
class KVCache:
@@ -31,3 +36,50 @@ class KVCache:
raise KeyError("URN错误，资源不存在")
return None
return MediaSource.model_validate_json(cache_json)
@staticmethod
def batch_update_cloudflare_kv(caches: List[MediaSource]):
with httpx.Client() as client:
try:
response = client.put(
f"https://api.cloudflare.com/client/v4/accounts/{cf_account_id}/storage/kv/namespaces/{cf_kv_namespace_id}/bulk",
headers={"Authorization": f"Bearer {cf_kv_api_token}"},
json=[
{
"based64": False,
"key": cache.urn,
"value": cache.model_dump_json(),
}
for cache in caches
]
)
response.raise_for_status()
except httpx.RequestError as e:
logger.error(f"An error occurred while put kv to cloudflare")
raise e
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
raise e
except Exception as e:
logger.error(f"An unexpected error occurred: {str(e)}")
raise e
@staticmethod
def batch_remove_cloudflare_kv(keys: List[str]):
with httpx.Client() as client:
try:
response = client.post(
f"https://api.cloudflare.com/client/v4/accounts/{cf_account_id}/storage/kv/namespaces/{cf_kv_namespace_id}/bulk/delete",
headers={"Authorization": f"Bearer {cf_kv_api_token}"},
json=keys
)
response.raise_for_status()
except httpx.RequestError as e:
logger.error(f"An error occurred while put kv to cloudflare")
raise e
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error occurred while get kv from cloudflare {str(e)}")
raise e
except Exception as e:
logger.error(f"An unexpected error occurred: {str(e)}")
raise e

56
src/BowongModalFunctions/utils/ModalUtils.py Normal file
View File

@@ -0,0 +1,56 @@
from typing import Tuple, Optional, Any
import modal
from loguru import logger
from modal.call_graph import InputStatus
from BowongModalFunctions.models.web_model import TaskStatus, SentryTransactionInfo, ErrorCode
class ModalUtils:
@staticmethod
async def get_modal_task_status(task_id: str) -> Tuple[
TaskStatus, Optional[int], Optional[str], Optional[Any], Optional[SentryTransactionInfo]]:
"""
使用modal任务id查看任务运行状态和结果
:param task_id: modal 任务id
:return: (TaskStaus 任务运行状态, errorCode 错误代码, errorReason 错误原因, results 任务结果, sentryTransactionInfo Sentry跟踪信息)
"""
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
task = root.children[0]
if not task.function_call_id == task_id:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None
logger.info(task)
logger.info(f"Task function call status: {task.status}")
match task.status:
case InputStatus.PENDING:
return TaskStatus.running, None, None, None, None
case InputStatus.SUCCESS:
try:
results, sentry_trace = fn_task.get(timeout=2)
return TaskStatus.success, ErrorCode.SUCCESS.value, None, results, sentry_trace
except modal.exception.OutputExpiredError:
return TaskStatus.expired, ErrorCode.NOT_FOUND.value, "EXPIRED", None, None
except TimeoutError:
return TaskStatus.running, None, None, None, None
except Exception as e:
logger.exception(e)
return TaskStatus.failed, ErrorCode.SYSTEM_ERROR.value, 'FAILURE', None, None
case _:
error = 'FAILURE'
match task.status:
case InputStatus.INIT_FAILURE:
error = 'INIT_FAILURE'
case InputStatus.TERMINATED:
error = 'TERMINATED'
case InputStatus.TIMEOUT:
error = 'TIMEOUT'
return TaskStatus.failed, ErrorCode.SYSTEM_ERROR.value, error, None, None
except Exception as e:
logger.exception(e)
return TaskStatus.failed, ErrorCode.NOT_FOUND.value, "NOT_FOUND", None, None

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

@@ -1,5 +1,4 @@
import asyncio
from typing import Optional, Tuple, Any, Coroutine
from typing import Optional, Tuple, Any
import httpx
import psutil
import sentry_sdk
@@ -17,7 +16,7 @@ class SentryUtils:
cpu_count = psutil.cpu_count()
mem = psutil.virtual_memory()
return (cpu_freq, cpu_count, mem)
return cpu_freq, cpu_count, mem
@staticmethod
def sentry_tracker(sentry_trace_id: Optional[str], sentry_baggage: Optional[str],

1
src/BowongModalFunctions/utils/VideoUtils.py
View File

@@ -2,7 +2,6 @@ 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

4
src/cluster/app.py
View File

@@ -16,5 +16,5 @@ app = modal.App(config.modal_app_name,
app.include(media_app)
app.include(ffmpeg_app)
app.include(web_app)
app.include(comfyui_v1_app)
app.include(comfyui_v2_app)
# app.include(comfyui_v1_app)
# app.include(comfyui_v2_app)

17
src/cluster/ffmpeg_app.py
View File

@@ -5,7 +5,7 @@ ffmpeg_worker_image = (
modal.Image.debian_slim(python_version="3.11")
.apt_install('ffmpeg')
.pip_install_from_pyproject("../pyproject.toml")
.env(dotenv_values("../.runtime.env"))
.env(dotenv_values("../../.runtime.env"))
.add_local_python_source('cluster')
.add_local_python_source('BowongModalFunctions')
)
@@ -16,8 +16,8 @@ app = modal.App(
include_source=False)
with ffmpeg_worker_image.imports():
import shutil, psutil, os, backoff, sentry_sdk
from typing import List, Optional, Tuple, Any
import shutil, os, backoff, sentry_sdk
from typing import List, Optional, Tuple
from loguru import logger
from modal import current_function_call_id
from ffmpeg.asyncio import FFmpeg
@@ -44,15 +44,6 @@ with ffmpeg_worker_image.imports():
)
@sentry_sdk.trace
def capture_hardware_info() -> Tuple[Any, int, Any]:
cpu_freq = psutil.cpu_freq()
cpu_count = psutil.cpu_count()
mem = psutil.virtual_memory()
return (cpu_freq, cpu_count, mem)
def ffmpeg_init(use_ffprobe: bool = False) -> FFmpeg:
if use_ffprobe:
ffmpeg_cmd = FFmpeg('ffprobe')
@@ -455,7 +446,7 @@ with ffmpeg_worker_image.imports():
output_path=output_path)
s3_outputs = local_copy_to_s3([local_output_filepath])
return s3_outputs
return s3_outputs[0]
result = await ffmpeg_process(video=media, bgm=bgm, video_volume=video_volume,
music_volume=music_volume, noise_sample=noise_sample)

2
src/cluster/video.py
View File

@@ -136,7 +136,7 @@ with downloader_image.imports():
file_extension = media_info.Type
cache_dir = f"/{config.S3_mount_dir}/{media.protocol.value}/{media.endpoint}/{media.bucket}"
cache_file = media.path if '.' in media.path else f"{media.path}.{file_extension}"
return (cache_dir, cache_file, media_info.MediaUrl)
return cache_dir, cache_file, media_info.MediaUrl
else:
raise FileNotFoundError(
f"FileId : {media.path} not found in SubAppId: {media.bucket} at {media.endpoint}")

16
uv.lock generated
View File

@@ -197,7 +197,9 @@ dependencies = [
[package.optional-dependencies]
cli = [
{ name = "gunicorn" },
{ name = "twine" },
{ name = "uvicorn", extra = ["standard"] },
]
[package.metadata]
@@ -207,6 +209,7 @@ requires-dist = [
{ name = "cos-python-sdk-v5", specifier = ">=1.9.36" },
{ name = "crcmod", specifier = ">=1.7" },
{ name = "fastapi", extras = ["standard"], specifier = ">=0.115.12" },
{ name = "gunicorn", marker = "extra == 'cli'", specifier = ">=23.0.0" },
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "loguru", specifier = ">=0.7.3" },
{ name = "modal", specifier = ">=0.76.3" },
@@ -225,6 +228,7 @@ requires-dist = [
{ name = "tencentcloud-sdk-python-common", specifier = ">=3.0.1363" },
{ name = "tencentcloud-sdk-python-vod", specifier = ">=3.0.1363" },
{ name = "twine", marker = "extra == 'cli'", specifier = ">=6.1.0" },
{ name = "uvicorn", extras = ["standard"], marker = "extra == 'cli'", specifier = ">=0.34.2" },
{ name = "webvtt-py", specifier = ">=0.5.1" },
]
provides-extras = ["cli"]
@@ -670,6 +674,18 @@ dependencies = [
]
sdist = { url = "https://files.pythonhosted.org/packages/79/b9/55936e462a5925190d7427e880b3033601d1effd13809b483d13a926061a/grpclib-0.4.7.tar.gz", hash = "sha256:2988ef57c02b22b7a2e8e961792c41ccf97efc2ace91ae7a5b0de03c363823c3", size = 61254 }
[[package]]
name = "gunicorn"
version = "23.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "packaging" },
]
sdist = { url = "https://files.pythonhosted.org/packages/34/72/9614c465dc206155d93eff0ca20d42e1e35afc533971379482de953521a4/gunicorn-23.0.0.tar.gz", hash = "sha256:f014447a0101dc57e294f6c18ca6b40227a4c90e9bdb586042628030cba004ec", size = 375031 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/7d/6dac2a6e1eba33ee43f318edbed4ff29151a49b5d37f080aad1e6469bca4/gunicorn-23.0.0-py3-none-any.whl", hash = "sha256:ec400d38950de4dfd418cff8328b2c8faed0edb0d517d3394e457c317908ca4d", size = 85029 },
]
[[package]]
name = "h11"
version = "0.16.0"