From 903a1fcc0a4d01ec9640aa95040f6b43df4f56e9 Mon Sep 17 00:00:00 2001 From: "kyj@bowong.ai" Date: Fri, 16 May 2025 17:19:26 +0800 Subject: [PATCH] =?UTF-8?q?PERF=20=E9=80=82=E9=85=8D=E6=96=B0=E7=BB=93?= =?UTF-8?q?=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 3 +- .idea/misc.xml | 2 +- .idea/modalDeploy.iml | 2 +- .runtime.env | 4 +- pyproject_comfyui.toml | 28 + src/BowongModalFunctions/api.py | 124 +++- src/BowongModalFunctions/config.py | 3 +- src/BowongModalFunctions/models/web_model.py | 29 + .../template/template_v1.json | 272 +++++++++ .../template/template_v2.json | 268 +++++++++ src/cluster/app.py | 4 + src/cluster/comfyui_v1.py | 549 ++++++++++++++++++ src/cluster/comfyui_v2.py | 538 +++++++++++++++++ src/cluster/ffmpeg_app.py | 10 +- src/cluster/video.py | 7 +- src/cluster/web.py | 3 +- 16 files changed, 1822 insertions(+), 24 deletions(-) create mode 100644 pyproject_comfyui.toml create mode 100644 src/BowongModalFunctions/template/template_v1.json create mode 100644 src/BowongModalFunctions/template/template_v2.json create mode 100644 src/cluster/comfyui_v1.py create mode 100644 src/cluster/comfyui_v2.py diff --git a/.gitignore b/.gitignore index d7949a5..a23fe58 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ .pypirc -.env \ No newline at end of file +.env +.idea \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml index 0fa8607..b657481 100644 --- a/.idea/misc.xml +++ b/.idea/misc.xml @@ -3,5 +3,5 @@ - + \ No newline at end of file diff --git a/.idea/modalDeploy.iml b/.idea/modalDeploy.iml index 62d1b49..b118175 100644 --- a/.idea/modalDeploy.iml +++ b/.idea/modalDeploy.iml @@ -5,7 +5,7 @@ - + \ No newline at end of file diff --git a/.runtime.env b/.runtime.env index 6595d0c..5229e3c 100644 --- a/.runtime.env +++ b/.runtime.env @@ -1,5 +1,5 @@ -MODAL_ENVIRONMENT=test -modal_app_name=bowong-ai-video +MODAL_ENVIRONMENT=dev +modal_app_name=cluster-test S3_mount_dir=/mntS3 S3_bucket_name=modal-media-cache S3_region=ap-northeast-2 diff --git a/pyproject_comfyui.toml b/pyproject_comfyui.toml new file mode 100644 index 0000000..5bec768 --- /dev/null +++ b/pyproject_comfyui.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["setuptools"] +build-backend = "setuptools.build_meta" + +[project] +name = "modal_comfyui" +version = "1.0.0" +requires-python = ">=3.10" +dependencies = [ + "fastapi[standard]==0.115.4", + "cos-python-sdk-v5", + "sqlalchemy", + "ultralytics", + "tencentcloud-sdk-python", + "pymysql", + "Pillow", + "ffmpy", + "opencv-python", + "av", + "imageio", + "loguru", + "openai-whisper", + "sentry-sdk", + "pydantic", + "pydantic_settings", + "conformer==0.3.2", + "einops>0.6.1" +] \ No newline at end of file diff --git a/src/BowongModalFunctions/api.py b/src/BowongModalFunctions/api.py index 8cd8ddd..7ff86e5 100644 --- a/src/BowongModalFunctions/api.py +++ b/src/BowongModalFunctions/api.py @@ -6,7 +6,7 @@ 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 fastapi import FastAPI, Response, Depends, Header +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 @@ -22,7 +22,8 @@ from .models.media_model import (MediaSource, MediaCacheStatus, CacheResult, DownloadResult) -from .models.web_model import (TaskStatus, ErrorCode, +from .models.web_model import (TaskStatus, + ErrorCode, SentryTransactionInfo, SentryTransactionHeader, ModalTaskResponse, @@ -41,7 +42,9 @@ from .models.web_model import (TaskStatus, ErrorCode, FFMPEGOverlayGifTaskStatusResponse, FFMPEGSubtitleTaskStatusResponse, FFMPEGZoomLoopTaskStatusResponse, - FFMPEGMixBgmWithNoiseReduceStatusResponse + FFMPEGMixBgmWithNoiseReduceStatusResponse, + ComfyTaskRequest, + ComfyTaskStatusResponse ) from .config import WorkerConfig @@ -52,13 +55,13 @@ config = WorkerConfig() web_app = FastAPI(title="Modal worker API", summary="Modal Worker的API, 包括缓存视频, 发起生产任务等", servers=[ - {'url': 'https://bowongai-dev--bowong-ai-video-fastapi-webapp.modal.run', + {'url': f'https://bowongai-dev--{config.modal_app_name}-fastapi-webapp.modal.run', 'description': 'modal 开发环境服务'}, {'url': 'https://modal-dev.bowong.cc', 'description': 'modal 开发环境服务独立域名'}, - {'url': 'https://bowongai-test--bowong-ai-video-fastapi-webapp.modal.run', + {'url': f'https://bowongai-test--{config.modal_app_name}-fastapi-webapp.modal.run', 'description': 'modal 测试环境服务'}, - {'url': 'https://bowongai-main--bowong-ai-video-fastapi-webapp.modal.run', + {'url': f'https://bowongai-main--{config.modal_app_name}-fastapi-webapp.modal.run', 'description': 'modal 生产环境服务' } ]) @@ -113,6 +116,18 @@ sentry_header_schema = { } } +ALIAS_MAP = { + "/comfyui": "/comfyui/v2", +} + + +@web_app.middleware("http") +@web_app.middleware("https") +async def alias_middleware(request: Request, call_next): + if request.url.path in ALIAS_MAP.keys(): + request.scope["path"] = ALIAS_MAP[request.url.path] + return await call_next(request) + @sentry_sdk.trace def batch_update_cloudflare_kv(caches: List[MediaSource]): @@ -644,3 +659,100 @@ async def bgm_nosie_reduce_status(task_id: str, response: Response) -> FFMPEGMix response.headers["x-baggage"] = transaction.x_baggage return FFMPEGMixBgmWithNoiseReduceStatusResponse(taskId=task_id, status=task_status, code=code, error=reason, result=result) + + +@web_app.post("/comfyui/v1", tags=["ComfyUI"], summary="发起ComfyUIBase任务", description="发起ComfyUIBase任务", + responses={ + status.HTTP_200_OK: { + "description": "", + "headers": sentry_header_schema + }, + }, + dependencies=[Depends(verify_token)] + ) +async def comfyui_v1(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]): + sentry_trace = None + if headers.x_trace_id and headers.x_baggage: + sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage) + cls = modal.Cls.from_name(config.modal_app_name, "ComfyUI", environment_name=config.modal_environment) + fn_call = cls().api.spawn(item.video_path.path, 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) + return ModalTaskResponse(success=True, taskId=fn_call.object_id) + + +@web_app.get("/comfyui/v1/{task_id}", tags=["ComfyUI"], summary="查询ComfyUIBase任务", + description="查询ComfyUIBase任务", + responses={ + status.HTTP_200_OK: { + "description": "", + "headers": sentry_header_schema + }, + }, + dependencies=[Depends(verify_token)] + ) +async def comfyui_v1_status(task_id: str, response: Response): + task_status, code, reason, result, transaction = await get_modal_task_status(task_id) + media = None + if transaction: + response.headers["x-trace-id"] = transaction.x_trace_id + response.headers["x-baggage"] = transaction.x_baggage + if task_status == "success": + task_status = result["status"] + if task_status != "success": + reason = result["msg"] + else: + media = MediaSource.from_str("s3://" + "/".join( + [config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, result["file_name"]])) + return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code, + error=reason, result=media.urn if media else None) + + +@web_app.post("/comfyui/v2", tags=["ComfyUI"], summary="发起ComfyUILatentSync1.5任务", + description="发起ComfyUILatentSync1.5任务", + responses={ + status.HTTP_200_OK: { + "description": "", + "headers": sentry_header_schema + }, + }, + dependencies=[Depends(verify_token)] + ) +async def comfyui_v2(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]): + sentry_trace = None + if headers.x_trace_id and headers.x_baggage: + sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage) + cls = modal.Cls.from_name(config.modal_app_name, "ComfyUILatentSync15", environment_name=config.modal_environment) + fn_call = cls().api.spawn(item.video_path.path, 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) + return ModalTaskResponse(success=True, taskId=fn_call.object_id) + + +@web_app.get("/comfyui/v2/{task_id}", tags=["ComfyUI"], summary="查询ComfyUILatentSync1.5任务", + description="查询ComfyUILatentSync1.5任务", + responses={ + status.HTTP_200_OK: { + "description": "", + "headers": sentry_header_schema + }, + }, + dependencies=[Depends(verify_token)] + ) +async def comfyui_v2_status(task_id: str, response: Response): + task_status, code, reason, result, transaction = await get_modal_task_status(task_id) + media = None + + if transaction: + response.headers["x-trace-id"] = transaction.x_trace_id + response.headers["x-baggage"] = transaction.x_baggage + if task_status == "success": + task_status = result["status"] + if task_status != "success": + reason = result["msg"] + else: + media = MediaSource.from_str("s3://" + "/".join( + [config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, result["file_name"]])) + return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code, + error=reason, + result=media.urn if media else "") + + diff --git a/src/BowongModalFunctions/config.py b/src/BowongModalFunctions/config.py index 32383e9..f02e205 100644 --- a/src/BowongModalFunctions/config.py +++ b/src/BowongModalFunctions/config.py @@ -2,7 +2,6 @@ from typing import Optional, Any from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict - class WorkerConfig(BaseSettings): video_downloader_concurrency: int = Field(default=10, description="处理缓存任务的并行数") ffmpeg_worker_concurrency: int = Field(default=10, description="处理视频合成任务的并行数") @@ -17,5 +16,7 @@ class WorkerConfig(BaseSettings): modal_kv_name: str = Field(default='media-cache', description="Modal视频缓存KV库") modal_environment: str = Field(default="dev", description="Modal worker运行环境") modal_app_name: str = Field(default='bowong-ai-video', description="Modal App集群名称") + comfyui_s3_input: Optional[str] = Field(default="comfyui-input", description="ComfyUI input S3文件夹名") + comfyui_s3_output: Optional[str] = Field(default="comfyui-output", description="ComfyUI output S3文件夹名") modal_config: Any = SettingsConfigDict() diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py index 661455b..807a3d5 100644 --- a/src/BowongModalFunctions/models/web_model.py +++ b/src/BowongModalFunctions/models/web_model.py @@ -1,3 +1,4 @@ +import uuid from enum import Enum from typing import List, Union, Optional @@ -307,3 +308,31 @@ 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="任务运行状态") + error: Optional[str] = Field(description="任务错误原因", default=None) + code: Optional[int] = Field(description="任务错误原因代码", default=None) + result: Optional[str] = Field(description="任务运行结果", default=None) + +class ComfyTaskRequest(BaseModel): + 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") + anchor_id: str = Field(default="dawan", description="讲话人ID") + speed:float = Field(default=1, description="讲话语速") + + @field_validator('video_path', mode='before') + @classmethod + def parse_inputs(cls, v: Union[str, MediaSource]): + if isinstance(v, str): + return MediaSource.from_str(v) + elif isinstance(v, MediaSource): + return v + else: + raise TypeError(v) \ No newline at end of file diff --git a/src/BowongModalFunctions/template/template_v1.json b/src/BowongModalFunctions/template/template_v1.json new file mode 100644 index 0000000..3fa586c --- /dev/null +++ b/src/BowongModalFunctions/template/template_v1.json @@ -0,0 +1,272 @@ +{ + "211": { + "inputs": { + "input_id": "tts_text1", + "default_value": "整个版型前短后长的设计，总长度是六十八点五，小个子都可以驾驭。", + "display_name": "文本1", + "description": "文本1" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "217": { + "inputs": {}, + "class_type": "BW_RandomSeed", + "_meta": { + "title": "不忘科技-随机种子-🐱" + } + }, + "225": { + "inputs": { + "frame_rate": [ + "426", + 0 + ], + "loop_count": 0, + "filename_prefix": "ed0fc06a-0128-4156-bee3-a859f9623258", + "format": "video/h264-mp4", + "pix_fmt": "yuv420p", + "crf": 19, + "save_metadata": false, + "trim_to_audio": false, + "pingpong": false, + "save_output": true, + "images": [ + "516", + 0 + ], + "audio": [ + "515", + 0 + ] + }, + "class_type": "VHS_VideoCombine", + "_meta": { + "title": "Video Combine 🎥🅥🅗🅢" + } + }, + "330": { + "inputs": { + "input_id": "start_time", + "default_value": "00:01:13.600", + "display_name": "", + "description": "00:01:01" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "331": { + "inputs": { + "input_id": "tts_text3", + "default_value": "", + "display_name": "文本3", + "description": "文本3" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "399": { + "inputs": { + "input_id": "tts_text2", + "default_value": "", + "display_name": "文本2", + "description": "文本2" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "406": { + "inputs": { + "input_id": "tts_text4", + "default_value": "", + "display_name": "文本4", + "description": "文本4" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "422": { + "inputs": { + "start_time": [ + "330", + 0 + ], + "end_padding": 0.4, + "fps": [ + "426", + 0 + ], + "audio": [ + "515", + 0 + ] + }, + "class_type": "VideoPointCompute", + "_meta": { + "title": "视频帧位计算" + } + }, + "424": { + "inputs": { + "input_id": "video_path", + "default_value": "/root/comfy/ComfyUI/input_s3/markers/30198.mp4", + "display_name": "", + "description": "" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "426": { + "inputs": { + "Number": "25" + }, + "class_type": "Int", + "_meta": { + "title": "帧率" + } + }, + "496": { + "inputs": { + "delimiter": "\\n", + "clean_whitespace": "true", + "text_a": [ + "211", + 0 + ], + "text_b": [ + "399", + 0 + ], + "text_c": [ + "331", + 0 + ], + "text_d": [ + "406", + 0 + ] + }, + "class_type": "Text Concatenate", + "_meta": { + "title": "Text Concatenate" + } + }, + "498": { + "inputs": { + "video_path": [ + "424", + 0 + ], + "start_point": [ + "422", + 0 + ], + "duration": [ + "422", + 1 + ], + "fps": [ + "426", + 0 + ], + "force_match_fps": true + }, + "class_type": "VideoCutByFramePoint", + "_meta": { + "title": "视频剪裁(精确帧位)" + } + }, + "511": { + "inputs": { + "video_path": [ + "512", + 0 + ], + "fps": 30 + }, + "class_type": "VideoChangeFPS", + "_meta": { + "title": "视频转换帧率" + } + }, + "512": { + "inputs": { + "index": -1, + "filenames": [ + "225", + 0 + ] + }, + "class_type": "VHS_SelectFilename", + "_meta": { + "title": "Select Filename 🎥🅥🅗🅢" + } + }, + "515": { + "inputs": { + "text_content": [ + "496", + 0 + ], + "anchor_id": "dawan", + "speed": 1, + "auto_download": false + }, + "class_type": "CosyVoiceTTS", + "_meta": { + "title": "CosyVoice文本转语音" + } + }, + "516": { + "inputs": { + "seed": [ + "217", + 0 + ], + "out_video_fps": 25, + "out_video_crf": 2, + "save": false, + "images": [ + "517", + 0 + ], + "audio": [ + "517", + 1 + ] + }, + "class_type": "D_LatentSyncNode", + "_meta": { + "title": "LatentSync Node" + } + }, + "517": { + "inputs": { + "mode": "normal", + "images": [ + "498", + 0 + ], + "audio": [ + "515", + 0 + ] + }, + "class_type": "D_VideoLengthAdjuster", + "_meta": { + "title": "Video Length Adjuster" + } + } +} \ No newline at end of file diff --git a/src/BowongModalFunctions/template/template_v2.json b/src/BowongModalFunctions/template/template_v2.json new file mode 100644 index 0000000..9315d79 --- /dev/null +++ b/src/BowongModalFunctions/template/template_v2.json @@ -0,0 +1,268 @@ +{ + "211": { + "inputs": { + "input_id": "tts_text1", + "default_value": "前面的领子设计很优雅，", + "display_name": "文本1", + "description": "文本1" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "217": { + "inputs": {}, + "class_type": "BW_RandomSeed", + "_meta": { + "title": "不忘科技-随机种子-🐱" + } + }, + "225": { + "inputs": { + "frame_rate": [ + "426", + 0 + ], + "loop_count": 0, + "filename_prefix": "e2f30db2-cfc9-4325-92d4-15eab21b14a6", + "format": "video/h264-mp4", + "pix_fmt": "yuv420p", + "crf": 19, + "save_metadata": false, + "trim_to_audio": false, + "pingpong": false, + "save_output": true, + "images": [ + "508", + 0 + ], + "audio": [ + "515", + 0 + ] + }, + "class_type": "VHS_VideoCombine", + "_meta": { + "title": "Video Combine 🎥🅥🅗🅢" + } + }, + "330": { + "inputs": { + "input_id": "start_time", + "default_value": "00:00:02.000", + "display_name": "", + "description": "00:01:01" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "331": { + "inputs": { + "input_id": "tts_text3", + "default_value": "再配个小白鞋，休闲又时尚。", + "display_name": "文本3", + "description": "文本3" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "399": { + "inputs": { + "input_id": "tts_text2", + "default_value": "一共只有5件，到手价三千八百六。", + "display_name": "文本2", + "description": "文本2" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "406": { + "inputs": { + "input_id": "tts_text4", + "default_value": "这个款其实挺年轻的，", + "display_name": "文本4", + "description": "文本4" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "422": { + "inputs": { + "start_time": [ + "330", + 0 + ], + "end_padding": 0.4, + "fps": [ + "426", + 0 + ], + "audio": [ + "515", + 0 + ] + }, + "class_type": "VideoPointCompute", + "_meta": { + "title": "视频帧位计算" + } + }, + "424": { + "inputs": { + "input_id": "video_path", + "default_value": "input_s3/markers/5810/5810.mp4", + "display_name": "", + "description": "" + }, + "class_type": "ComfyUIDeployExternalText", + "_meta": { + "title": "External Text (ComfyUI Deploy)" + } + }, + "426": { + "inputs": { + "Number": "25" + }, + "class_type": "Int", + "_meta": { + "title": "帧率" + } + }, + "496": { + "inputs": { + "delimiter": "\\n", + "clean_whitespace": "true", + "text_a": [ + "211", + 0 + ], + "text_b": [ + "399", + 0 + ], + "text_c": [ + "331", + 0 + ], + "text_d": [ + "406", + 0 + ] + }, + "class_type": "Text Concatenate", + "_meta": { + "title": "Text Concatenate" + } + }, + "498": { + "inputs": { + "video_path": [ + "424", + 0 + ], + "start_point": [ + "422", + 0 + ], + "duration": [ + "422", + 1 + ], + "fps": [ + "426", + 0 + ], + "force_match_fps": true + }, + "class_type": "VideoCutByFramePoint", + "_meta": { + "title": "视频剪裁(精确帧位)" + } + }, + "502": { + "inputs": { + "images": [ + "498", + 0 + ], + "audio": [ + "515", + 0 + ] + }, + "class_type": "VideoLengthAdjuster", + "_meta": { + "title": "Video Length Adjuster" + } + }, + "508": { + "inputs": { + "seed": [ + "217", + 0 + ], + "images": [ + "502", + 0 + ], + "audio": [ + "502", + 1 + ] + }, + "class_type": "LatentSync1.5", + "_meta": { + "title": "LatentSync1.5" + } + }, + "511": { + "inputs": { + "video_path": [ + "512", + 0 + ], + "fps": 30 + }, + "class_type": "VideoChangeFPS", + "_meta": { + "title": "视频转换帧率" + } + }, + "512": { + "inputs": { + "index": -1, + "filenames": [ + "225", + 0 + ] + }, + "class_type": "VHS_SelectFilename", + "_meta": { + "title": "Select Filename 🎥🅥🅗🅢" + } + }, + "515": { + "inputs": { + "text_content": [ + "496", + 0 + ], + "anchor_id": "huangzitao_high", + "speed": 1.14, + "auto_download": false + }, + "class_type": "CosyVoiceTTS", + "_meta": { + "title": "CosyVoice文本转语音" + } + } +} \ No newline at end of file diff --git a/src/cluster/app.py b/src/cluster/app.py index 20f5ca7..344e116 100644 --- a/src/cluster/app.py +++ b/src/cluster/app.py @@ -3,6 +3,8 @@ from BowongModalFunctions.config import WorkerConfig from .video import app as media_app from .web import app as web_app from .ffmpeg_app import app as ffmpeg_app +from .comfyui_v1 import app as comfyui_v1_app +from .comfyui_v2 import app as comfyui_v2_app config = WorkerConfig() @@ -14,3 +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) diff --git a/src/cluster/comfyui_v1.py b/src/cluster/comfyui_v1.py new file mode 100644 index 0000000..8fcbb96 --- /dev/null +++ b/src/cluster/comfyui_v1.py @@ -0,0 +1,549 @@ +# ComfyUI模板--Base Auth +import modal +from dotenv import dotenv_values + +comfyui_image = ( + modal.Image.debian_slim( + python_version="3.10" + ) + .pip_install_from_pyproject("pyproject_comfyui.toml") + .apt_install("wget", "software-properties-common") + .run_commands("add-apt-repository -y contrib " + "&& wget https://developer.download.nvidia.com/compute/cuda/repos/debian12/x86_64/cuda-keyring_1.1-1_all.deb&&dpkg -i cuda-keyring_1.1-1_all.deb") + .apt_install("git", "gcc", "libportaudio2", "cuda-toolkit", "ffmpeg") + .add_local_file("whl/comfy_cli-0.0.0-py3-none-any.whl", "/root/comfy_cli-0.0.0-py3-none-any.whl",copy=True) + .pip_install("/root/comfy_cli-0.0.0-py3-none-any.whl") + .run_commands("comfy --skip-prompt install --nvidia --version 0.3.10") + .env(dotenv_values(".runtime.env")) + .workdir("/root/comfy") +) + +comfyui_image = ( + comfyui_image.run_commands("comfy node install https://github.com/M1kep/ComfyLiterals") + .run_commands("comfy node install https://github.com/evanspearman/ComfyMath") + .run_commands("comfy node install https://github.com/Kosinkadink/ComfyUI-AnimateDiff-Evolved") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-Bowong.git") + .run_commands("comfy node install https://github.com/crystian/ComfyUI-Crystools") + .run_commands("comfy node install https://github.com/pythongosssss/ComfyUI-Custom-Scripts") + .run_commands("comfy node install https://github.com/BennyKok/comfyui-deploy") + .run_commands("comfy node install https://github.com/yolain/ComfyUI-Easy-Use") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-VideoHelperSuite.git") + .run_commands("comfy node install https://github.com/jags111/efficiency-nodes-comfyui") + .run_commands("comfy node install https://github.com/WASasquatch/was-node-suite-comfyui") + .run_commands("comfy node install https://github.com/rgthree/rgthree-comfy") + .run_commands("comfy node install https://github.com/cubiq/ComfyUI_essentials") + .run_commands("comfy node install https://github.com/melMass/comfy_mtb") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI_SparkTTS.git") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-LatentSync-Node.git") + .run_commands( + "mkdir -p /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model && ln -s /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Node/checkpoints && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-LatentSync-Node/checkpoints && ln -s /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Node/checkpoints /root/comfy/ComfyUI/custom_nodes/ComfyUI-LatentSync-Node/checkpoints" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints && mkdir -p /root/.cache/torch/hub/checkpoints && rm -rf /root/.cache/torch/hub/checkpoints && ln -s /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints /root/.cache/torch/hub/checkpoints" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/stabilityai && ln -s /root/comfy/ComfyUI/models/stabilityai /root/comfy/ComfyUI/stabilityai" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M && mkdir -p /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && rm -rf /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && ln -s /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M" + ).run_commands( + "rm -rf /root/comfy/ComfyUI/models" + ) + .add_local_file("config/config.yaml", "/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",copy=True) + .add_local_file("config/config.py","/root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/tools/config.py",copy=True) + .add_local_file("src/BowongModalFunctions/template/template_v1.json", "/root/src/BowongModalFunctions/template/template_v1.json", copy=True) + .add_local_python_source('BowongModalFunctions') + .add_local_python_source('cluster') +) + +app = modal.App(name="ComfyUI v1", image=comfyui_image, include_source=False) +app.set_description("ComfyUI v1 Server") + +# @app.cls() +# @modal.concurrent(max_inputs=1) +# class testCls: +# @modal.method() +# def api(self,a,b): +# pass + +with comfyui_image.imports(): + import json + import os + import time + import shutil + import subprocess + import uuid + import psutil + from typing import Tuple, Any, Dict, Optional + import loguru + import sentry_sdk + from modal import current_function_call_id + from BowongModalFunctions.config import WorkerConfig + from BowongModalFunctions.models.web_model import SentryTransactionInfo + from sentry_sdk.integrations.loguru import LoguruIntegration + + config = WorkerConfig() + vol = modal.Volume.from_name("comfyui-model", create_if_missing=True, environment_name=config.modal_environment) + secret = modal.Secret.from_name("aws-s3-secret", environment_name=config.modal_environment) + output_dir = "/root/comfy/ComfyUI/output" + + sentry_sdk.init( + dsn="https://7e3e37e770420e289b3972e1b1b9a794@sentry.bowongai.com/11", + send_default_pii=True, + attach_stacktrace=True, + traces_sample_rate=1.0, # 采样率设为1.0表示捕获所有事务 + profiles_sample_rate=1.0, + add_full_stack=True, + shutdown_timeout=2, + integrations=[LoguruIntegration()], + environment=config.modal_environment, + ) + + + @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) + + + @app.cls( + max_containers=200, + min_containers=0, + buffer_containers=0, + scaledown_window=120, + timeout=1200, + gpu=["L4"], + cpu=(2, 64), + memory=(20480, 131072), + enable_memory_snapshot=False, + secrets=[secret], + volumes={ + "/root/comfy/ComfyUI/models": vol, + "/root/comfy/ComfyUI/input_s3": modal.CloudBucketMount( + bucket_name=config.S3_bucket_name, + secret=secret, + key_prefix=config.comfyui_s3_input+"/" + ), + "/root/comfy/ComfyUI/output_s3": modal.CloudBucketMount( + bucket_name=config.S3_bucket_name, + secret=secret, + key_prefix=config.comfyui_s3_output+"/" + ), + }, + ) + @modal.concurrent(max_inputs=1) + class ComfyUI: + @modal.enter() + def launch_comfy_background(self): + self.session_id = str(uuid.uuid4()) + self.file_prefix = None + cmd = "echo client_uuid: {} && comfy launch --background".format( + self.session_id, self.session_id, self.session_id) + subprocess.run(cmd, shell=True, check=True) + + @modal.method() + def infer(self, workflow: Dict): + self.poll_server_health() + self.prompt_uuid = str(uuid.uuid4()) + print("Workflow JSON:") + print(json.dumps(workflow, indent=4, ensure_ascii=False)) + for node in workflow.values(): + if node.get("class_type") == "ComfyUIDeployExternalText": + if node.get("inputs").get("input_id") == "file_path": + file_to_move = node.get("inputs").get("default_value") + if file_to_move: + os.makedirs(os.path.dirname(file_to_move.replace("input_s3", "input")), exist_ok=True) + try: + shutil.copy(file_to_move, file_to_move.replace("input_s3", "input")) + except: + try: + print("Try download file from S3 manually") + # S3 Fallback + import boto3 + import yaml + with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml", + encoding="utf-8", mode="r+") as myconfig: + yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader) + awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"], + aws_secret_access_key=yaml_config["aws_access_key"]) + awss3.meta.client.download_file(config.S3_bucket_name, config.comfyui_s3_input+"/"+file_to_move.split("input_s3/")[1], + file_to_move.replace("input_s3", "input")) + except: + raise Exception("Failed to download file from S3 manually") + node["inputs"]["default_value"] = node["inputs"]["default_value"].replace("input_s3", + "input") + with open(f"/root/{self.prompt_uuid}.json", "w", encoding="utf-8") as fi: + fi.write(json.dumps(workflow, ensure_ascii=False)) + cmd = f"comfy run --workflow /root/{self.prompt_uuid}.json --wait --timeout 1190 --verbose" + + self.file_prefix = [ + node.get("inputs") + for node in workflow.values() + if node.get("class_type") == "VHS_VideoCombine" + ][0]["filename_prefix"] + + subprocess.run(cmd, shell=True, check=True, timeout=1195) + # 获取按照文件时间创建排序的列表，默认是按时间升序 + file_list = os.listdir(output_dir) + new_file_list = sorted(file_list, key=lambda file: os.path.getctime(os.path.join(output_dir, file)), + reverse=True) + for f in new_file_list: + if f.startswith(self.file_prefix): + os.makedirs(os.path.dirname(os.path.join(output_dir.replace("output", "output_s3"), f)), + exist_ok=True) + try: + shutil.copy(os.path.join(output_dir, f), + os.path.join(output_dir.replace("output", "output_s3"), f)) + except: + try: + print("Try move file to S3 manually") + # S3 Fallback + import boto3 + import yaml + with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml", + encoding="utf-8", mode="r+") as myconfig: + yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader) + awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"], + aws_secret_access_key=yaml_config["aws_access_key"]) + awss3.meta.client.upload_file(os.path.join(output_dir, f), config.S3_bucket_name, config.comfyui_s3_output+"/"+f) + except: + raise Exception("Failed to move file to S3 manually") + return f + + @modal.method() + def api(self, video_path: str, start_time: str = "00:00:03.600", filename_prefix=str(uuid.uuid4()), + tts_text1: str = "测试", + tts_text2: str = "", tts_text3: str = "", tts_text4: str = "", anchor_id: str = "dawan", + speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None): + # 导入必要模块 + import datetime + import tempfile + import traceback + + # 生成当前函数调用的唯一ID + function_call_id = current_function_call_id() + print(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}") + + # 使用临时文件 + temp_log_path = None + error_reported = False # 添加标志跟踪是否已报告错误 + + # 创建主Sentry事务 + if sentry_trace and sentry_trace.x_trace_id and sentry_trace.x_baggage: + transaction = sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id, + "baggage": sentry_trace.x_baggage, }) + else: + transaction = sentry_sdk.start_transaction(op="video.processing", name=f"API调用: {video_path}") + sentry_trace = SentryTransactionInfo(x_trace_id=transaction.trace_id, x_baggage=transaction.containing_transaction.get_baggage().serialize()) + with transaction: + # 设置全局Sentry上下文 + with sentry_sdk.configure_scope() as scope: + # 添加更丰富的上下文数据 + scope.set_tag("session_id", self.session_id) + scope.set_tag("function_call_id", function_call_id) + scope.set_tag("video_path", video_path) + scope.set_context("api_parameters", { + "current_function_call_id": function_call_id, + "session_id": self.session_id, + "video_path": video_path, + "start_time": start_time, + "filename_prefix": filename_prefix, + "tts_text1": tts_text1, + "tts_text2": tts_text2, + "tts_text3": tts_text3, + "tts_text4": tts_text4, + "anchor_id": anchor_id, + "speed": speed + }) + + try: + # 使用Span跟踪日志文件创建 + with sentry_sdk.start_span(op="db.file", description="创建日志文件") as span: + # 创建临时日志文件 + temp_log_file = tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt') + temp_log_path = temp_log_file.name + print(f"创建临时日志文件: {temp_log_path}") + + span.set_data("log_path", temp_log_path) + span.set_tag("file_operation", "create") + + # 写入初始日志内容 + temp_log_file.write(f"任务开始时间: {datetime.datetime.now()}\n") + temp_log_file.write(f"会话ID: {self.session_id}\n") + temp_log_file.write(f"调用ID: {function_call_id}\n") + temp_log_file.write(f"文件前缀: {filename_prefix}\n") + temp_log_file.write(f"参数: video_path={video_path}, start_time={start_time}\n") + temp_log_file.flush() + temp_log_file.close() + + # 使用Span跟踪模板处理 + with sentry_sdk.start_span(op="template.json", description="处理工作流模板") as span: + # 正常业务逻辑 + with open("/root/src/BowongModalFunctions/template/template_v1.json", "r", + encoding="utf-8") as template: + workflow_file = json.loads(template.read()) + + # 使用子span跟踪模板修改 + with sentry_sdk.start_span(op="template.modify", description="修改模板参数") as modify_span: + modify_span.set_data("original_keys", len(workflow_file)) + + workflow_file["211"]["inputs"]["default_value"] = tts_text1 + workflow_file["399"]["inputs"]["default_value"] = tts_text2 + workflow_file["331"]["inputs"]["default_value"] = tts_text3 + workflow_file["406"]["inputs"]["default_value"] = tts_text4 + workflow_file["424"]["inputs"][ + "default_value"] = "/root/comfy/ComfyUI/input_s3/" + video_path + workflow_file["225"]["inputs"]["filename_prefix"] = filename_prefix + workflow_file["330"]["inputs"]["default_value"] = start_time + workflow_file["515"]["inputs"]["anchor_id"] = anchor_id + workflow_file["515"]["inputs"]["speed"] = speed + + modify_span.set_data("modified_nodes", + ["211", "399", "331", "406", "424", "225", "330", "515"]) + + span.set_data("workflow_size", len(workflow_file)) + span.set_tag("workflow_type", "video_processing") + + try: + # 记录执行的关键步骤作为breadcrumbs + sentry_sdk.add_breadcrumb( + category="process", + message="开始调用infer方法", + level="info" + ) + + # 使用Span跟踪推理过程 + with sentry_sdk.start_span(op="ai.inference", description="执行模型推理") as infer_span: + infer_span.set_data("workflow_size", len(workflow_file)) + infer_start_time = datetime.datetime.now() + + # 创建预处理子span + with sentry_sdk.start_span(op="inference.prepare", description="准备推理环境") as prep_span: + prep_span.set_data("timestamp", datetime.datetime.now().isoformat()) + prep_span.set_tag("model_type", "video_generation") + + # 调用推理方法 + fname = self.infer.local(workflow_file) + + # 创建后处理子span + with sentry_sdk.start_span(op="inference.postprocess", + description="处理推理结果") as post_span: + post_span.set_data("output_file", fname) + post_span.set_data("timestamp", datetime.datetime.now().isoformat()) + + infer_end_time = datetime.datetime.now() + duration_seconds = (infer_end_time - infer_start_time).total_seconds() + + infer_span.set_data("start_time", infer_start_time.isoformat()) + infer_span.set_data("end_time", infer_end_time.isoformat()) + infer_span.set_data("duration_seconds", duration_seconds) + infer_span.set_data("output_file", fname) + infer_span.set_tag("success", "true" if fname else "false") + + sentry_sdk.add_breadcrumb( + category="process", + message="infer方法调用完成", + level="info" + ) + + if fname is None: + with sentry_sdk.start_span(op="error.handling", description="处理文件缺失错误"): + raise RuntimeError("Output File not found") + + # 使用Span跟踪返回结果 + with sentry_sdk.start_span(op="result.preparation", description="准备返回结果"): + j = {"status": "success", "file_name": fname} + loguru.logger.success(j) + + # 设置事务状态为成功 + transaction.set_status("ok") + + return j, sentry_trace + + except Exception as infer_error: + # 使用原始异常对象 + sentry_sdk.add_breadcrumb( + category="error", + message=f"infer方法调用失败: {str(infer_error)}", + level="error" + ) + with sentry_sdk.start_span(op="error.handling", description="处理推理错误") as error_span: + error_span.set_data("error_type", type(infer_error).__name__) + error_span.set_data("error_message", str(infer_error)) + error_span.set_tag("error", "true") + raise infer_error + + except Exception as e: + # 异常处理 - 使用捕获的异常对象e + j = {"status": "fail", "msg": str(e)} + + try: + # 使用Span跟踪错误记录过程 + with sentry_sdk.start_span(op="error.logging", description="记录错误详情") as error_span: + # 更新日志文件 + with open(temp_log_path, "a", encoding="utf-8") as f: + # 记录异常详情 + f.write(f"\n====== 异常详情 ======\n") + f.write(f"异常类型: {type(e).__name__}\n") + f.write(f"异常信息: {str(e)}\n") + f.write(f"发生时间: {datetime.datetime.now()}\n") + f.write(f"会话ID: {self.session_id}\n") + f.write(f"调用ID: {function_call_id}\n") + + # 记录完整堆栈 + f.write("\n====== 堆栈跟踪 ======\n") + f.write(traceback.format_exc()) + + # 尝试添加ComfyUI日志 + with sentry_sdk.start_span(op="db.file.read", + description="读取ComfyUI日志") as log_span: + if os.path.exists("/root/comfy/ComfyUI/user/comfyui.log"): + try: + with open("/root/comfy/ComfyUI/user/comfyui.log", "r", + encoding="utf-8") as cf: + f.write("\n====== ComfyUI日志 ======\n") + f.write(cf.read()) + log_span.set_tag("success", "true") + except Exception as log_read_error: + f.write(f"\n无法读取ComfyUI日志: {str(log_read_error)}\n") + log_span.set_data("error", str(log_read_error)) + log_span.set_tag("success", "false") + else: + log_span.set_tag("file_exists", "false") + + error_span.set_data("log_path", temp_log_path) + error_span.set_tag("error_type", type(e).__name__) + + # 仅当尚未报告错误时才添加附件和报告异常 + if not error_reported: + with sentry_sdk.configure_scope() as scope: + with sentry_sdk.start_span(op="error.report", + description="向Sentry报告错误") as report_span: + scope.add_attachment( + path=temp_log_path, + filename=f"full_{function_call_id}.txt", + add_to_transactions=True + ) + + # 添加更多异常上下文 + scope.set_context("error_details", { + "error_type": type(e).__name__, + "error_message": str(e), + "time_of_error": datetime.datetime.now().isoformat() + }) + + # 设置事务状态为错误 + transaction.set_status("internal_error") + + # 使用异常对象e + sentry_sdk.capture_exception(e) + error_reported = True + print("已向Sentry报告异常") + report_span.set_tag("reported", "true") + + except Exception as log_error: + print(f"处理日志出错: {log_error}") + # 设置事务状态为错误 + transaction.set_status("internal_error") + + # 仅当尚未报告错误时才报告原始异常 + if not error_reported: + sentry_sdk.capture_exception(e) + print("已向Sentry报告原始异常（日志处理失败）") + + return j, sentry_trace + + finally: + # 使用Span跟踪清理过程 + with sentry_sdk.start_span(op="system.cleanup", description="清理资源") as cleanup_span: + # 清理临时文件和重启ComfyUI + try: + # 尝试复制日志到S3 + if temp_log_path and os.path.exists(temp_log_path): + with sentry_sdk.start_span(op="db.file.copy", description="复制日志到S3") as copy_span: + log_dir = f"/root/comfy/ComfyUI/output_s3/logs/{self.session_id}" + try: + os.makedirs(log_dir, exist_ok=True) + shutil.copy(temp_log_path, f"{log_dir}/full.txt") + print(f"已复制日志到: {log_dir}/full.txt") + copy_span.set_data("source", temp_log_path) + copy_span.set_data("destination", f"{log_dir}/full.txt") + copy_span.set_tag("success", "true") + except Exception as copy_error: + print(f"复制日志到S3失败: {copy_error}") + copy_span.set_data("error", str(copy_error)) + copy_span.set_tag("success", "false") + + # 清理临时文件 + with sentry_sdk.start_span(op="db.file.delete", description="删除临时文件") as del_span: + try: + os.unlink(temp_log_path) + print(f"已清理临时文件: {temp_log_path}") + del_span.set_data("file", temp_log_path) + del_span.set_tag("success", "true") + except Exception as unlink_error: + print(f"清理临时文件失败: {unlink_error}") + del_span.set_data("error", str(unlink_error)) + del_span.set_tag("success", "false") + except Exception as cleanup_error: + print(f"清理过程出错: {cleanup_error}") + cleanup_span.set_data("error", str(cleanup_error)) + cleanup_span.set_tag("success", "false") + + print("清理资源并重启ComfyUI") + with sentry_sdk.start_span(op="system.restart", description="重启ComfyUI") as restart_span: + try: + # 停止ComfyUI + with sentry_sdk.start_span(op="system.stop", description="停止ComfyUI") as stop_span: + cmd = "comfy stop" + subprocess.run(cmd, shell=True, check=True) + time.sleep(1) + stop_span.set_tag("success", "true") + except Exception as stop_error: + print(f"停止ComfyUI失败: {stop_error}") + restart_span.set_data("stop_error", str(stop_error)) + restart_span.set_tag("stop_success", "false") + + try: + # 重启ComfyUI + with sentry_sdk.start_span(op="system.start", description="启动ComfyUI") as start_span: + cmd = "comfy launch --background" + subprocess.run(cmd, shell=True, check=True) + start_span.set_tag("success", "true") + except Exception as start_error: + print(f"启动ComfyUI失败: {start_error}") + restart_span.set_data("start_error", str(start_error)) + restart_span.set_tag("start_success", "false") + modal.experimental.stop_fetching_inputs() + + + def poll_server_health(self): + import socket + import urllib + + try: + # dummy request to check if the server is healthy + req = urllib.request.Request("http://127.0.0.1:8188/system_stats") + urllib.request.urlopen(req, timeout=5) + print("ComfyUI server is healthy") + except (socket.timeout, urllib.error.URLError) as e: + # if no response in 5 seconds, stop the container; Modal will schedule queued inputs on a new container + print(f"Server health check failed: {str(e)} restarting ComfyUI...") + try: + cmd = "comfy stop" + try: + subprocess.run(cmd, shell=True, check=True) + except: + pass + cmd = "comfy launch --background" + try: + subprocess.run(cmd, shell=True, check=True) + except: + raise Exception("Failed to launch ComfyUI") + except: + modal.experimental.stop_fetching_inputs() + raise Exception("ComfyUI server is not healthy, restart failed, stopping container") \ No newline at end of file diff --git a/src/cluster/comfyui_v2.py b/src/cluster/comfyui_v2.py new file mode 100644 index 0000000..2c97b51 --- /dev/null +++ b/src/cluster/comfyui_v2.py @@ -0,0 +1,538 @@ +# ComfyUI模板--Base Auth LatentSync1.5 +import modal +from dotenv import dotenv_values + +comfyui_latentsync_1_5_image = ( + modal.Image.debian_slim( + python_version="3.10" + ) + .pip_install_from_pyproject("pyproject_comfyui.toml") + .apt_install("wget", "software-properties-common") + .run_commands("add-apt-repository -y contrib " + "&& wget https://developer.download.nvidia.com/compute/cuda/repos/debian12/x86_64/cuda-keyring_1.1-1_all.deb&&dpkg -i cuda-keyring_1.1-1_all.deb") + .apt_install("git", "gcc", "libportaudio2", "cuda-toolkit", "ffmpeg") + .add_local_file("whl/comfy_cli-0.0.0-py3-none-any.whl", "/root/comfy_cli-0.0.0-py3-none-any.whl", copy=True) + .pip_install("/root/comfy_cli-0.0.0-py3-none-any.whl") + .run_commands("comfy --skip-prompt install --nvidia --version 0.3.10") + .env(dotenv_values(".runtime.env")) + .workdir("/root/comfy") +) + +comfyui_latentsync_1_5_image = ( + comfyui_latentsync_1_5_image.run_commands("comfy node install https://github.com/M1kep/ComfyLiterals") + .run_commands("comfy node install https://github.com/evanspearman/ComfyMath") + .run_commands("comfy node install https://github.com/Kosinkadink/ComfyUI-AnimateDiff-Evolved") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-Bowong.git") + .run_commands("comfy node install https://github.com/crystian/ComfyUI-Crystools") + .run_commands("comfy node install https://github.com/pythongosssss/ComfyUI-Custom-Scripts") + .run_commands("comfy node install https://github.com/BennyKok/comfyui-deploy") + .run_commands("comfy node install https://github.com/yolain/ComfyUI-Easy-Use") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-VideoHelperSuite.git") + .run_commands("comfy node install https://github.com/jags111/efficiency-nodes-comfyui") + .run_commands("comfy node install https://github.com/WASasquatch/was-node-suite-comfyui") + .run_commands("comfy node install https://github.com/rgthree/rgthree-comfy") + .run_commands("comfy node install https://github.com/cubiq/ComfyUI_essentials") + .run_commands("comfy node install https://github.com/melMass/comfy_mtb") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI_SparkTTS.git") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git") + .run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/Comfy-LatentSync-Next-Node.git") + .run_commands( + "mkdir -p /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model && ln -s /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Next-Node/checkpoints && rm -rf /root/comfy/ComfyUI/custom_nodes/Comfy-LatentSync-Next-Node/checkpoints && ln -s /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Next-Node/checkpoints /root/comfy/ComfyUI/custom_nodes/Comfy-LatentSync-Next-Node/checkpoints" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints && mkdir -p /root/.cache/torch/hub/checkpoints && rm -rf /root/.cache/torch/hub/checkpoints && ln -s /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints /root/.cache/torch/hub/checkpoints" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/stabilityai && ln -s /root/comfy/ComfyUI/models/stabilityai /root/comfy/ComfyUI/stabilityai" + ).run_commands( + "mkdir -p /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M && mkdir -p /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && rm -rf /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && ln -s /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M" + ).run_commands( + "rm -rf /root/comfy/ComfyUI/models" + ) + .add_local_file("config/config.yaml", "/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml", copy=True) + .add_local_file("config/config.py", "/root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/tools/config.py", copy=True) + .add_local_file("src/BowongModalFunctions/template/template_v2.json", "/root/src/BowongModalFunctions/template/template_v2.json", copy=True) + .add_local_python_source('cluster') + .add_local_python_source('BowongModalFunctions') +) + +app = modal.App(name="ComfyUI v2", image=comfyui_latentsync_1_5_image, include_source=False) +app.set_description("ComfyUI v2 Server") + + +with comfyui_latentsync_1_5_image.imports(): + import json + import os + import shutil + import subprocess + import time + import uuid + from typing import Dict, Tuple, Any, Optional + import loguru + import sentry_sdk + import psutil + from BowongModalFunctions.config import WorkerConfig + from modal import current_function_call_id + from sentry_sdk.integrations.loguru import LoguruIntegration + from BowongModalFunctions.models.web_model import SentryTransactionInfo + + config = WorkerConfig() + vol = modal.Volume.from_name("comfyui-model", create_if_missing=True, environment_name=config.modal_environment) + secret = modal.Secret.from_name("aws-s3-secret", environment_name=config.modal_environment) + output_dir = "/root/comfy/ComfyUI/output" + + sentry_sdk.init( + dsn="https://7e3e37e770420e289b3972e1b1b9a794@sentry.bowongai.com/11", + send_default_pii=True, + attach_stacktrace=True, + traces_sample_rate=1.0, # 采样率设为1.0表示捕获所有事务 + profiles_sample_rate=1.0, + add_full_stack=True, + shutdown_timeout=2, + integrations=[LoguruIntegration()], + environment=config.modal_environment, + ) + + @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) + + @app.cls( + max_containers=200, + min_containers=0, + buffer_containers=0, + scaledown_window=120, + timeout=1200, + gpu=["L4"], + cpu=(2,64), + memory=(20480,131072), + enable_memory_snapshot=False, + secrets=[secret], + volumes={ + "/root/comfy/ComfyUI/models": vol, + "/root/comfy/ComfyUI/input_s3": modal.CloudBucketMount( + bucket_name=config.S3_bucket_name, + secret=secret, + key_prefix=config.comfyui_s3_input+"/" + ), + "/root/comfy/ComfyUI/output_s3": modal.CloudBucketMount( + bucket_name=config.S3_bucket_name, + secret=secret, + key_prefix=config.comfyui_s3_output+"/" + ), + }, + ) + @modal.concurrent(max_inputs=1) + class ComfyUILatentSync15: + @modal.enter() + def launch_comfy_background(self): + self.session_id = str(uuid.uuid4()) + self.file_prefix = None + cmd = "echo client_uuid: {} && comfy launch --background".format(self.session_id,self.session_id, self.session_id) + subprocess.run(cmd, shell=True, check=True) + + @modal.method() + def infer(self, workflow: Dict): + self.poll_server_health() + self.prompt_uuid = str(uuid.uuid4()) + print("Workflow JSON:") + print(json.dumps(workflow, indent=4, ensure_ascii=False)) + for node in workflow.values(): + if node.get("class_type") == "ComfyUIDeployExternalText": + if node.get("inputs").get("input_id") == "file_path": + file_to_move = node.get("inputs").get("default_value") + if file_to_move: + os.makedirs(os.path.dirname(file_to_move.replace("input_s3", "input")), exist_ok=True) + try: + shutil.copy(file_to_move, file_to_move.replace("input_s3", "input")) + except: + try: + print("Try download file to S3 manually") + # S3 Fallback + import boto3 + import yaml + with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",encoding="utf-8",mode="r+") as myconfig: + yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader) + awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"], + aws_secret_access_key=yaml_config["aws_access_key"]) + awss3.meta.client.download_file(config.S3_bucket_name, config.comfyui_s3_input + "/" + + file_to_move.split("input_s3/")[1], + file_to_move.replace("input_s3", "input")) + except: + raise Exception("Failed to download file from S3 manually") + node["inputs"]["default_value"] = node["inputs"]["default_value"].replace("input_s3", "input") + with open(f"/root/{self.prompt_uuid}.json", "w", encoding="utf-8") as fi: + fi.write(json.dumps(workflow, ensure_ascii=False)) + cmd = f"comfy run --workflow /root/{self.prompt_uuid}.json --wait --timeout 1190 --verbose" + + self.file_prefix = [ + node.get("inputs") + for node in workflow.values() + if node.get("class_type") == "VHS_VideoCombine" + ][0]["filename_prefix"] + + subprocess.run(cmd, shell=True, check=True, timeout=1195) + + file_list = os.listdir(output_dir) + # 获取按照文件时间创建排序的列表，默认是按时间升序 + new_file_list = sorted(file_list, key=lambda file: os.path.getctime(os.path.join(output_dir, file)), + reverse=True) + for f in new_file_list: + if f.startswith(self.file_prefix): + os.makedirs(os.path.dirname(os.path.join(output_dir.replace("output", "output_s3"), f)), exist_ok=True) + try: + shutil.copy(os.path.join(output_dir, f), os.path.join(output_dir.replace("output", "output_s3"), f)) + except: + try: + print("Try move file to S3 manually") + # S3 Fallback + import boto3 + import yaml + with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml", encoding="utf-8",mode="r+") as myconfig: + yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader) + awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"], + aws_secret_access_key=yaml_config["aws_access_key"]) + awss3.meta.client.upload_file(os.path.join(output_dir, f), config.S3_bucket_name, config.comfyui_s3_output+"/"+f) + except: + raise Exception("Failed to move file to S3 manually") + return f + + + @modal.method() + def api(self, video_path: str, start_time: str = "00:00:03.600", filename_prefix=str(uuid.uuid4()), + tts_text1: str = "测试", + tts_text2: str = "", tts_text3: str = "", tts_text4: str = "", anchor_id: str = "dawan", + speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None): + # 导入必要模块 + import datetime + import tempfile + import traceback + + # 生成当前函数调用的唯一ID + function_call_id = current_function_call_id() + print(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}") + + # 使用临时文件 + temp_log_path = None + error_reported = False # 添加标志跟踪是否已报告错误 + + # 创建主Sentry事务 + if sentry_trace and sentry_trace.x_trace_id and sentry_trace.x_baggage: + transaction = sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id, + "baggage": sentry_trace.x_baggage, }) + else: + transaction = sentry_sdk.start_transaction(op="video.processing", name=f"API调用: {video_path}") + sentry_trace = SentryTransactionInfo(x_trace_id=transaction.trace_id, x_baggage=transaction.containing_transaction.get_baggage().serialize()) + with transaction: + # 设置全局Sentry上下文 + with sentry_sdk.configure_scope() as scope: + # 添加更丰富的上下文数据 + scope.set_tag("session_id", self.session_id) + scope.set_tag("function_call_id", function_call_id) + scope.set_tag("video_path", video_path) + scope.set_context("api_parameters", { + "current_function_call_id": function_call_id, + "session_id": self.session_id, + "video_path": video_path, + "start_time": start_time, + "filename_prefix": filename_prefix, + "tts_text1": tts_text1, + "tts_text2": tts_text2, + "tts_text3": tts_text3, + "tts_text4": tts_text4, + "anchor_id": anchor_id, + "speed": speed + }) + + try: + # 使用Span跟踪日志文件创建 + with sentry_sdk.start_span(op="db.file", description="创建日志文件") as span: + # 创建临时日志文件 + temp_log_file = tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt') + temp_log_path = temp_log_file.name + print(f"创建临时日志文件: {temp_log_path}") + + span.set_data("log_path", temp_log_path) + span.set_tag("file_operation", "create") + + # 写入初始日志内容 + temp_log_file.write(f"任务开始时间: {datetime.datetime.now()}\n") + temp_log_file.write(f"会话ID: {self.session_id}\n") + temp_log_file.write(f"调用ID: {function_call_id}\n") + temp_log_file.write(f"文件前缀: {filename_prefix}\n") + temp_log_file.write(f"参数: video_path={video_path}, start_time={start_time}\n") + temp_log_file.flush() + temp_log_file.close() + + # 使用Span跟踪模板处理 + with sentry_sdk.start_span(op="template.json", description="处理工作流模板") as span: + # 正常业务逻辑 + with open("/root/src/BowongModalFunctions/template/template_v2.json", "r", + encoding="utf-8") as template: + workflow_file = json.loads(template.read()) + + # 使用子span跟踪模板修改 + with sentry_sdk.start_span(op="template.modify", description="修改模板参数") as modify_span: + modify_span.set_data("original_keys", len(workflow_file)) + + workflow_file["211"]["inputs"]["default_value"] = tts_text1 + workflow_file["399"]["inputs"]["default_value"] = tts_text2 + workflow_file["331"]["inputs"]["default_value"] = tts_text3 + workflow_file["406"]["inputs"]["default_value"] = tts_text4 + workflow_file["424"]["inputs"][ + "default_value"] = "/root/comfy/ComfyUI/input_s3/" + video_path + workflow_file["225"]["inputs"]["filename_prefix"] = filename_prefix + workflow_file["330"]["inputs"]["default_value"] = start_time + workflow_file["515"]["inputs"]["anchor_id"] = anchor_id + workflow_file["515"]["inputs"]["speed"] = speed + + modify_span.set_data("modified_nodes", + ["211", "399", "331", "406", "424", "225", "330", "515"]) + + span.set_data("workflow_size", len(workflow_file)) + span.set_tag("workflow_type", "video_processing") + + try: + # 记录执行的关键步骤作为breadcrumbs + sentry_sdk.add_breadcrumb( + category="process", + message="开始调用infer方法", + level="info" + ) + + # 使用Span跟踪推理过程 + with sentry_sdk.start_span(op="ai.inference", description="执行模型推理") as infer_span: + infer_span.set_data("workflow_size", len(workflow_file)) + infer_start_time = datetime.datetime.now() + + # 创建预处理子span + with sentry_sdk.start_span(op="inference.prepare", description="准备推理环境") as prep_span: + prep_span.set_data("timestamp", datetime.datetime.now().isoformat()) + prep_span.set_tag("model_type", "video_generation") + + # 调用推理方法 + fname = self.infer.local(workflow_file) + + # 创建后处理子span + with sentry_sdk.start_span(op="inference.postprocess", + description="处理推理结果") as post_span: + post_span.set_data("output_file", fname) + post_span.set_data("timestamp", datetime.datetime.now().isoformat()) + + infer_end_time = datetime.datetime.now() + duration_seconds = (infer_end_time - infer_start_time).total_seconds() + + infer_span.set_data("start_time", infer_start_time.isoformat()) + infer_span.set_data("end_time", infer_end_time.isoformat()) + infer_span.set_data("duration_seconds", duration_seconds) + infer_span.set_data("output_file", fname) + infer_span.set_tag("success", "true" if fname else "false") + + sentry_sdk.add_breadcrumb( + category="process", + message="infer方法调用完成", + level="info" + ) + + if fname is None: + with sentry_sdk.start_span(op="error.handling", description="处理文件缺失错误"): + raise RuntimeError("Output File not found") + + # 使用Span跟踪返回结果 + with sentry_sdk.start_span(op="result.preparation", description="准备返回结果"): + j = {"status": "success", "file_name": fname} + loguru.logger.success(j) + + # 设置事务状态为成功 + transaction.set_status("ok") + + return j, sentry_trace + + except Exception as infer_error: + # 使用原始异常对象 + sentry_sdk.add_breadcrumb( + category="error", + message=f"infer方法调用失败: {str(infer_error)}", + level="error" + ) + with sentry_sdk.start_span(op="error.handling", description="处理推理错误") as error_span: + error_span.set_data("error_type", type(infer_error).__name__) + error_span.set_data("error_message", str(infer_error)) + error_span.set_tag("error", "true") + raise infer_error + + except Exception as e: + # 异常处理 - 使用捕获的异常对象e + j = {"status": "fail", "msg": str(e)} + + try: + # 使用Span跟踪错误记录过程 + with sentry_sdk.start_span(op="error.logging", description="记录错误详情") as error_span: + # 更新日志文件 + with open(temp_log_path, "a", encoding="utf-8") as f: + # 记录异常详情 + f.write(f"\n====== 异常详情 ======\n") + f.write(f"异常类型: {type(e).__name__}\n") + f.write(f"异常信息: {str(e)}\n") + f.write(f"发生时间: {datetime.datetime.now()}\n") + f.write(f"会话ID: {self.session_id}\n") + f.write(f"调用ID: {function_call_id}\n") + + # 记录完整堆栈 + f.write("\n====== 堆栈跟踪 ======\n") + f.write(traceback.format_exc()) + + # 尝试添加ComfyUI日志 + with sentry_sdk.start_span(op="db.file.read", + description="读取ComfyUI日志") as log_span: + if os.path.exists("/root/comfy/ComfyUI/user/comfyui.log"): + try: + with open("/root/comfy/ComfyUI/user/comfyui.log", "r", + encoding="utf-8") as cf: + f.write("\n====== ComfyUI日志 ======\n") + f.write(cf.read()) + log_span.set_tag("success", "true") + except Exception as log_read_error: + f.write(f"\n无法读取ComfyUI日志: {str(log_read_error)}\n") + log_span.set_data("error", str(log_read_error)) + log_span.set_tag("success", "false") + else: + log_span.set_tag("file_exists", "false") + + error_span.set_data("log_path", temp_log_path) + error_span.set_tag("error_type", type(e).__name__) + + # 仅当尚未报告错误时才添加附件和报告异常 + if not error_reported: + with sentry_sdk.configure_scope() as scope: + with sentry_sdk.start_span(op="error.report", + description="向Sentry报告错误") as report_span: + scope.add_attachment( + path=temp_log_path, + filename=f"full_{function_call_id}.txt", + add_to_transactions=True + ) + + # 添加更多异常上下文 + scope.set_context("error_details", { + "error_type": type(e).__name__, + "error_message": str(e), + "time_of_error": datetime.datetime.now().isoformat() + }) + + # 设置事务状态为错误 + transaction.set_status("internal_error") + + # 使用异常对象e + sentry_sdk.capture_exception(e) + error_reported = True + print("已向Sentry报告异常") + report_span.set_tag("reported", "true") + + except Exception as log_error: + print(f"处理日志出错: {log_error}") + # 设置事务状态为错误 + transaction.set_status("internal_error") + + # 仅当尚未报告错误时才报告原始异常 + if not error_reported: + sentry_sdk.capture_exception(e) + print("已向Sentry报告原始异常（日志处理失败）") + + return j, sentry_trace + + finally: + # 使用Span跟踪清理过程 + with sentry_sdk.start_span(op="system.cleanup", description="清理资源") as cleanup_span: + # 清理临时文件和重启ComfyUI + try: + # 尝试复制日志到S3 + if temp_log_path and os.path.exists(temp_log_path): + with sentry_sdk.start_span(op="db.file.copy", description="复制日志到S3") as copy_span: + log_dir = f"/root/comfy/ComfyUI/output_s3/logs/{self.session_id}" + try: + os.makedirs(log_dir, exist_ok=True) + shutil.copy(temp_log_path, f"{log_dir}/full.txt") + print(f"已复制日志到: {log_dir}/full.txt") + copy_span.set_data("source", temp_log_path) + copy_span.set_data("destination", f"{log_dir}/full.txt") + copy_span.set_tag("success", "true") + except Exception as copy_error: + print(f"复制日志到S3失败: {copy_error}") + copy_span.set_data("error", str(copy_error)) + copy_span.set_tag("success", "false") + + # 清理临时文件 + with sentry_sdk.start_span(op="db.file.delete", description="删除临时文件") as del_span: + try: + os.unlink(temp_log_path) + print(f"已清理临时文件: {temp_log_path}") + del_span.set_data("file", temp_log_path) + del_span.set_tag("success", "true") + except Exception as unlink_error: + print(f"清理临时文件失败: {unlink_error}") + del_span.set_data("error", str(unlink_error)) + del_span.set_tag("success", "false") + except Exception as cleanup_error: + print(f"清理过程出错: {cleanup_error}") + cleanup_span.set_data("error", str(cleanup_error)) + cleanup_span.set_tag("success", "false") + + print("清理资源并重启ComfyUI") + with sentry_sdk.start_span(op="system.restart", description="重启ComfyUI") as restart_span: + try: + # 停止ComfyUI + with sentry_sdk.start_span(op="system.stop", description="停止ComfyUI") as stop_span: + cmd = "comfy stop" + subprocess.run(cmd, shell=True, check=True) + time.sleep(1) + stop_span.set_tag("success", "true") + except Exception as stop_error: + print(f"停止ComfyUI失败: {stop_error}") + restart_span.set_data("stop_error", str(stop_error)) + restart_span.set_tag("stop_success", "false") + + try: + # 重启ComfyUI + with sentry_sdk.start_span(op="system.start", description="启动ComfyUI") as start_span: + cmd = "comfy launch --background" + subprocess.run(cmd, shell=True, check=True) + start_span.set_tag("success", "true") + except Exception as start_error: + print(f"启动ComfyUI失败: {start_error}") + restart_span.set_data("start_error", str(start_error)) + restart_span.set_tag("start_success", "false") + modal.experimental.stop_fetching_inputs() + + + def poll_server_health(self): + import socket + import urllib + + try: + # dummy request to check if the server is healthy + req = urllib.request.Request("http://127.0.0.1:8188/system_stats") + urllib.request.urlopen(req, timeout=5) + print("ComfyUI server is healthy") + except (socket.timeout, urllib.error.URLError) as e: + # if no response in 5 seconds, stop the container; Modal will schedule queued inputs on a new container + print(f"Server health check failed: {str(e)} restarting ComfyUI...") + try: + cmd = "comfy stop" + try: + subprocess.run(cmd, shell=True, check=True) + except: + pass + cmd = "comfy launch --background" + try: + subprocess.run(cmd, shell=True, check=True) + except: + raise Exception("Failed to launch ComfyUI") + except: + modal.experimental.stop_fetching_inputs() + raise Exception("ComfyUI server is not healthy, restart failed, stopping container") \ No newline at end of file diff --git a/src/cluster/ffmpeg_app.py b/src/cluster/ffmpeg_app.py index f93d0fb..6aed804 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,10 +16,9 @@ app = modal.App( include_source=False) with ffmpeg_worker_image.imports(): - import shutil, psutil, os, backoff, json, sentry_sdk - from typing import List, Optional, Tuple, Dict, Any + import shutil, psutil, os, backoff, sentry_sdk + from typing import List, Optional, Tuple, Any from loguru import logger - import httpx from modal import current_function_call_id from ffmpeg.asyncio import FFmpeg from BowongModalFunctions.utils.SentryUtils import SentryUtils @@ -27,8 +26,7 @@ with ffmpeg_worker_image.imports(): from BowongModalFunctions.utils.VideoUtils import VideoUtils from BowongModalFunctions.models.ffmpeg_worker_model import FFMpegSliceSegment from BowongModalFunctions.models.media_model import MediaSources, MediaSource, MediaProtocol - from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify, WebhookMethodEnum, \ - FFMPEGConcatTaskStatusResponse, TaskStatus, ErrorCode + from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify from BowongModalFunctions.config import WorkerConfig config = WorkerConfig() diff --git a/src/cluster/video.py b/src/cluster/video.py index 8fa12b6..9206c88 100644 --- a/src/cluster/video.py +++ b/src/cluster/video.py @@ -5,7 +5,7 @@ downloader_image = ( modal.Image .debian_slim(python_version="3.11") .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') ) @@ -23,7 +23,7 @@ with downloader_image.imports(): import sentry_sdk from sentry_sdk.integrations.loguru import LoguruIntegration from tqdm import tqdm - from typing import Tuple, Dict, List + from typing import Tuple, List from loguru import logger from datetime import datetime, UTC, timedelta from modal import current_function_call_id @@ -107,7 +107,6 @@ with downloader_image.imports(): @app.function(cpu=1, timeout=1800, cloud="aws", - # region='ap-northeast', max_containers=config.video_downloader_concurrency, volumes={ "/mntS3": modal.CloudBucketMount( @@ -243,7 +242,7 @@ with downloader_image.imports(): "baggage": sentry_trace.x_baggage, }) as transaction: transaction.set_context("runtime_environment", { "MODAL_CLOUD_PROVIDER": os.environ.get('MODAL_CLOUD_PROVIDER', 'unknown'), - "MODAL_ENVIRONMENT": os.environ.get('MODAL_ENVIRONMENT', 'unknown'), + "MODAL_ENVIRONMENT": config.modal_environment, "MODAL_IMAGE_ID": os.environ.get('MODAL_IMAGE_ID', 'unknown'), "MODAL_IS_REMOTE": os.environ.get('MODAL_IS_REMOTE', 'unknown'), "MODAL_REGION": os.environ.get('MODAL_REGION', 'unknown'), diff --git a/src/cluster/web.py b/src/cluster/web.py index 8166fd3..7d55b6a 100644 --- a/src/cluster/web.py +++ b/src/cluster/web.py @@ -5,7 +5,7 @@ fastapi_image = ( modal.Image .debian_slim(python_version="3.11") .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') ) @@ -24,7 +24,6 @@ with fastapi_image.imports(): modal.Secret.from_name("cf-kv-secret", environment_name='dev'), ]) @modal.concurrent(max_inputs=100) - # @modal.asgi_app(custom_domains=["modal-dev.bowong.cc"]) @modal.asgi_app() def fastapi_webapp(): return web_app