diff --git a/.gitignore b/.gitignore index a23fe58..a7272ff 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ .pypirc .env -.idea \ No newline at end of file +.runtime.env +.idea +.venv +test* \ No newline at end of file diff --git a/.runtime.env b/.runtime.env index dbc5e5b..5229e3c 100644 --- a/.runtime.env +++ b/.runtime.env @@ -1,5 +1,5 @@ MODAL_ENVIRONMENT=dev -modal_app_name=bowong-ai-video +modal_app_name=cluster-test S3_mount_dir=/mntS3 S3_bucket_name=modal-media-cache S3_region=ap-northeast-2 diff --git a/src/BowongModalFunctions/api.py b/src/BowongModalFunctions/api.py index c7b16cb..8e07730 100644 --- a/src/BowongModalFunctions/api.py +++ b/src/BowongModalFunctions/api.py @@ -1,23 +1,11 @@ -import modal -from typing import Annotated -from fastapi import FastAPI, Response, Depends, Header, Request -from starlette import status +from fastapi import FastAPI, Request 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.ModalUtils import ModalUtils -from .models.media_model import MediaSource -from .models.web_model import (SentryTransactionInfo, - SentryTransactionHeader, - ModalTaskResponse, - ComfyTaskRequest, - ComfyTaskStatusResponse - ) -from .router import ffmpeg, cache +from .router import ffmpeg, cache, comfyui from .config import WorkerConfig config = WorkerConfig() @@ -93,100 +81,4 @@ async def scalar(): web_app.include_router(ffmpeg.router) web_app.include_router(cache.router) - - -@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, - item.webhook) - 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 ModalUtils.get_modal_task_status(task_id) - media = None - if transaction: - response.headers["x-trace-id"] = transaction.x_trace_id - response.headers["x-baggage"] = transaction.x_baggage - if task_status == "success": - task_status = result["status"] - if task_status != "success": - reason = result["msg"] - else: - media = MediaSource.from_str("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, - item.webhook) - 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 ModalUtils.get_modal_task_status(task_id) - media = None - - if transaction: - response.headers["x-trace-id"] = transaction.x_trace_id - response.headers["x-baggage"] = transaction.x_baggage - if task_status == "success": - task_status = result["status"] - if task_status != "success": - reason = result["msg"] - else: - media = MediaSource.from_str("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 "") +web_app.include_router(comfyui.router) diff --git a/src/BowongModalFunctions/models/media_model.py b/src/BowongModalFunctions/models/media_model.py index a697f7e..66c9b18 100644 --- a/src/BowongModalFunctions/models/media_model.py +++ b/src/BowongModalFunctions/models/media_model.py @@ -1,4 +1,5 @@ import os +import re from datetime import datetime from enum import Enum from functools import cached_property @@ -54,15 +55,33 @@ class MediaSource(BaseModel): endpoint=parsed_url.netloc, # domain of http url bucket=None, urn=media_url) - elif media_url.startswith('s3://'): # s3://{endpoint}/{bucket}/{url} + elif media_url.startswith('s3://'): + + pattern = r"^s3://[a-z]{2}-[a-z]+-\d/.*$" + if not re.match(pattern, media_url): + media_url = media_url.replace("s3://", f"s3://{s3_region}/") + # s3://{endpoint}/{bucket}/{url} paths = media_url[5:].split('/') + + if len(paths) < 3: raise ValidationError("URN-s3 格式错误") - return MediaSource(path='/'.join(paths[2:]), + + media_source = MediaSource(path='/'.join(paths[2:]), protocol=MediaProtocol.s3, endpoint=paths[0], bucket=paths[1], urn=media_url) + + s3_mount_path = config.S3_mount_dir + cache_path = os.path.join(s3_mount_path, media_source.cache_filepath) + + # 校验媒体文件是否存在缓存中 + if not os.path.exists(cache_path): + raise ValueError(f"媒体文件 {media_source.cache_filepath} 不存在于缓存中") + + return media_source + elif media_url.startswith('vod://'): # vod://{endpoint}/{subAppId}/{fileId} paths = media_url[6:].split('/') if len(paths) < 3: @@ -124,6 +143,7 @@ class MediaSource(BaseModel): case _: return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.path}" + @computed_field(description="文件后缀名") @cached_property def file_extension(self) -> Optional[str]: diff --git a/src/BowongModalFunctions/models/web_model.py b/src/BowongModalFunctions/models/web_model.py index f18c3c8..658062f 100644 --- a/src/BowongModalFunctions/models/web_model.py +++ b/src/BowongModalFunctions/models/web_model.py @@ -319,7 +319,7 @@ class ComfyTaskStatusResponse(BaseModel): class ComfyTaskRequest(BaseFFMPEGTaskRequest): video_path: MediaSource = Field( - default=MediaSource.from_str("s3://modal-media-cache/comfyui-input/markers/1016.mp4"), description="视频源") + default=None, 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") diff --git a/src/BowongModalFunctions/router/comfyui.py b/src/BowongModalFunctions/router/comfyui.py new file mode 100644 index 0000000..dbea7ca --- /dev/null +++ b/src/BowongModalFunctions/router/comfyui.py @@ -0,0 +1,127 @@ +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 ..models.media_model import MediaSource +from ..utils.ModalUtils import ModalUtils +from ..models.web_model import (SentryTransactionHeader, + ModalTaskResponse, SentryTransactionInfo, + ComfyTaskRequest, ComfyTaskStatusResponse) + +config = WorkerConfig() + +router = APIRouter(prefix="/comfyui") + +sentry_header_schema = { + "x-trace-id": { + "description": "Sentry Transaction ID", + "schema": { + "type": "string", + } + }, + "x-baggage": { + "description": "Sentry Transaction baggage", + "schema": { + "type": "string", + } + } +} + +@router.post("/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.replace(config.comfyui_s3_input+"/",""), item.start_time, item.filename_prefix, item.tts_text1, + item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace, + item.webhook) + return ModalTaskResponse(success=True, taskId=fn_call.object_id) + + +@router.get("/v1/{task_id}", tags=["ComfyUI"], summary="查询ComfyUIBase任务", + description="查询ComfyUIBase任务", + 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 ModalUtils.get_modal_task_status(task_id) + media = None + if transaction: + response.headers["x-trace-id"] = transaction.x_trace_id + response.headers["x-baggage"] = transaction.x_baggage + if task_status == "success": + task_status = result["status"] + if task_status != "success": + reason = result["msg"] + else: + media = MediaSource.from_str(result["file_name"]) + return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code, + error=reason, result=media.urn if media else None) + + +@router.post("/v2", tags=["ComfyUI"], summary="发起ComfyUILatentSync1.5任务", + description="发起ComfyUILatentSync1.5任务", + responses={ + status.HTTP_200_OK: { + "description": "", + "headers": sentry_header_schema + }, + }, + dependencies=[Depends(verify_token)] + ) +async def comfyui_v2(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]): + sentry_trace = None + if headers.x_trace_id and headers.x_baggage: + sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage) + cls = modal.Cls.from_name(config.modal_app_name, "ComfyUILatentSync15", environment_name=config.modal_environment) + fn_call = cls().api.spawn(item.video_path.path.replace(config.comfyui_s3_input+"/",""), item.start_time, item.filename_prefix, item.tts_text1, + item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace, + item.webhook) + return ModalTaskResponse(success=True, taskId=fn_call.object_id) + + +@router.get("/v2/{task_id}", tags=["ComfyUI"], summary="查询ComfyUILatentSync1.5任务", + description="查询ComfyUILatentSync1.5任务", + responses={ + status.HTTP_200_OK: { + "description": "", + "headers": sentry_header_schema + }, + }, + dependencies=[Depends(verify_token)] + ) +async def comfyui_v2_status(task_id: str, response: Response): + task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id) + media = None + + if transaction: + response.headers["x-trace-id"] = transaction.x_trace_id + response.headers["x-baggage"] = transaction.x_baggage + if task_status == "success": + task_status = result["status"] + if task_status != "success": + reason = result["msg"] + else: + media = MediaSource.from_str(result["file_name"]) + return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code, + error=reason, + result=media.urn if media else "") \ No newline at end of file diff --git a/src/BowongModalFunctions/utils/SentryUtils.py b/src/BowongModalFunctions/utils/SentryUtils.py index 599bc8d..f5b1817 100644 --- a/src/BowongModalFunctions/utils/SentryUtils.py +++ b/src/BowongModalFunctions/utils/SentryUtils.py @@ -55,6 +55,10 @@ class SentryUtils: status = TaskStatus.success error = None code = ErrorCode.SUCCESS.value + if not result: + status = TaskStatus.failed + error = "Internal Error" + code = ErrorCode.BUSINESS_ERROR.value except Exception as e: logger.exception(e) result = None @@ -72,9 +76,13 @@ class SentryUtils: code=code, results=[result] if isinstance(result, str) else result, ).model_dump()) - logger.info(f"webhook response = {response}") case WebhookMethodEnum.GET: - response = client.post(url=webhook.endpoint.__str__()) + response = client.get(url=webhook.endpoint.__str__(), params=BaseFFMPEGTaskStatusResponse( + taskId=func_id, status=status, error=error, + code=code, + results=[result] if isinstance(result, str) else result, + ).model_dump()) + logger.info(f"webhook response = {response}") response.raise_for_status() return result diff --git a/src/cluster/app.py b/src/cluster/app.py index eaa291f..344e116 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/comfyui_v1.py b/src/cluster/comfyui_v1.py index fa346fd..27e2d50 100644 --- a/src/cluster/comfyui_v1.py +++ b/src/cluster/comfyui_v1.py @@ -1,4 +1,5 @@ # ComfyUI模板--Base Auth +import backoff import modal from dotenv import dotenv_values @@ -11,8 +12,7 @@ comfyui_image = ( .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") + .pip_install("comfy_cli==0.0.0", index_url="https://packages-1747622887395:0ee15474ccd7b27b57ca63a9306327678e6c2631@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple") .run_commands("comfy --skip-prompt install --nvidia --version 0.3.10") .env(dotenv_values("../.runtime.env")) .workdir("/root/comfy") @@ -50,8 +50,6 @@ comfyui_image = ( ).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("BowongModalFunctions/template/template_v1.json", "/root/src/BowongModalFunctions/template/template_v1.json", copy=True) .add_local_python_source('BowongModalFunctions') .add_local_python_source('cluster') @@ -70,17 +68,12 @@ app.set_description("ComfyUI v1 Server") 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 - import datetime - import tempfile - import traceback from modal import current_function_call_id from BowongModalFunctions.config import WorkerConfig from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify @@ -90,6 +83,7 @@ with comfyui_image.imports(): 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) + custom_secret = modal.Secret.from_name("comfyui-custom-secret", environment_name=config.modal_environment) output_dir = "/root/comfy/ComfyUI/output" sentry_sdk.init( @@ -124,7 +118,7 @@ with comfyui_image.imports(): cpu=(2, 64), memory=(20480, 131072), enable_memory_snapshot=False, - secrets=[secret], + secrets=[secret, custom_secret], volumes={ "/root/comfy/ComfyUI/models": vol, "/root/comfy/ComfyUI/input_s3": modal.CloudBucketMount( @@ -151,20 +145,25 @@ with comfyui_image.imports(): @modal.method() async 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, webhook:WebhookNotify = None): + tts_text1: str = "测试", tts_text2: str = "", tts_text3: str = "", tts_text4: str = "", + anchor_id: str = "dawan", speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None, + webhook: WebhookNotify = None): + + # 导入必要模块 + import datetime + import tempfile + import traceback + from loguru import logger # 生成当前函数调用的唯一ID function_call_id = current_function_call_id() - print(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}") + logger.info(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}") @SentryUtils.webhook_handler(webhook=webhook, func_id=function_call_id) - async def infer(workflow: Dict): + async def infer(workflow: Dict) -> str | None: self.poll_server_health() self.prompt_uuid = str(uuid.uuid4()) - print("Workflow JSON:") - print(json.dumps(workflow, indent=4, ensure_ascii=False)) + logger.info(f"Workflow JSON: \n{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": @@ -175,13 +174,16 @@ with comfyui_image.imports(): shutil.copy(file_to_move, file_to_move.replace("input_s3", "input")) except: try: - print("Try download file from S3 manually") + logger.warning("尝试手动从S3下载文件") # 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) + if "aws_key_id" in list(os.environ.keys()): + yaml_config = os.environ + else: + 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, @@ -196,15 +198,15 @@ with comfyui_image.imports(): 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"] + 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: @@ -216,20 +218,23 @@ with comfyui_image.imports(): os.path.join(output_dir.replace("output", "output_s3"), f)) except: try: - print("Try move file to S3 manually") + logger.warning("尝试手动将文件移动到S3") # 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) + if "aws_key_id" in list(os.environ.keys()): + yaml_config = os.environ + else: + 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 + return "s3://" + "/".join([config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, f]) # 使用临时文件 temp_log_path = None @@ -237,31 +242,23 @@ with comfyui_image.imports(): # 创建主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, }) + 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()) + 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 - }) + sentry_sdk.set_tag("session_id", self.session_id) + sentry_sdk.set_tag("function_call_id", function_call_id) + sentry_sdk.set_tag("video_path", video_path) + sentry_sdk.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跟踪日志文件创建 @@ -269,7 +266,7 @@ with comfyui_image.imports(): # 创建临时日志文件 temp_log_file = tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt') temp_log_path = temp_log_file.name - print(f"创建临时日志文件: {temp_log_path}") + logger.debug(f"创建临时日志文件: {temp_log_path}") span.set_data("log_path", temp_log_path) span.set_tag("file_operation", "create") @@ -283,6 +280,9 @@ with comfyui_image.imports(): temp_log_file.flush() temp_log_file.close() + # 设置状态为成功 + span.set_status("ok") + # 使用Span跟踪模板处理 with sentry_sdk.start_span(op="template.json", description="处理工作流模板") as span: # 正常业务逻辑 @@ -307,17 +307,17 @@ with comfyui_image.imports(): modify_span.set_data("modified_nodes", ["211", "399", "331", "406", "424", "225", "330", "515"]) + # 设置状态为成功 + modify_span.set_status("ok") span.set_data("workflow_size", len(workflow_file)) span.set_tag("workflow_type", "video_processing") + # 设置状态为成功 + span.set_status("ok") try: # 记录执行的关键步骤作为breadcrumbs - sentry_sdk.add_breadcrumb( - category="process", - message="开始调用infer方法", - level="info" - ) + sentry_sdk.add_breadcrumb(category="process", message="开始调用infer方法", level="info") # 使用Span跟踪推理过程 with sentry_sdk.start_span(op="ai.inference", description="执行模型推理") as infer_span: @@ -328,6 +328,8 @@ with comfyui_image.imports(): 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") + # 设置状态为成功 + prep_span.set_status("ok") # 调用推理方法 fname = await infer(workflow_file) @@ -337,6 +339,8 @@ with comfyui_image.imports(): description="处理推理结果") as post_span: post_span.set_data("output_file", fname) post_span.set_data("timestamp", datetime.datetime.now().isoformat()) + # 设置状态为成功 + post_span.set_status("ok") infer_end_time = datetime.datetime.now() duration_seconds = (infer_end_time - infer_start_time).total_seconds() @@ -346,21 +350,24 @@ with comfyui_image.imports(): 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") + # 设置状态为成功 + infer_span.set_status("ok") - sentry_sdk.add_breadcrumb( - category="process", - message="infer方法调用完成", - level="info" - ) + sentry_sdk.add_breadcrumb(category="process", message="infer方法调用完成", level="info") if fname is None: - with sentry_sdk.start_span(op="error.handling", description="处理文件缺失错误"): + with sentry_sdk.start_span(op="error.handling", + description="处理文件缺失错误") as error_span: + error_span.set_status("internal_error") # 设置状态为错误 raise RuntimeError("Output File not found") # 使用Span跟踪返回结果 - with sentry_sdk.start_span(op="result.preparation", description="准备返回结果"): + with sentry_sdk.start_span(op="result.preparation", description="准备返回结果") as result_span: j = {"status": "success", "file_name": fname} - loguru.logger.success(j) + logger.success(f"处理成功: {j}") + + # 设置span状态为成功 + result_span.set_status("ok") # 设置事务状态为成功 transaction.set_status("ok") @@ -369,20 +376,21 @@ with comfyui_image.imports(): except Exception as infer_error: # 使用原始异常对象 - sentry_sdk.add_breadcrumb( - category="error", - message=f"infer方法调用失败: {str(infer_error)}", - level="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") + # 设置状态为错误 + error_span.set_status("internal_error") + logger.error(f"推理失败: {str(infer_error)}") raise infer_error except Exception as e: # 异常处理 - 使用捕获的异常对象e - j = {"status": "fail", "msg": str(e)} + j = {"status": "failed", "msg": str(e)} + logger.error(f"处理异常: {str(e)}") try: # 使用Span跟踪错误记录过程 @@ -411,58 +419,67 @@ with comfyui_image.imports(): f.write("\n====== ComfyUI日志 ======\n") f.write(cf.read()) log_span.set_tag("success", "true") + log_span.set_status("ok") # 设置状态为成功 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") + # 设置状态为错误 + log_span.set_status("internal_error") + logger.warning(f"无法读取ComfyUI日志: {str(log_read_error)}") else: log_span.set_tag("file_exists", "false") + # 设置状态为通知 + log_span.set_status("unavailable") + logger.warning("ComfyUI日志文件不存在") error_span.set_data("log_path", temp_log_path) error_span.set_tag("error_type", type(e).__name__) + # 设置状态为错误 + error_span.set_status("internal_error") # 仅当尚未报告错误时才添加附件和报告异常 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 - ) + with sentry_sdk.start_span(op="error.report", + description="向Sentry报告错误") as report_span: + # 使用新的API替换configure_scope + # 添加附件 + sentry_sdk.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() - }) + # 添加更多异常上下文 + sentry_sdk.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") + # 设置事务状态为错误 + transaction.set_status("internal_error") + # 设置span状态为错误 + report_span.set_status("internal_error") - # 使用异常对象e - sentry_sdk.capture_exception(e) - error_reported = True - print("已向Sentry报告异常") - report_span.set_tag("reported", "true") + # 使用异常对象e + sentry_sdk.capture_exception(e) + error_reported = True + logger.info("已向Sentry报告异常") + report_span.set_tag("reported", "true") except Exception as log_error: - print(f"处理日志出错: {log_error}") + logger.error(f"处理日志出错: {log_error}") # 设置事务状态为错误 transaction.set_status("internal_error") # 仅当尚未报告错误时才报告原始异常 if not error_reported: sentry_sdk.capture_exception(e) - print("已向Sentry报告原始异常（日志处理失败）") + logger.info("已向Sentry报告原始异常（日志处理失败）") return j, sentry_trace finally: # 使用Span跟踪清理过程 with sentry_sdk.start_span(op="system.cleanup", description="清理资源") as cleanup_span: + cleanup_successful = True # 跟踪清理是否成功的标志 + # 清理临时文件和重启ComfyUI try: # 尝试复制日志到S3 @@ -472,56 +489,47 @@ with comfyui_image.imports(): try: os.makedirs(log_dir, exist_ok=True) shutil.copy(temp_log_path, f"{log_dir}/full.txt") - print(f"已复制日志到: {log_dir}/full.txt") + logger.info(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") + # 设置状态为成功 + copy_span.set_status("ok") except Exception as copy_error: - print(f"复制日志到S3失败: {copy_error}") + logger.error(f"复制日志到S3失败: {copy_error}") copy_span.set_data("error", str(copy_error)) copy_span.set_tag("success", "false") + # 设置状态为错误 + copy_span.set_status("internal_error") + cleanup_successful = 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}") + logger.debug(f"已清理临时文件: {temp_log_path}") del_span.set_data("file", temp_log_path) del_span.set_tag("success", "true") + # 设置状态为成功 + del_span.set_status("ok") except Exception as unlink_error: - print(f"清理临时文件失败: {unlink_error}") + logger.error(f"清理临时文件失败: {unlink_error}") del_span.set_data("error", str(unlink_error)) del_span.set_tag("success", "false") + # 设置状态为错误 + del_span.set_status("internal_error") + cleanup_successful = False except Exception as cleanup_error: - print(f"清理过程出错: {cleanup_error}") + logger.error(f"清理过程出错: {cleanup_error}") cleanup_span.set_data("error", str(cleanup_error)) cleanup_span.set_tag("success", "false") + cleanup_successful = 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() + # 根据清理和重启结果设置总体清理状态 + if cleanup_successful: + cleanup_span.set_status("ok") + else: + cleanup_span.set_status("internal_error") def poll_server_health(self): @@ -529,7 +537,6 @@ with comfyui_image.imports(): 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") diff --git a/src/cluster/comfyui_v2.py b/src/cluster/comfyui_v2.py index 632b775..d445b5f 100644 --- a/src/cluster/comfyui_v2.py +++ b/src/cluster/comfyui_v2.py @@ -3,20 +3,16 @@ import modal from dotenv import dotenv_values comfyui_latentsync_1_5_image = ( - modal.Image.debian_slim( - python_version="3.10" - ) + 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") + "&& 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") + .pip_install("comfy_cli==0.0.0",index_url="https://packages-1747622887395:0ee15474ccd7b27b57ca63a9306327678e6c2631@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple") .run_commands("comfy --skip-prompt install --nvidia --version 0.3.10") .env(dotenv_values("../.runtime.env")) - .workdir("/root/comfy") -) + .workdir("/root/comfy")) comfyui_latentsync_1_5_image = ( comfyui_latentsync_1_5_image.run_commands("comfy node install https://github.com/M1kep/ComfyLiterals") @@ -37,44 +33,29 @@ comfyui_latentsync_1_5_image = ( .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("BowongModalFunctions/template/template_v2.json", "/root/src/BowongModalFunctions/template/template_v2.json", copy=True) + .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("BowongModalFunctions/template/template_v2.json","/root/src/BowongModalFunctions/template/template_v2.json",copy=True) .add_local_python_source('cluster') - .add_local_python_source('BowongModalFunctions') -) + .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 - import datetime - import tempfile - import traceback + from loguru import logger from BowongModalFunctions.config import WorkerConfig from modal import current_function_call_id from sentry_sdk.integrations.loguru import LoguruIntegration @@ -84,6 +65,7 @@ with comfyui_latentsync_1_5_image.imports(): 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) + custom_secret = modal.Secret.from_name("comfyui-custom-secret", environment_name=config.modal_environment) output_dir = "/root/comfy/ComfyUI/output" sentry_sdk.init( @@ -106,6 +88,7 @@ with comfyui_latentsync_1_5_image.imports(): return (cpu_freq, cpu_count, mem) + @app.cls( max_containers=200, min_containers=0, @@ -113,49 +96,46 @@ with comfyui_latentsync_1_5_image.imports(): scaledown_window=120, timeout=1200, gpu=["L4"], - cpu=(2,64), - memory=(20480,131072), + 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+"/" - ), - }, - ) + secrets=[secret, custom_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) + 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() async 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, webhook:WebhookNotify = None): + tts_text1: str = "测试", tts_text2: str = "", tts_text3: str = "", tts_text4: str = "", + anchor_id: str = "dawan", speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None, + webhook: WebhookNotify = None): + + # 导入必要模块 + import datetime + import tempfile + import traceback + from loguru import logger # 生成当前函数调用的唯一ID function_call_id = current_function_call_id() - print(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}") + logger.info(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}") @SentryUtils.webhook_handler(webhook=webhook, func_id=function_call_id) - async def infer(workflow: Dict): + async def infer(workflow: Dict) -> str | None: self.poll_server_health() self.prompt_uuid = str(uuid.uuid4()) - print("Workflow JSON:") - print(json.dumps(workflow, indent=4, ensure_ascii=False)) + logger.info(f"Workflow JSON: \n{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": @@ -166,13 +146,16 @@ with comfyui_latentsync_1_5_image.imports(): shutil.copy(file_to_move, file_to_move.replace("input_s3", "input")) except: try: - print("Try download file to S3 manually") + logger.warning("尝试手动从S3下载文件") # 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) + if "aws_key_id" in list(os.environ.keys()): + yaml_config = os.environ + else: + 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, @@ -187,11 +170,8 @@ with comfyui_latentsync_1_5_image.imports(): 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"] + 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) @@ -208,20 +188,23 @@ with comfyui_latentsync_1_5_image.imports(): os.path.join(output_dir.replace("output", "output_s3"), f)) except: try: - print("Try move file to S3 manually") + logger.warning("尝试手动将文件移动到S3") # 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) + if "aws_key_id" in list(os.environ.keys()): + yaml_config = os.environ + else: + 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 + return "s3://" + "/".join([config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, f]) # 使用临时文件 temp_log_path = None @@ -229,31 +212,22 @@ with comfyui_latentsync_1_5_image.imports(): # 创建主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, }) + 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()) + 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 - }) + sentry_sdk.set_tag("session_id", self.session_id) + sentry_sdk.set_tag("function_call_id", function_call_id) + sentry_sdk.set_tag("video_path", video_path) + sentry_sdk.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跟踪日志文件创建 @@ -261,7 +235,7 @@ with comfyui_latentsync_1_5_image.imports(): # 创建临时日志文件 temp_log_file = tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt') temp_log_path = temp_log_file.name - print(f"创建临时日志文件: {temp_log_path}") + logger.debug(f"创建临时日志文件: {temp_log_path}") span.set_data("log_path", temp_log_path) span.set_tag("file_operation", "create") @@ -275,6 +249,9 @@ with comfyui_latentsync_1_5_image.imports(): temp_log_file.flush() temp_log_file.close() + # 设置状态为成功 + span.set_status("ok") + # 使用Span跟踪模板处理 with sentry_sdk.start_span(op="template.json", description="处理工作流模板") as span: # 正常业务逻辑 @@ -290,8 +267,7 @@ with comfyui_latentsync_1_5_image.imports(): 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["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 @@ -299,17 +275,17 @@ with comfyui_latentsync_1_5_image.imports(): modify_span.set_data("modified_nodes", ["211", "399", "331", "406", "424", "225", "330", "515"]) + # 设置状态为成功 + modify_span.set_status("ok") span.set_data("workflow_size", len(workflow_file)) span.set_tag("workflow_type", "video_processing") + # 设置状态为成功 + span.set_status("ok") try: # 记录执行的关键步骤作为breadcrumbs - sentry_sdk.add_breadcrumb( - category="process", - message="开始调用infer方法", - level="info" - ) + sentry_sdk.add_breadcrumb(category="process", message="开始调用infer方法", level="info") # 使用Span跟踪推理过程 with sentry_sdk.start_span(op="ai.inference", description="执行模型推理") as infer_span: @@ -320,6 +296,8 @@ with comfyui_latentsync_1_5_image.imports(): 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") + # 设置状态为成功 + prep_span.set_status("ok") # 调用推理方法 fname = await infer(workflow_file) @@ -329,6 +307,8 @@ with comfyui_latentsync_1_5_image.imports(): description="处理推理结果") as post_span: post_span.set_data("output_file", fname) post_span.set_data("timestamp", datetime.datetime.now().isoformat()) + # 设置状态为成功 + post_span.set_status("ok") infer_end_time = datetime.datetime.now() duration_seconds = (infer_end_time - infer_start_time).total_seconds() @@ -338,21 +318,24 @@ with comfyui_latentsync_1_5_image.imports(): 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") + # 设置状态为成功 + infer_span.set_status("ok") - sentry_sdk.add_breadcrumb( - category="process", - message="infer方法调用完成", - level="info" - ) + sentry_sdk.add_breadcrumb(category="process", message="infer方法调用完成", level="info") if fname is None: - with sentry_sdk.start_span(op="error.handling", description="处理文件缺失错误"): + with sentry_sdk.start_span(op="error.handling", + description="处理文件缺失错误") as error_span: + error_span.set_status("internal_error") # 设置状态为错误 raise RuntimeError("Output File not found") # 使用Span跟踪返回结果 - with sentry_sdk.start_span(op="result.preparation", description="准备返回结果"): + with sentry_sdk.start_span(op="result.preparation", description="准备返回结果") as result_span: j = {"status": "success", "file_name": fname} - loguru.logger.success(j) + logger.success(f"处理成功: {j}") + + # 设置span状态为成功 + result_span.set_status("ok") # 设置事务状态为成功 transaction.set_status("ok") @@ -361,20 +344,21 @@ with comfyui_latentsync_1_5_image.imports(): except Exception as infer_error: # 使用原始异常对象 - sentry_sdk.add_breadcrumb( - category="error", - message=f"infer方法调用失败: {str(infer_error)}", - level="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") + # 设置状态为错误 + error_span.set_status("internal_error") + logger.error(f"推理失败: {str(infer_error)}") raise infer_error except Exception as e: # 异常处理 - 使用捕获的异常对象e - j = {"status": "fail", "msg": str(e)} + j = {"status": "failed", "msg": str(e)} + logger.error(f"处理异常: {str(e)}") try: # 使用Span跟踪错误记录过程 @@ -403,58 +387,67 @@ with comfyui_latentsync_1_5_image.imports(): f.write("\n====== ComfyUI日志 ======\n") f.write(cf.read()) log_span.set_tag("success", "true") + log_span.set_status("ok") # 设置状态为成功 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") + # 设置状态为错误 + log_span.set_status("internal_error") + logger.warning(f"无法读取ComfyUI日志: {str(log_read_error)}") else: log_span.set_tag("file_exists", "false") + # 设置状态为通知 + log_span.set_status("unavailable") + logger.warning("ComfyUI日志文件不存在") error_span.set_data("log_path", temp_log_path) error_span.set_tag("error_type", type(e).__name__) + # 设置状态为错误 + error_span.set_status("internal_error") # 仅当尚未报告错误时才添加附件和报告异常 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 - ) + with sentry_sdk.start_span(op="error.report", + description="向Sentry报告错误") as report_span: + # 使用新的API替换configure_scope + # 添加附件 + sentry_sdk.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() - }) + # 添加更多异常上下文 + sentry_sdk.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") + # 设置事务状态为错误 + transaction.set_status("internal_error") + # 设置span状态为错误 + report_span.set_status("internal_error") - # 使用异常对象e - sentry_sdk.capture_exception(e) - error_reported = True - print("已向Sentry报告异常") - report_span.set_tag("reported", "true") + # 使用异常对象e + sentry_sdk.capture_exception(e) + error_reported = True + logger.info("已向Sentry报告异常") + report_span.set_tag("reported", "true") except Exception as log_error: - print(f"处理日志出错: {log_error}") + logger.error(f"处理日志出错: {log_error}") # 设置事务状态为错误 transaction.set_status("internal_error") # 仅当尚未报告错误时才报告原始异常 if not error_reported: sentry_sdk.capture_exception(e) - print("已向Sentry报告原始异常（日志处理失败）") + logger.info("已向Sentry报告原始异常（日志处理失败）") return j, sentry_trace finally: # 使用Span跟踪清理过程 with sentry_sdk.start_span(op="system.cleanup", description="清理资源") as cleanup_span: + cleanup_successful = True # 跟踪清理是否成功的标志 + # 清理临时文件和重启ComfyUI try: # 尝试复制日志到S3 @@ -464,56 +457,47 @@ with comfyui_latentsync_1_5_image.imports(): try: os.makedirs(log_dir, exist_ok=True) shutil.copy(temp_log_path, f"{log_dir}/full.txt") - print(f"已复制日志到: {log_dir}/full.txt") + logger.info(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") + # 设置状态为成功 + copy_span.set_status("ok") except Exception as copy_error: - print(f"复制日志到S3失败: {copy_error}") + logger.error(f"复制日志到S3失败: {copy_error}") copy_span.set_data("error", str(copy_error)) copy_span.set_tag("success", "false") + # 设置状态为错误 + copy_span.set_status("internal_error") + cleanup_successful = 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}") + logger.debug(f"已清理临时文件: {temp_log_path}") del_span.set_data("file", temp_log_path) del_span.set_tag("success", "true") + # 设置状态为成功 + del_span.set_status("ok") except Exception as unlink_error: - print(f"清理临时文件失败: {unlink_error}") + logger.error(f"清理临时文件失败: {unlink_error}") del_span.set_data("error", str(unlink_error)) del_span.set_tag("success", "false") + # 设置状态为错误 + del_span.set_status("internal_error") + cleanup_successful = False except Exception as cleanup_error: - print(f"清理过程出错: {cleanup_error}") + logger.error(f"清理过程出错: {cleanup_error}") cleanup_span.set_data("error", str(cleanup_error)) cleanup_span.set_tag("success", "false") + cleanup_successful = 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() + # 根据清理和重启结果设置总体清理状态 + if cleanup_successful: + cleanup_span.set_status("ok") + else: + cleanup_span.set_status("internal_error") def poll_server_health(self): @@ -524,10 +508,10 @@ with comfyui_latentsync_1_5_image.imports(): # 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") + logger.info("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...") + logger.error(f"Server health check failed: {str(e)} restarting ComfyUI...") try: cmd = "comfy stop" try: @@ -541,4 +525,4 @@ with comfyui_latentsync_1_5_image.imports(): 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 + raise Exception("ComfyUI server is not healthy, restart failed, stopping container") diff --git a/src/cluster/web.py b/src/cluster/web.py index 1823060..75d7842 100644 --- a/src/cluster/web.py +++ b/src/cluster/web.py @@ -1,10 +1,6 @@ import modal from dotenv import dotenv_values -from BowongModalFunctions.config import WorkerConfig - -config = WorkerConfig() - fastapi_image = ( modal.Image .debian_slim(python_version="3.11") @@ -17,12 +13,13 @@ fastapi_image = ( app = modal.App( name="web_app", image=fastapi_image, - secrets=[modal.Secret.from_name("aws-s3-secret", environment_name=config.modal_environment)], include_source=False) with fastapi_image.imports(): from BowongModalFunctions.api import web_app + from BowongModalFunctions.config import WorkerConfig + config = WorkerConfig() @app.function(scaledown_window=60, secrets=[ @@ -34,7 +31,8 @@ with fastapi_image.imports(): secret=modal.Secret.from_name("aws-s3-secret", environment_name=config.modal_environment), ), - }, ) + }, + ) @modal.concurrent(max_inputs=100) @modal.asgi_app() def fastapi_webapp():