Compare commits

...
Author SHA1 Message Date
bennykok 61acffd355 fix 2024-09-18 08:20:35 -07:00
bennykok aa47f3523f fix 2024-09-17 23:36:24 -07:00
bennykok 7ed4284a6f fix 2024-09-17 23:25:19 -07:00
bennykok a403daa314 fix 2024-09-17 23:09:42 -07:00
bennykok ba9b187dcc fix 2024-09-17 22:59:27 -07:00
bennykok 1243fa4e58 fix 2024-09-17 22:55:08 -07:00
bennykok 0d1537963c fix 2024-09-17 21:48:42 -07:00
bennykok 0083b38dcc chore: log image size 2024-09-17 20:44:44 -07:00
bennykok b8dded1535 Revert "fix: roll back to unique session per request"
This reverts commit 5a78ca97bd.
2024-09-17 20:26:39 -07:00
bennykok 4927d81e73 chore: accept cd_token 2024-09-17 18:57:15 -07:00
bennykok fb6bb2357a Reapply "fix: back to sequential file upload"
This reverts commit 1f5a88b888.
2024-09-17 14:28:56 -07:00
bennykok 086d642360 Merge branch 'benny/log-sync' into public-main 2024-09-17 14:27:59 -07:00
bennykok 212daa838c Revert "feat: experiment with await + asyncio.gather for multi file in same node"
This reverts commit c08b68c41f.
2024-09-17 14:25:13 -07:00
bennykok c08b68c41f feat: experiment with await + asyncio.gather for multi file in same node 2024-09-17 12:56:42 -07:00
bennykok 5a78ca97bd fix: roll back to unique session per request 2024-09-16 23:57:40 -07:00
bennykok 1f5a88b888 Revert "fix: back to sequential file upload"
This reverts commit 3d099f88ea.
2024-09-16 23:55:16 -07:00
bennykok 946571e32e fix: await 2024-09-16 18:54:05 -07:00
bennykok e692beb009 feat: realtime log sync 2024-09-16 15:34:20 -07:00
bennykok 3d099f88ea fix: back to sequential file upload 2024-09-16 13:55:02 -07:00
karrix 65f7576748 fix: non type error when upload output 2024-09-16 12:45:53 -07:00
bennykok 2d72cd8175 fix: batch zip image input 2024-09-14 21:49:17 -07:00
bennykok 5554c95f44 Merge branch 'benny/auth_token' into public-main 2024-09-12 14:14:16 -07:00
bennykok c1003f7e31 Merge branch 'benny/zip-batch-image' into public-main 2024-09-12 14:14:08 -07:00
EdwinWong 71d60a5dd1 fix: comfydeploy node backward compatible in every comfyui 2024-09-10 01:03:50 -07:00
bennykok e011711600 feat: zip batch image support 2024-09-09 17:49:39 -07:00
bennykok 4df9d38e56 feat: embed file public status into image output 2024-09-03 23:07:48 -07:00
bennykok 9cd626e1f6 feat: send token for cd update api 2024-09-03 21:58:39 -07:00
bennykok 503dca8fb6 chore: add log 2024-08-30 12:16:41 -07:00
bennykok 73c149b4cb fix node meta 2024-08-30 12:16:41 -07:00
bennykok 65b5b0b8c7 fix: remove content length 2024-08-30 12:16:41 -07:00
bennykok 9d6ee85402 fix: upload file acl 2024-08-30 12:16:41 -07:00
bennykok cdaed8a571 fix: include upload time 2024-08-30 12:16:41 -07:00
bennykok 3129e89cce fix: log file error log 2024-08-30 12:16:41 -07:00
bennykok 7a693eabc8 fix: size 2024-08-30 12:16:41 -07:00
bennykok 8f677e520d chore: log more test for upload file debug 2024-08-30 12:16:41 -07:00
bennykok 4c8d32c5b0 fix 2024-08-30 12:16:41 -07:00
nick a99d2568e0 video and lora node fix 2024-08-28 13:08:15 -07:00
6 changed files with 336 additions and 136 deletions
+22 -2
View File
@@ -39,14 +39,34 @@ class ComfyUIDeployExternalImageBatch:
CATEGORY = "image" CATEGORY = "image"
def process_image(self, image):
image = ImageOps.exif_transpose(image)
image = image.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image_tensor = torch.from_numpy(image)[None,]
return image_tensor
def run(self, input_id, images=None, default_value=None, display_name=None, description=None): def run(self, input_id, images=None, default_value=None, display_name=None, description=None):
import requests
import zipfile
import io
processed_images = [] processed_images = []
try: try:
images_list = json.loads(images) # Assuming images is a JSON array string images_list = json.loads(images) # Assuming images is a JSON array string
print(images_list) print(images_list)
for img_input in images_list: for img_input in images_list:
if img_input.startswith('http'): if img_input.startswith('http') and img_input.endswith('.zip'):
import requests print("Fetching zip file from url: ", img_input)
response = requests.get(img_input)
zip_file = zipfile.ZipFile(io.BytesIO(response.content))
for file_name in zip_file.namelist():
if file_name.lower().endswith(('.png', '.jpg', '.jpeg')):
with zip_file.open(file_name) as file:
image = Image.open(file)
image = self.process_image(image)
processed_images.append(image)
elif img_input.startswith('http'):
from io import BytesIO from io import BytesIO
print("Fetching image from url: ", img_input) print("Fetching image from url: ", img_input)
response = requests.get(img_input) response = requests.get(img_input)
+13 -5
View File
@@ -25,10 +25,10 @@ class ComfyUIDeployExternalLora:
}, },
"optional": { "optional": {
"default_lora_name": (folder_paths.get_filename_list("loras"),), "default_lora_name": (folder_paths.get_filename_list("loras"),),
"lora_save_name": ( # if `default_lora_name` is a link to download a file, we will attempt to save it with this name "lora_save_name": ( # if `default_lora_name` is a link to download a file, we will attempt to save it with this name
"STRING", "STRING",
{"multiline": False, "default": ""}, {"multiline": False, "default": ""},
), ),
"display_name": ( "display_name": (
"STRING", "STRING",
{"multiline": False, "default": ""}, {"multiline": False, "default": ""},
@@ -51,7 +51,15 @@ class ComfyUIDeployExternalLora:
CATEGORY = "deploy" CATEGORY = "deploy"
def run(self, input_id, default_lora_name=None, lora_save_name=None, display_name=None, description=None, lora_url=None): def run(
self,
input_id,
default_lora_name=None,
lora_save_name=None,
display_name=None,
description=None,
lora_url=None,
):
import requests import requests
import os import os
import uuid import uuid
@@ -71,9 +79,9 @@ class ComfyUIDeployExternalLora:
folder_paths.folder_names_and_paths["loras"][0][0], lora_save_name folder_paths.folder_names_and_paths["loras"][0][0], lora_save_name
) )
print(destination_path) print(destination_path)
print("Downloading external lora - " + input_id + " to " + destination_path) print("Downloading external lora - " + lora_url + " to " + destination_path)
response = requests.get( response = requests.get(
input_id, lora_url,
headers={"User-Agent": "Mozilla/5.0"}, headers={"User-Agent": "Mozilla/5.0"},
allow_redirects=True, allow_redirects=True,
) )
+1 -1
View File
@@ -834,7 +834,7 @@ class ComfyUIDeployExternalVideo:
): ):
out_file.write(chunk) out_file.write(chunk)
else: else:
video = kwargs.get("default_video", "") video = kwargs.get("default_video", None)
if video is None: if video is None:
raise "No default video given and no external video provided" raise "No default video given and no external video provided"
video_path = folder_paths.get_annotated_filepath(video.strip('"')) video_path = folder_paths.get_annotated_filepath(video.strip('"'))
+209 -40
View File
@@ -1,4 +1,5 @@
from io import BytesIO from io import BytesIO
from pprint import pprint
from aiohttp import web from aiohttp import web
import os import os
import requests import requests
@@ -23,7 +24,7 @@ from typing import Dict, List, Union, Any, Optional
from PIL import Image from PIL import Image
import copy import copy
import struct import struct
from aiohttp import web, ClientSession, ClientError, ClientTimeout from aiohttp import web, ClientSession, ClientError, ClientTimeout, ClientResponseError
import atexit import atexit
# Global session # Global session
@@ -56,39 +57,65 @@ retry_delay_multiplier = float(os.environ.get('RETRY_DELAY_MULTIPLIER', '2'))
print(f"max_retries: {max_retries}, retry_delay_multiplier: {retry_delay_multiplier}") print(f"max_retries: {max_retries}, retry_delay_multiplier: {retry_delay_multiplier}")
async def async_request_with_retry(method, url, disable_timeout=False, **kwargs): import time
async def async_request_with_retry(method, url, disable_timeout=False, token=None, **kwargs):
global client_session global client_session
await ensure_client_session() await ensure_client_session()
# async with aiohttp.ClientSession() as client_session:
retry_delay = 1 # Start with 1 second delay retry_delay = 1 # Start with 1 second delay
initial_timeout = 5 # 5 seconds timeout for the initial connection initial_timeout = 5 # 5 seconds timeout for the initial connection
start_time = time.time()
for attempt in range(max_retries): for attempt in range(max_retries):
try: try:
# Set a timeout for the initial connection
if not disable_timeout: if not disable_timeout:
timeout = ClientTimeout(total=None, connect=initial_timeout) timeout = ClientTimeout(total=None, connect=initial_timeout)
kwargs['timeout'] = timeout kwargs['timeout'] = timeout
if token is not None:
if 'headers' not in kwargs:
kwargs['headers'] = {}
kwargs['headers']['Authorization'] = f"Bearer {token}"
request_start = time.time()
async with client_session.request(method, url, **kwargs) as response: async with client_session.request(method, url, **kwargs) as response:
request_end = time.time()
logger.info(f"Request attempt {attempt + 1} took {request_end - request_start:.2f} seconds")
if response.status != 200:
error_body = await response.text()
logger.error(f"Request failed with status {response.status} and body {error_body}")
# raise Exception(f"Request failed with status {response.status}")
response.raise_for_status() response.raise_for_status()
if method.upper() == 'GET': if method.upper() == 'GET':
await response.read() await response.read()
total_time = time.time() - start_time
logger.info(f"Request succeeded after {total_time:.2f} seconds (attempt {attempt + 1}/{max_retries})")
return response return response
except asyncio.TimeoutError: except asyncio.TimeoutError:
logger.warning(f"Request timed out after {initial_timeout} seconds (attempt {attempt + 1}/{max_retries})") logger.warning(f"Request timed out after {initial_timeout} seconds (attempt {attempt + 1}/{max_retries})")
except ClientError as e: except ClientError as e:
end_time = time.time()
logger.error(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}")
logger.error(f"Time taken for failed attempt: {end_time - request_start:.2f} seconds")
logger.error(f"Total time elapsed: {end_time - start_time:.2f} seconds")
# Log the response body for ClientError as well
if hasattr(e, 'response') and e.response is not None:
error_body = await e.response.text()
logger.error(f"Error response body: {error_body}")
if attempt == max_retries - 1: if attempt == max_retries - 1:
logger.error(f"Request failed after {max_retries} attempts: {e}") logger.error(f"Request failed after {max_retries} attempts: {e}")
# raise raise
logger.warning(f"Request failed (attempt {attempt + 1}/{max_retries}): {e}")
# Wait before retrying
await asyncio.sleep(retry_delay) await asyncio.sleep(retry_delay)
retry_delay *= retry_delay_multiplier # Exponential backoff retry_delay *= retry_delay_multiplier
# If all retries fail, raise an exception total_time = time.time() - start_time
raise Exception(f"Request failed after {max_retries} attempts") raise Exception(f"Request failed after {max_retries} attempts and {total_time:.2f} seconds")
from logging import basicConfig, getLogger from logging import basicConfig, getLogger
@@ -339,7 +366,18 @@ def send_prompt(sid: str, inputs: StreamingPrompt):
@server.PromptServer.instance.routes.post("/comfyui-deploy/run") @server.PromptServer.instance.routes.post("/comfyui-deploy/run")
async def comfy_deploy_run(request): async def comfy_deploy_run(request):
# Extract the bearer token from the Authorization header
data = await request.json() data = await request.json()
if "cd_token" in data:
token = data["cd_token"]
else:
auth_header = request.headers.get('Authorization')
token = None
if auth_header:
parts = auth_header.split()
if len(parts) == 2 and parts[0].lower() == 'bearer':
token = parts[1]
# In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky # In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
workflow_api = data.get("workflow_api_raw") workflow_api = data.get("workflow_api_raw")
@@ -354,13 +392,14 @@ async def comfy_deploy_run(request):
prompt = { prompt = {
"prompt": workflow_api, "prompt": workflow_api,
"client_id": "comfy_deploy_instance", #api.client_id "client_id": "comfy_deploy_instance", #api.client_id
"prompt_id": prompt_id "prompt_id": prompt_id,
} }
prompt_metadata[prompt_id] = SimplePrompt( prompt_metadata[prompt_id] = SimplePrompt(
status_endpoint=data.get('status_endpoint'), status_endpoint=data.get('status_endpoint'),
file_upload_endpoint=data.get('file_upload_endpoint'), file_upload_endpoint=data.get('file_upload_endpoint'),
workflow_api=workflow_api workflow_api=workflow_api,
token=token
) )
try: try:
@@ -398,7 +437,7 @@ async def comfy_deploy_run(request):
return web.json_response(res, status=status) return web.json_response(res, status=status)
async def stream_prompt(data): async def stream_prompt(data, token):
# In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky # In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
workflow_api = data.get("workflow_api_raw") workflow_api = data.get("workflow_api_raw")
# The prompt id generated from comfy deploy, can be None # The prompt id generated from comfy deploy, can be None
@@ -418,7 +457,8 @@ async def stream_prompt(data):
prompt_metadata[prompt_id] = SimplePrompt( prompt_metadata[prompt_id] = SimplePrompt(
status_endpoint=data.get('status_endpoint'), status_endpoint=data.get('status_endpoint'),
file_upload_endpoint=data.get('file_upload_endpoint'), file_upload_endpoint=data.get('file_upload_endpoint'),
workflow_api=workflow_api workflow_api=workflow_api,
token=token
) )
# log('info', "Begin prompt", prompt=prompt) # log('info', "Begin prompt", prompt=prompt)
@@ -467,6 +507,14 @@ comfy_message_queues: Dict[str, asyncio.Queue] = {}
async def stream_response(request): async def stream_response(request):
response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'text/event-stream'}) response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'text/event-stream'})
await response.prepare(request) await response.prepare(request)
# Extract the bearer token from the Authorization header
auth_header = request.headers.get('Authorization')
token = None
if auth_header:
parts = auth_header.split()
if len(parts) == 2 and parts[0].lower() == 'bearer':
token = parts[1]
pending = True pending = True
data = await request.json() data = await request.json()
@@ -478,7 +526,7 @@ async def stream_response(request):
log('info', 'Streaming prompt') log('info', 'Streaming prompt')
try: try:
result = await stream_prompt(data=data) result = await stream_prompt(data=data, token=token)
await response.write(f"event: event_update\ndata: {json.dumps(result)}\n\n".encode('utf-8')) await response.write(f"event: event_update\ndata: {json.dumps(result)}\n\n".encode('utf-8'))
# await response.write(.encode('utf-8')) # await response.write(.encode('utf-8'))
await response.drain() # Ensure the buffer is flushed await response.drain() # Ensure the buffer is flushed
@@ -607,9 +655,10 @@ async def upload_file_endpoint(request):
with open(file_path, 'rb') as f: with open(file_path, 'rb') as f:
headers = { headers = {
"Content-Type": file_type, "Content-Type": file_type,
# "x-amz-acl": "public-read", # "Content-Length": str(file_size)
"Content-Length": str(file_size)
} }
if content.get('include_acl') is True:
headers["x-amz-acl"] = "public-read"
upload_response = await async_request_with_retry('PUT', upload_url, data=f, headers=headers) upload_response = await async_request_with_retry('PUT', upload_url, data=f, headers=headers)
if upload_response.status == 200: if upload_response.status == 200:
return web.json_response({ return web.json_response({
@@ -925,6 +974,7 @@ async def send_json_override(self, event, data, sid=None):
"data": data "data": data
}) })
asyncio.create_task(update_run_ws_event(prompt_id, event, data))
# event_emitter.emit("send_json", { # event_emitter.emit("send_json", {
# "event": event, # "event": event,
# "data": data # "data": data
@@ -1022,6 +1072,7 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
return return
status_endpoint = prompt_metadata[prompt_id].status_endpoint status_endpoint = prompt_metadata[prompt_id].status_endpoint
token = prompt_metadata[prompt_id].token
if (status_endpoint is None): if (status_endpoint is None):
return return
@@ -1045,7 +1096,27 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
}) })
# requests.post(status_endpoint, json=body) # requests.post(status_endpoint, json=body)
await async_request_with_retry('POST', status_endpoint, json=body) await async_request_with_retry('POST', status_endpoint, token=token, json=body)
async def update_run_ws_event(prompt_id: str, event: str, data: dict):
if prompt_id not in prompt_metadata:
return
# print("update_run_ws_event", prompt_id, event, data)
status_endpoint = prompt_metadata[prompt_id].status_endpoint
if status_endpoint is None:
return
token = prompt_metadata[prompt_id].token
body = {
"run_id": prompt_id,
"ws_event": {
"event": event,
"data": data,
},
}
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
async def update_run(prompt_id: str, status: Status): async def update_run(prompt_id: str, status: Status):
@@ -1076,7 +1147,8 @@ async def update_run(prompt_id: str, status: Status):
try: try:
# requests.post(status_endpoint, json=body) # requests.post(status_endpoint, json=body)
if (status_endpoint is not None): if (status_endpoint is not None):
await async_request_with_retry('POST', status_endpoint, json=body) token = prompt_metadata[prompt_id].token
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
if (status_endpoint is not None) and cd_enable_run_log and (status == Status.SUCCESS or status == Status.FAILED): if (status_endpoint is not None) and cd_enable_run_log and (status == Status.SUCCESS or status == Status.FAILED):
try: try:
@@ -1106,7 +1178,7 @@ async def update_run(prompt_id: str, status: Status):
] ]
} }
await async_request_with_retry('POST', status_endpoint, json=body) await async_request_with_retry('POST', status_endpoint, token=token, json=body)
# requests.post(status_endpoint, json=body) # requests.post(status_endpoint, json=body)
except Exception as log_error: except Exception as log_error:
logger.info(f"Error reading log file: {log_error}") logger.info(f"Error reading log file: {log_error}")
@@ -1127,7 +1199,48 @@ async def update_run(prompt_id: str, status: Status):
}) })
async def upload_file(prompt_id, filename, subfolder=None, content_type="image/png", type="output"): async def file_sender(file_object, chunk_size):
while True:
chunk = await file_object.read(chunk_size)
if not chunk:
break
yield chunk
chunk_size = 1024 * 1024 # 1MB chunks, adjust as needed
async def upload_with_retry(session, url, headers, data, max_retries=3, initial_delay=1):
start_time = time.time() # Start timing here
for attempt in range(max_retries):
try:
async with session.put(url, headers=headers, data=data) as response:
upload_duration = time.time() - start_time
logger.info(f"Upload attempt {attempt + 1} completed in {upload_duration:.2f} seconds")
logger.info(f"Upload response status: {response.status}")
response.raise_for_status() # This will raise an exception for 4xx and 5xx status codes
response_text = await response.text()
logger.info(f"Response body: {response_text[:1000]}...")
logger.info("Upload successful")
return response # Successful upload, exit the retry loop
except (ClientError, ClientResponseError) as e:
logger.error(f"Upload attempt {attempt + 1} failed: {str(e)}")
if attempt < max_retries - 1: # If it's not the last attempt
delay = initial_delay * (2 ** attempt) # Exponential backoff
logger.info(f"Retrying in {delay} seconds...")
await asyncio.sleep(delay)
else:
logger.error("Max retries reached. Upload failed.")
raise # Re-raise the last exception if all retries are exhausted
except Exception as e:
logger.error(f"Unexpected error during upload: {str(e)}")
logger.error(traceback.format_exc())
raise # Re-raise unexpected exceptions immediately
async def upload_file(prompt_id, filename, subfolder=None, content_type="image/png", type="output", item=None):
""" """
Uploads file to S3 bucket using S3 client object Uploads file to S3 bucket using S3 client object
:return: None :return: None
@@ -1157,7 +1270,7 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
logger.info(f"Uploading file {file}") logger.info(f"Uploading file {file}")
file_upload_endpoint = prompt_metadata[prompt_id].file_upload_endpoint file_upload_endpoint = prompt_metadata[prompt_id].file_upload_endpoint
token = prompt_metadata[prompt_id].token
filename = quote(filename) filename = quote(filename)
prompt_id = quote(prompt_id) prompt_id = quote(prompt_id)
content_type = quote(content_type) content_type = quote(content_type)
@@ -1165,25 +1278,52 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
target_url = f"{file_upload_endpoint}?file_name={filename}&run_id={prompt_id}&type={content_type}&version=v2" target_url = f"{file_upload_endpoint}?file_name={filename}&run_id={prompt_id}&type={content_type}&version=v2"
start_time = time.time() # Start timing here start_time = time.time() # Start timing here
result = await async_request_with_retry("GET", target_url, disable_timeout=True) logger.info(f"Target URL: {target_url}")
result = await async_request_with_retry("GET", target_url, disable_timeout=True, token=token)
end_time = time.time() # End timing after the request is complete end_time = time.time() # End timing after the request is complete
logger.info("Time taken for getting file upload endpoint: {:.2f} seconds".format(end_time - start_time)) logger.info("Time taken for getting file upload endpoint: {:.2f} seconds".format(end_time - start_time))
ok = await result.json() ok = await result.json()
start_time = time.time() # Start timing here logger.info(f"Result: {ok}")
async with aiofiles.open(file, 'rb') as f: async with aiofiles.open(file, 'rb') as f:
data = await f.read() data = await f.read()
size = str(len(data))
# logger.info(f"Image size: {size}")
start_time = time.time() # Start timing here
headers = { headers = {
# "x-amz-acl": "public-read",
"Content-Type": content_type, "Content-Type": content_type,
"Content-Length": str(len(data)), "Content-Length": size,
} }
logger.info(headers)
if ok.get('include_acl') is True:
headers["x-amz-acl"] = "public-read"
# response = requests.put(ok.get("url"), headers=headers, data=data) # response = requests.put(ok.get("url"), headers=headers, data=data)
response = await async_request_with_retry('PUT', ok.get("url"), headers=headers, data=data) # response = await async_request_with_retry('PUT', ok.get("url"), headers=headers, data=data)
logger.info(f"Upload file response status: {response.status}, status text: {response.reason}") # logger.info(f"Upload file response status: {response.status}, status text: {response.reason}")
async with aiohttp.ClientSession() as session:
try:
response = await upload_with_retry(session, ok.get("url"), headers, data)
# Process successful response...
except Exception as e:
# Handle final failure...
logger.error(f"Upload ultimately failed: {str(e)}")
end_time = time.time() # End timing after the request is complete end_time = time.time() # End timing after the request is complete
logger.info("Upload time: {:.2f} seconds".format(end_time - start_time)) logger.info("Upload time: {:.2f} seconds".format(end_time - start_time))
if item is not None:
file_download_url = ok.get("download_url")
if file_download_url is not None:
item["url"] = file_download_url
item["upload_duration"] = end_time - start_time
if ok.get("is_public") is not None:
item["is_public"] = ok.get("is_public")
def have_pending_upload(prompt_id): def have_pending_upload(prompt_id):
if prompt_id in prompt_metadata and len(prompt_metadata[prompt_id].uploading_nodes) > 0: if prompt_id in prompt_metadata and len(prompt_metadata[prompt_id].uploading_nodes) > 0:
@@ -1297,25 +1437,50 @@ async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, d
item.get("filename"), item.get("filename"),
subfolder=item.get("subfolder"), subfolder=item.get("subfolder"),
type=item.get("type"), type=item.get("type"),
content_type=file_type content_type=file_type,
item=item
)) ))
# await upload_file(
# prompt_id,
# item.get("filename"),
# subfolder=item.get("subfolder"),
# type=item.get("type"),
# content_type=file_type,
# item=item
# )
# Execute all upload tasks concurrently # Execute all upload tasks concurrently
await asyncio.gather(*upload_tasks) await asyncio.gather(*upload_tasks)
# Upload files in the background # Upload files in the background
async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=True): async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=True, node_meta=None):
try: try:
# await handle_upload(prompt_id, data, 'images', "content_type", "image/png")
# await handle_upload(prompt_id, data, 'files', "content_type", "image/png")
# await handle_upload(prompt_id, data, 'gifs', "format", "image/gif")
# await handle_upload(prompt_id, data, 'mesh', "format", "application/octet-stream")
upload_tasks = [ upload_tasks = [
handle_upload(prompt_id, data, 'images', "content_type", "image/png"), handle_upload(prompt_id, data, "images", "content_type", "image/png"),
handle_upload(prompt_id, data, 'files', "content_type", "image/png"), handle_upload(prompt_id, data, "files", "content_type", "image/png"),
handle_upload(prompt_id, data, 'gifs', "format", "image/gif"), handle_upload(prompt_id, data, "gifs", "format", "image/gif"),
handle_upload(prompt_id, data, 'mesh', "format", "application/octet-stream") handle_upload(
prompt_id, data, "mesh", "format", "application/octet-stream"
),
] ]
await asyncio.gather(*upload_tasks) await asyncio.gather(*upload_tasks)
status_endpoint = prompt_metadata[prompt_id].status_endpoint
token = prompt_metadata[prompt_id].token
if have_upload: if have_upload:
if status_endpoint is not None:
body = {
"run_id": prompt_id,
"output_data": data,
"node_meta": node_meta,
}
# pprint(body)
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
await update_file_status(prompt_id, data, False, node_id=node_id) await update_file_status(prompt_id, data, False, node_id=node_id)
except Exception as e: except Exception as e:
await handle_error(prompt_id, data, e) await handle_error(prompt_id, data, e)
@@ -1334,7 +1499,10 @@ async def update_run_with_output(prompt_id, data, node_id=None, node_meta=None):
"output_data": data, "output_data": data,
"node_meta": node_meta, "node_meta": node_meta,
} }
have_upload_media = 'images' in data or 'files' in data or 'gifs' in data or 'mesh' in data pprint(body)
have_upload_media = False
if data is not None:
have_upload_media = 'images' in data or 'files' in data or 'gifs' in data or 'mesh' in data
if bypass_upload and have_upload_media: if bypass_upload and have_upload_media:
print("CD_BYPASS_UPLOAD is enabled, skipping the upload of the output:", node_id) print("CD_BYPASS_UPLOAD is enabled, skipping the upload of the output:", node_id)
return return
@@ -1346,15 +1514,16 @@ async def update_run_with_output(prompt_id, data, node_id=None, node_meta=None):
if have_upload_media: if have_upload_media:
await update_file_status(prompt_id, data, True, node_id=node_id) await update_file_status(prompt_id, data, True, node_id=node_id)
asyncio.create_task(upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload_media)) # asyncio.create_task(upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload_media, node_meta=node_meta))
await upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload_media, node_meta=node_meta)
# await upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload) # await upload_in_background(prompt_id, data, node_id=node_id, have_upload=have_upload)
except Exception as e: except Exception as e:
await handle_error(prompt_id, data, e) await handle_error(prompt_id, data, e)
# requests.post(status_endpoint, json=body) # requests.post(status_endpoint, json=body)
if status_endpoint is not None: elif status_endpoint is not None:
await async_request_with_retry('POST', status_endpoint, json=body) token = prompt_metadata[prompt_id].token
await async_request_with_retry('POST', status_endpoint, token=token, json=body)
await send('outputs_uploaded', { await send('outputs_uploaded', {
"prompt_id": prompt_id "prompt_id": prompt_id
+2
View File
@@ -29,6 +29,8 @@ class SimplePrompt(BaseModel):
status_endpoint: Optional[str] status_endpoint: Optional[str]
file_upload_endpoint: Optional[str] file_upload_endpoint: Optional[str]
token: Optional[str]
workflow_api: dict workflow_api: dict
status: Status = Status.NOT_STARTED status: Status = Status.NOT_STARTED
progress: set = set() progress: set = set()
+89 -88
View File
@@ -192,11 +192,13 @@ const ext = {
registerCustomNodes() { registerCustomNodes() {
/** @type {LGraphNode}*/ /** @type {LGraphNode}*/
class ComfyDeploy { class ComfyDeploy extends LGraphNode {
color = LGraphCanvas.node_colors.yellow.color;
bgcolor = LGraphCanvas.node_colors.yellow.bgcolor;
groupcolor = LGraphCanvas.node_colors.yellow.groupcolor;
constructor() { constructor() {
super();
this.color = LGraphCanvas.node_colors.yellow.color;
this.bgcolor = LGraphCanvas.node_colors.yellow.bgcolor;
this.groupcolor = LGraphCanvas.node_colors.yellow.groupcolor;
if (!this.properties) { if (!this.properties) {
this.properties = {}; this.properties = {};
this.properties.workflow_name = ""; this.properties.workflow_name = "";
@@ -204,65 +206,75 @@ const ext = {
this.properties.version = ""; this.properties.version = "";
} }
ComfyWidgets.STRING( this.addWidget(
this, "text",
"workflow_name", "workflow_name",
[ this.properties.workflow_name,
"", (v) => {
{ this.properties.workflow_name = v;
default: this.properties.workflow_name, },
multiline: false, { multiline: false }
},
],
app,
); );
ComfyWidgets.STRING( this.addWidget(
this, "text",
"workflow_id", "workflow_id",
[ this.properties.workflow_id,
"", (v) => {
{ this.properties.workflow_id = v;
default: this.properties.workflow_id, },
multiline: false, { multiline: false }
},
],
app,
); );
ComfyWidgets.STRING( this.addWidget(
this, "text",
"version", "version",
["", { default: this.properties.version, multiline: false }], this.properties.version,
app, (v) => {
this.properties.version = v;
},
{ multiline: false }
); );
// this.widgets.forEach((w) => {
// // w.computeSize = () => [200,10]
// w.computedHeight = 2;
// })
this.widgets_start_y = 10; this.widgets_start_y = 10;
this.setSize(this.computeSize());
// const config = { };
// console.log(this);
this.serialize_widgets = true; this.serialize_widgets = true;
this.isVirtualNode = true; this.isVirtualNode = true;
} }
onExecute() {
// This method is called when the node is executed
// You can add any necessary logic here
}
onSerialize(o) {
// This method is called when the node is being serialized
// Ensure all necessary data is saved
if (!o.properties) {
o.properties = {};
}
o.properties.workflow_name = this.properties.workflow_name;
o.properties.workflow_id = this.properties.workflow_id;
o.properties.version = this.properties.version;
}
onConfigure(o) {
// This method is called when the node is being configured (e.g., when loading a saved graph)
// Ensure all necessary data is restored
if (o.properties) {
this.properties = { ...this.properties, ...o.properties };
this.widgets[0].value = this.properties.workflow_name || "";
this.widgets[1].value = this.properties.workflow_id || "";
this.widgets[2].value = this.properties.version || "1";
}
}
} }
// Load default visibility // Register the node type
LiteGraph.registerNodeType("ComfyDeploy", Object.assign(ComfyDeploy, {
LiteGraph.registerNodeType( title: "Comfy Deploy",
"ComfyDeploy", title_mode: LiteGraph.NORMAL_TITLE,
Object.assign(ComfyDeploy, { collapsable: true,
title_mode: LiteGraph.NORMAL_TITLE, }));
title: "Comfy Deploy",
collapsable: true,
}),
);
ComfyDeploy.category = "deploy"; ComfyDeploy.category = "deploy";
}, },
@@ -431,10 +443,10 @@ function createDynamicUIHtml(data) {
<h3 style="font-size: 14px; font-weight: semibold; margin-bottom: 8px;">Missing Nodes</h3> <h3 style="font-size: 14px; font-weight: semibold; margin-bottom: 8px;">Missing Nodes</h3>
<p style="font-size: 12px;">These nodes are not found with any matching custom_nodes in the ComfyUI Manager Database</p> <p style="font-size: 12px;">These nodes are not found with any matching custom_nodes in the ComfyUI Manager Database</p>
${data.missing_nodes ${data.missing_nodes
.map((node) => { .map((node) => {
return `<p style="font-size: 14px; color: #d69e2e;">${node}</p>`; return `<p style="font-size: 14px; color: #d69e2e;">${node}</p>`;
}) })
.join("")} .join("")}
</div> </div>
`; `;
} }
@@ -442,17 +454,14 @@ function createDynamicUIHtml(data) {
Object.values(data.custom_nodes).forEach((node) => { Object.values(data.custom_nodes).forEach((node) => {
html += ` html += `
<div style="border-bottom: 1px solid #e2e8f0; padding-top: 16px;"> <div style="border-bottom: 1px solid #e2e8f0; padding-top: 16px;">
<a href="${ <a href="${node.url
node.url }" target="_blank" style="font-size: 18px; font-weight: semibold; color: white; text-decoration: none;">${node.name
}" target="_blank" style="font-size: 18px; font-weight: semibold; color: white; text-decoration: none;">${ }</a>
node.name
}</a>
<p style="font-size: 14px; color: #4b5563;">${node.hash}</p> <p style="font-size: 14px; color: #4b5563;">${node.hash}</p>
${ ${node.warning
node.warning ? `<p style="font-size: 14px; color: #d69e2e;">${node.warning}</p>`
? `<p style="font-size: 14px; color: #d69e2e;">${node.warning}</p>` : ""
: "" }
}
</div> </div>
`; `;
}); });
@@ -466,9 +475,8 @@ function createDynamicUIHtml(data) {
Object.entries(data.models).forEach(([section, items]) => { Object.entries(data.models).forEach(([section, items]) => {
html += ` html += `
<div style="border-bottom: 1px solid #e2e8f0; padding-top: 8px; padding-bottom: 8px;"> <div style="border-bottom: 1px solid #e2e8f0; padding-top: 8px; padding-bottom: 8px;">
<h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${ <h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${section.charAt(0).toUpperCase() + section.slice(1)
section.charAt(0).toUpperCase() + section.slice(1) }</h3>`;
}</h3>`;
items.forEach((item) => { items.forEach((item) => {
html += `<p style="font-size: 14px; color: ${textColor};">${item.name}</p>`; html += `<p style="font-size: 14px; color: ${textColor};">${item.name}</p>`;
}); });
@@ -484,9 +492,8 @@ function createDynamicUIHtml(data) {
Object.entries(data.files).forEach(([section, items]) => { Object.entries(data.files).forEach(([section, items]) => {
html += ` html += `
<div style="border-bottom: 1px solid #e2e8f0; padding-top: 8px; padding-bottom: 8px;"> <div style="border-bottom: 1px solid #e2e8f0; padding-top: 8px; padding-bottom: 8px;">
<h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${ <h3 style="font-size: 18px; font-weight: semibold; margin-bottom: 8px;">${section.charAt(0).toUpperCase() + section.slice(1)
section.charAt(0).toUpperCase() + section.slice(1) }</h3>`;
}</h3>`;
items.forEach((item) => { items.forEach((item) => {
html += `<p style="font-size: 14px; color: ${textColor};">${item.name}</p>`; html += `<p style="font-size: 14px; color: ${textColor};">${item.name}</p>`;
}); });
@@ -1006,14 +1013,12 @@ export class LoadingDialog extends ComfyDialog {
showLoading(title, message) { showLoading(title, message) {
this.show(` this.show(`
<div style="width: 400px; display: flex; gap: 18px; flex-direction: column; overflow: unset"> <div style="width: 400px; display: flex; gap: 18px; flex-direction: column; overflow: unset">
<h3 style="margin: 0px; display: flex; align-items: center; justify-content: center; gap: 12px;">${title} ${ <h3 style="margin: 0px; display: flex; align-items: center; justify-content: center; gap: 12px;">${title} ${this.loadingIcon
this.loadingIcon }</h3>
}</h3> ${message
${ ? `<label style="max-width: 100%; white-space: pre-wrap; word-wrap: break-word;">${message}</label>`
message : ""
? `<label style="max-width: 100%; white-space: pre-wrap; word-wrap: break-word;">${message}</label>` }
: ""
}
</div> </div>
`); `);
} }
@@ -1279,21 +1284,17 @@ export class ConfigDialog extends ComfyDialog {
</label> </label>
<label style="color: white; width: 100%;"> <label style="color: white; width: 100%;">
Endpoint: Endpoint:
<input id="endpoint" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="text" value="${ <input id="endpoint" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="text" value="${data.endpoint
data.endpoint }">
}">
</label> </label>
<div style="color: white;"> <div style="color: white;">
API Key: User / Org <button style="font-size: 18px;">${ API Key: User / Org <button style="font-size: 18px;">${data.displayName ?? ""
data.displayName ?? "" }</button>
}</button> <input id="apiKey" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="password" value="${data.apiKey
<input id="apiKey" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;" type="password" value="${ }">
data.apiKey
}">
<button id="loginButton" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;"> <button id="loginButton" style="margin-top: 8px; width: 100%; height:40px; box-sizing: border-box; padding: 0px 6px;">
${ ${data.apiKey ? "Re-login with ComfyDeploy" : "Login with ComfyDeploy"
data.apiKey ? "Re-login with ComfyDeploy" : "Login with ComfyDeploy" }
}
</button> </button>
</div> </div>
</div> </div>