From 668a5a33d4bc1dacd0bdde022bc73601e82cead0 Mon Sep 17 00:00:00 2001 From: "shuohigh@gmail.com" Date: Mon, 19 May 2025 15:02:34 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8C=89=E5=8A=9F=E8=83=BD=E6=A8=A1=E5=9D=97?= =?UTF-8?q?=E6=8B=86=E5=88=86API=E8=B7=AF=E7=94=B1=E5=88=B0=E4=B8=8D?= =?UTF-8?q?=E5=90=8C=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .runtime.env | 2 +- README.md | 6 + pyproject.toml | 6 +- src/BowongModalFunctions/__init__.py | 2 +- src/BowongModalFunctions/api.py | 616 +----------------- .../middleware/__init__.py | 0 .../middleware/authorization.py | 23 + .../models/media_model.py | 4 +- src/BowongModalFunctions/models/web_model.py | 22 +- src/BowongModalFunctions/router/__init__.py | 0 src/BowongModalFunctions/router/cache.py | 180 +++++ src/BowongModalFunctions/router/ffmpeg.py | 319 +++++++++ src/BowongModalFunctions/utils/KVCache.py | 56 +- src/BowongModalFunctions/utils/ModalUtils.py | 56 ++ src/BowongModalFunctions/utils/SentryUtils.py | 5 +- src/BowongModalFunctions/utils/VideoUtils.py | 1 - src/cluster/app.py | 4 +- src/cluster/ffmpeg_app.py | 17 +- src/cluster/video.py | 2 +- uv.lock | 16 + 20 files changed, 708 insertions(+), 629 deletions(-) create mode 100644 src/BowongModalFunctions/middleware/__init__.py create mode 100644 src/BowongModalFunctions/middleware/authorization.py create mode 100644 src/BowongModalFunctions/router/__init__.py create mode 100644 src/BowongModalFunctions/router/cache.py create mode 100644 src/BowongModalFunctions/router/ffmpeg.py create mode 100644 src/BowongModalFunctions/utils/ModalUtils.py diff --git a/.runtime.env b/.runtime.env index 5229e3c..dbc5e5b 100644 --- a/.runtime.env +++ b/.runtime.env @@ -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 diff --git a/README.md b/README.md index c6c77b1..14e7d2b 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,12 @@ modal deploy -m cluster.app ``` +## 本地验证fastapi部署 + +```bash +uvicorn BowongModalFunctions.api:web_app --host 0.0.0.0 --port 8080 +``` + # Pip ## 构建 diff --git a/pyproject.toml b/pyproject.toml index 4a0e084..63c431a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"] diff --git a/src/BowongModalFunctions/__init__.py b/src/BowongModalFunctions/__init__.py index e5c39f8..e14ac0f 100644 --- a/src/BowongModalFunctions/__init__.py +++ b/src/BowongModalFunctions/__init__.py @@ -1 +1 @@ -__version__ = "0.1.1-pre3" \ No newline at end of file +__version__: str = "0.1.2" \ No newline at end of file diff --git a/src/BowongModalFunctions/api.py b/src/BowongModalFunctions/api.py index fb3a9fc..ad79a64 100644 --- a/src/BowongModalFunctions/api.py +++ b/src/BowongModalFunctions/api.py @@ -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 "") - - diff --git a/src/BowongModalFunctions/middleware/__init__.py b/src/BowongModalFunctions/middleware/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/middleware/authorization.py b/src/BowongModalFunctions/middleware/authorization.py new file mode 100644 index 0000000..476e070 --- /dev/null +++ b/src/BowongModalFunctions/middleware/authorization.py @@ -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" diff --git a/src/BowongModalFunctions/models/media_model.py b/src/BowongModalFunctions/models/media_model.py index f71b532..7122f47 100644 --- a/src/BowongModalFunctions/models/media_model.py +++ b/src/BowongModalFunctions/models/media_model.py @@ -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): diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py index 2d27301..f18c3c8 100644 --- a/src/BowongModalFunctions/models/web_model.py +++ b/src/BowongModalFunctions/models/web_model.py @@ -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,16 +316,18 @@ 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") - tts_text2:str = Field(default="", description="tts文本2") - tts_text3:str = Field(default="", description="tts文本3") - tts_text4:str = Field(default="", description="tts文本4") + filename_prefix: str = Field(default=str(uuid.uuid4()), description="生成文件名前缀") + tts_text1: str = Field(default="好好看，这是我们家专门为女生定制的背心，", description="tts文本1") + tts_text2: str = Field(default="", description="tts文本2") + tts_text3: str = Field(default="", description="tts文本3") + tts_text4: str = Field(default="", description="tts文本4") anchor_id: str = Field(default="dawan", description="讲话人ID") - speed:float = Field(default=1, description="讲话语速") + speed: float = Field(default=1, description="讲话语速") @field_validator('video_path', mode='before') @classmethod @@ -335,4 +337,4 @@ class ComfyTaskRequest(BaseFFMPEGTaskRequest): elif isinstance(v, MediaSource): return v else: - raise TypeError(v) \ No newline at end of file + raise TypeError(v) diff --git a/src/BowongModalFunctions/router/__init__.py b/src/BowongModalFunctions/router/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/BowongModalFunctions/router/cache.py b/src/BowongModalFunctions/router/cache.py new file mode 100644 index 0000000..135e457 --- /dev/null +++ b/src/BowongModalFunctions/router/cache.py @@ -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}) diff --git a/src/BowongModalFunctions/router/ffmpeg.py b/src/BowongModalFunctions/router/ffmpeg.py new file mode 100644 index 0000000..da84f40 --- /dev/null +++ b/src/BowongModalFunctions/router/ffmpeg.py @@ -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) diff --git a/src/BowongModalFunctions/utils/KVCache.py b/src/BowongModalFunctions/utils/KVCache.py index aec9754..3fb0dfa 100644 --- a/src/BowongModalFunctions/utils/KVCache.py +++ b/src/BowongModalFunctions/utils/KVCache.py @@ -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 diff --git a/src/BowongModalFunctions/utils/ModalUtils.py b/src/BowongModalFunctions/utils/ModalUtils.py new file mode 100644 index 0000000..6ea7dbc --- /dev/null +++ b/src/BowongModalFunctions/utils/ModalUtils.py @@ -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 diff --git a/src/BowongModalFunctions/utils/SentryUtils.py b/src/BowongModalFunctions/utils/SentryUtils.py index f47cb3b..599bc8d 100644 --- a/src/BowongModalFunctions/utils/SentryUtils.py +++ b/src/BowongModalFunctions/utils/SentryUtils.py @@ -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], diff --git a/src/BowongModalFunctions/utils/VideoUtils.py b/src/BowongModalFunctions/utils/VideoUtils.py index 5c2de6a..42ff966 100644 --- a/src/BowongModalFunctions/utils/VideoUtils.py +++ b/src/BowongModalFunctions/utils/VideoUtils.py @@ -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 diff --git a/src/cluster/app.py b/src/cluster/app.py index 344e116..eaa291f 100644 --- a/src/cluster/app.py +++ b/src/cluster/app.py @@ -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) diff --git a/src/cluster/ffmpeg_app.py b/src/cluster/ffmpeg_app.py index c9dfa9b..e9cfd9e 100644 --- a/src/cluster/ffmpeg_app.py +++ b/src/cluster/ffmpeg_app.py @@ -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) diff --git a/src/cluster/video.py b/src/cluster/video.py index 9f4bc6a..2f33ce6 100644 --- a/src/cluster/video.py +++ b/src/cluster/video.py @@ -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}") diff --git a/uv.lock b/uv.lock index 16d32d4..d893b3c 100644 --- a/uv.lock +++ b/uv.lock @@ -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"