Revert "feat: experiment with await + asyncio.gather for multi file in same node"

This reverts commit c08b68c41f.

comfy-nodes/external_image_batch.py

@@ -39,14 +39,34 @@ class ComfyUIDeployExternalImageBatch:
 
     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):
+        import requests
+        import zipfile
+        import io
+        
         processed_images = []
         try:
             images_list = json.loads(images)  # Assuming images is a JSON array string
             print(images_list)
             for img_input in images_list:
-                if img_input.startswith('http'):
-                    import requests
+                if img_input.startswith('http') and img_input.endswith('.zip'):
+                    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
                     print("Fetching image from url: ", img_input)
                     response = requests.get(img_input)

custom_routes.py

@@ -59,7 +59,7 @@ print(f"max_retries: {max_retries}, retry_delay_multiplier: {retry_delay_multipl
 
 import time
 
-async def async_request_with_retry(method, url, disable_timeout=False, **kwargs):
+async def async_request_with_retry(method, url, disable_timeout=False, token=None, **kwargs):
     global client_session
     await ensure_client_session()
     retry_delay = 1  # Start with 1 second delay
@@ -72,6 +72,11 @@ async def async_request_with_retry(method, url, disable_timeout=False, **kwargs)
                 timeout = ClientTimeout(total=None, connect=initial_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:
                 request_end = time.time()
@@ -361,6 +366,14 @@ def send_prompt(sid: str, inputs: StreamingPrompt):
 
 @server.PromptServer.instance.routes.post("/comfyui-deploy/run")
 async def comfy_deploy_run(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]
+
     data = await request.json()
 
     # In older version, we use workflow_api, but this has inputs already swapped in nextjs frontend, which is tricky
@@ -376,13 +389,14 @@ async def comfy_deploy_run(request):
     prompt = {
         "prompt": workflow_api,
         "client_id": "comfy_deploy_instance", #api.client_id
-        "prompt_id": prompt_id
+        "prompt_id": prompt_id,
     }
 
     prompt_metadata[prompt_id] = SimplePrompt(
         status_endpoint=data.get('status_endpoint'),
         file_upload_endpoint=data.get('file_upload_endpoint'),
-        workflow_api=workflow_api
+        workflow_api=workflow_api,
+        token=token
     )
 
     try:
@@ -420,7 +434,7 @@ async def comfy_deploy_run(request):
 
     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
     workflow_api = data.get("workflow_api_raw")
     # The prompt id generated from comfy deploy, can be None
@@ -440,7 +454,8 @@ async def stream_prompt(data):
     prompt_metadata[prompt_id] = SimplePrompt(
         status_endpoint=data.get('status_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)
@@ -490,6 +505,14 @@ async def stream_response(request):
     response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'text/event-stream'})
     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
     data = await request.json()
 
@@ -500,7 +523,7 @@ async def stream_response(request):
         log('info', 'Streaming prompt')
 
         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(.encode('utf-8'))
             await response.drain()  # Ensure the buffer is flushed
@@ -948,6 +971,7 @@ async def send_json_override(self, event, data, sid=None):
             "data": data
         })
 
+    asyncio.create_task(update_run_ws_event(prompt_id, event, data))
     # event_emitter.emit("send_json", {
     #     "event": event,
     #     "data": data
@@ -1045,6 +1069,7 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
         return
 
     status_endpoint = prompt_metadata[prompt_id].status_endpoint
+    token = prompt_metadata[prompt_id].token
 
     if (status_endpoint is None):
         return
@@ -1068,7 +1093,27 @@ async def update_run_live_status(prompt_id, live_status, calculated_progress: fl
         })
 
     # 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):
@@ -1099,7 +1144,8 @@ async def update_run(prompt_id: str, status: Status):
         try:
             # requests.post(status_endpoint, json=body)
             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):
                 try:
@@ -1129,7 +1175,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)
                 except Exception as log_error:
                     logger.info(f"Error reading log file: {log_error}")
@@ -1180,7 +1226,7 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
     logger.info(f"Uploading file {file}")
 
     file_upload_endpoint = prompt_metadata[prompt_id].file_upload_endpoint
-
+    token = prompt_metadata[prompt_id].token
     filename = quote(filename)
     prompt_id = quote(prompt_id)
     content_type = quote(content_type)
@@ -1192,7 +1238,7 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
 
         start_time = time.time()  # Start timing here
         logger.info(f"Target URL: {target_url}")
-        result = await async_request_with_retry("GET", target_url, disable_timeout=True)
+        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
         logger.info("Time taken for getting file upload endpoint: {:.2f} seconds".format(end_time - start_time))
         ok = await result.json()
@@ -1219,6 +1265,8 @@ async def upload_file(prompt_id, filename, subfolder=None, content_type="image/p
             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):
     if prompt_id in prompt_metadata and len(prompt_metadata[prompt_id].uploading_nodes) > 0:
@@ -1311,7 +1359,7 @@ async def update_file_status(prompt_id: str, data, uploading, have_error=False,
 
 async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, default_content_type: str):
     items = data.get(key, [])
-    upload_tasks = []
+    # upload_tasks = []
 
     for item in items:
         # Skipping temp files
@@ -1327,31 +1375,36 @@ async def handle_upload(prompt_id: str, data, key: str, content_type_key: str, d
         elif file_extension == '.webp':
             file_type = 'image/webp'
 
-        upload_tasks.append(upload_file(
+        # upload_tasks.append(upload_file(
+        #     prompt_id,
+        #     item.get("filename"),
+        #     subfolder=item.get("subfolder"),
+        #     type=item.get("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
-    await asyncio.gather(*upload_tasks)
+    # await asyncio.gather(*upload_tasks)
 
 # Upload files in the background
 async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=True, node_meta=None):
     try:
-        upload_tasks = [
-            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, 'gifs', "format", "image/gif"),
-            handle_upload(prompt_id, data, 'mesh', "format", "application/octet-stream")
-        ]
-
-        await asyncio.gather(*upload_tasks)
+        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")
 
         status_endpoint = prompt_metadata[prompt_id].status_endpoint
+        token = prompt_metadata[prompt_id].token
         if have_upload:
             if status_endpoint is not None:
                 body = {
@@ -1360,7 +1413,7 @@ async def upload_in_background(prompt_id: str, data, node_id=None, have_upload=T
                     "node_meta": node_meta,
                 }
                 # pprint(body)
-                await async_request_with_retry('POST', status_endpoint, json=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)
     except Exception as e:
         await handle_error(prompt_id, data, e)
@@ -1379,7 +1432,10 @@ async def update_run_with_output(prompt_id, data, node_id=None, node_meta=None):
         "output_data": data,
         "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:
         print("CD_BYPASS_UPLOAD is enabled, skipping the upload of the output:", node_id)
         return
@@ -1391,14 +1447,16 @@ async def update_run_with_output(prompt_id, data, node_id=None, node_meta=None):
             if have_upload_media:
                 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, node_meta=node_meta))
+            # 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)
 
         except Exception as e:
             await handle_error(prompt_id, data, e)
     # requests.post(status_endpoint, json=body)
     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', {
         "prompt_id": prompt_id

globals.py

@@ -29,6 +29,8 @@ class SimplePrompt(BaseModel):
     status_endpoint: Optional[str]
     file_upload_endpoint: Optional[str]
     
+    token: Optional[str]
+    
     workflow_api: dict
     status: Status = Status.NOT_STARTED
     progress: set = set()