""" đ SMART ComfyUI Workflow to HTML Generator v2.0 Professional-grade HTML interface generator with modern UI/UX """ import gradio as gr import json import requests import logging from typing import Dict, List, Tuple, Optional from datetime import datetime from pathlib import Path import traceback # Configure logging with UTF-8 encoding for Windows import sys import io # Force UTF-8 output for Windows console if sys.platform == "win32": sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace") logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s", handlers=[ logging.FileHandler("generator.log", encoding="utf-8"), logging.StreamHandler(sys.stdout), ], ) logger = logging.getLogger(__name__) class SmartWorkflowGenerator: """Enhanced workflow generator with professional features""" VERSION = "2.0.0" # Predefined dropdown options matching ComfyUI DROPDOWN_OPTIONS = { "megapixel": [ "0.1", "0.15", "0.2", "0.25", "0.3", "0.35", "0.4", "0.45", "0.5", "0.55", "0.6", "0.65", "0.7", "0.75", "0.8", "0.85", "0.9", "0.95", "1.0", "1.1", "1.2", "1.3", "1.4", "1.5", "2.0", "2.5", "3.0", "4.0", ], "divisible_by": ["8", "16", "32", "64"], "aspect_ratio": [ "1:1 (Perfect Square)", "2:3 (Classic Portrait)", "3:4 (Golden Ratio)", "3:5 (Elegant Vertical)", "4:5 (Artistic Frame)", "5:7 (Balanced Portrait)", "5:8 (Tall Portrait)", "7:9 (Modern Portrait)", "9:16 (Slim Vertical)", "9:19 (Tall Slim)", "9:21 (Ultra Tall)", "9:32 (Skyline)", "3:2 (Golden Landscape)", "4:3 (Classic Landscape)", "5:3 (Wide Horizon)", "5:4 (Balanced Frame)", "7:5 (Elegant Landscape)", "8:5 (Cinematic View)", "9:7 (Artful Horizon)", "16:9 (Panorama)", "19:9 (Cinematic Ultrawide)", "21:9 (Epic Ultrawide)", "32:9 (Super Ultrawide)", ], "sampler_name": [ "euler", "euler_cfg_pp", "euler_ancestral", "euler_ancestral_cfg_pp", "heun", "heunpp2", "dpm_2", "dpm_2_ancestral", "lms", "dpm_fast", "dpm_adaptive", "dpmpp_2s_ancestral", "dpmpp_2s_ancestral_cfg_pp", "dpmpp_sde", "dpmpp_sde_gpu", "dpmpp_2m", "dpmpp_2m_cfg_pp", "dpmpp_2m_sde", "dpmpp_2m_sde_gpu", "dpmpp_3m_sde", "dpmpp_3m_sde_gpu", "ddpm", "lcm", "ipndm", "ipndm_v", "deis", "ddim", "uni_pc", "uni_pc_bh2", "res_momentumized", "res_multistep", "restart", ], "scheduler": [ "normal", "karras", "exponential", "sgm_uniform", "simple", "ddim_uniform", "beta", "linear_quadratic", "lcm", "turbo", "align_your_steps", "gits", ], "type": [ "flux", "flux2", "sd3", "sd15", "sd21", "sdxl", "stable_diffusion", "lumina2", "hunyuan_video", "auraflow", "ltxv", ], "device": ["default", "cpu", "cuda", "mps"], "format": [ "video/h264-mp4", "video/h265-mp4", "video/av1-mp4", "video/vp9-webm", "image/gif", "image/webp", ], "upscale_method": ["nearest-exact", "bilinear", "area", "bicubic", "lanczos"], "keep_proportion": ["crop", "pad", "disabled"], "crop_position": ["center", "top", "bottom", "left", "right"], "pix_fmt": ["yuv420p", "yuv422p", "yuv444p"], } # Hardcoded upload fields for nodes that don't have proper API flags UPLOAD_FIELDS = { # Video Helper Suite nodes "VHS_LoadAudioUpload": {"audio": "audio"}, "VHS_LoadVideo": {"video": "video"}, "VHS_LoadVideoPath": {"video": "video"}, "VHS_LoadImagePath": {"image": "image"}, # Add more as needed } def __init__(self): self.workflow = None self.parameters = [] self.nodes_by_category = {} self.node_defs = {} self.server_url = "http://127.0.0.1:8188" logger.info(f"đ Generator v{self.VERSION} initialized") def fetch_node_defs(self, server_url: str): """Fetch ALL node definitions from ComfyUI API""" try: logger.info(f"đĄ Connecting to ComfyUI: {server_url}") response = requests.get(f"{server_url}/object_info", timeout=10) if response.status_code == 200: self.node_defs = response.json() logger.info(f"â Fetched {len(self.node_defs)} node types from API") # Debug: Save API response for inspection try: with open("api_debug.json", "w", encoding="utf-8") as f: json.dump(self.node_defs, f, indent=2, ensure_ascii=False) logger.info("đ Saved API response to api_debug.json") except Exception as e: logger.warning(f"Could not save API debug: {e}") # Debug: Check for specific nodes if "TextEncodeAceStepAudio1.5" in self.node_defs: node_info = self.node_defs["TextEncodeAceStepAudio1.5"] logger.info(f"đ Found TextEncodeAceStepAudio1.5 in API") logger.info(f" Inputs: {list(node_info.get('input', {}).keys())}") else: logger.warning(f"â ī¸ TextEncodeAceStepAudio1.5 NOT found in API") return ( True, f"â API Connected: {len(self.node_defs)} node types fetched", ) logger.warning(f"â ī¸ HTTP {response.status_code}") return False, f"â ī¸ HTTP {response.status_code}" except requests.exceptions.ConnectionError: logger.error(f"â Connection refused: {server_url}") return ( False, f"â Could not connect to ComfyUI at {server_url}. Is ComfyUI running?", ) except requests.exceptions.Timeout: logger.error(f"â Connection timeout: {server_url}") return False, f"â Connection timeout. ComfyUI is not responding." except Exception as e: logger.error(f"â API Error: {str(e)}") return False, f"â Error: {str(e)}" def get_dropdown_options_from_api( self, class_type: str, input_name: str ) -> Tuple[Optional[List], bool, Optional[str]]: """Extract dropdown options from ComfyUI API with enhanced parsing. Returns: (options_list, has_upload, upload_type)""" if class_type not in self.node_defs: logger.debug(f"đ Node {class_type} not found in API definitions") return None, False node_def = self.node_defs[class_type] # Debug: Log the node structure for specific nodes if "TextEncode" in class_type or "Audio" in class_type or "Ace" in class_type: logger.info(f"đ Checking {class_type}.{input_name}") if "input" in node_def: logger.info(f" Input sections: {list(node_def['input'].keys())}") def extract_options(input_def, section_name): """Extract options from input definition with multiple format support. Returns: (options_list, has_upload, upload_type)""" if not isinstance(input_def, list): return None, False, None if len(input_def) == 0: return None, False, None _has_img_upload_tmp = False _has_aud_upload_tmp = False # Check for image_upload flag in metadata (second element) if len(input_def) > 1 and isinstance(input_def[1], dict): _has_img_upload_tmp = input_def[1].get("image_upload", False) _has_aud_upload_tmp = input_def[1].get("audio_upload", False) if _has_img_upload_tmp: logger.info(f"đŧī¸ {class_type}.{input_name} has image_upload support") if _has_aud_upload_tmp: logger.info(f"đ {class_type}.{input_name} has audio_upload support") # Format 1: COMBO widget - ["COMBO", {"options": [...]}] if ( isinstance(input_def[0], str) and len(input_def) > 1 and isinstance(input_def[1], dict) and "options" in input_def[1] ): options = input_def[1]["options"] if isinstance(options, list) and len(options) > 0: logger.info( f"đ {class_type}.{input_name} ({section_name}): {len(options)} options [COMBO]" ) has_upload = _has_img_upload_tmp or _has_aud_upload_tmp upload_type = ( "audio" if _has_aud_upload_tmp else ("image" if _has_img_upload_tmp else None) ) return options, has_upload, upload_type # Format 2: [["option1", "option2", ...], {}] - list of options in first element if isinstance(input_def[0], list) and len(input_def[0]) > 0: options = input_def[0] if all(isinstance(opt, str) for opt in options): logger.info( f"đ {class_type}.{input_name} ({section_name}): {len(options)} options [list]" ) has_upload = _has_img_upload_tmp or _has_aud_upload_tmp upload_type = ( "audio" if _has_aud_upload_tmp else ("image" if _has_img_upload_tmp else None) ) return options, has_upload, upload_type # Format 3: [{"options": [...]}, {}] - nested options dict in first element if isinstance(input_def[0], dict) and "options" in input_def[0]: options = input_def[0]["options"] if isinstance(options, list) and len(options) > 0: logger.info( f"đ {class_type}.{input_name} ({section_name}): {len(options)} options [nested]" ) has_upload = _has_img_upload_tmp or _has_aud_upload_tmp upload_type = ( "audio" if _has_aud_upload_tmp else ("image" if _has_img_upload_tmp else None) ) return options, has_upload, upload_type # Format 4: Direct list of strings ["option1", "option2"] if all(isinstance(opt, str) for opt in input_def): logger.info( f"đ {class_type}.{input_name} ({section_name}): {len(input_def)} options [direct]" ) has_upload = _has_img_upload_tmp or _has_aud_upload_tmp upload_type = ( "audio" if has_audio_upload else ("image" if _has_img_upload_tmp else None) ) return input_def, has_upload, upload_type has_upload = _has_img_upload_tmp or _has_aud_upload_tmp upload_type = ( "audio" if _has_aud_upload_tmp else ("image" if _has_img_upload_tmp else None) ) return None, has_upload, upload_type # Check required inputs if "input" in node_def: inputs = node_def["input"] if "required" in inputs and input_name in inputs["required"]: input_def = inputs["required"][input_name] options, has_upload, upload_type = extract_options( input_def, "required" ) if options or has_upload: return options, has_upload, upload_type # Check optional inputs if "optional" in inputs and input_name in inputs["optional"]: input_def = inputs["optional"][input_name] options, has_upload, upload_type = extract_options( input_def, "optional" ) if options or has_upload: return options, has_upload, upload_type # Check hidden inputs (some nodes put dropdowns here) if "hidden" in inputs and input_name in inputs["hidden"]: input_def = inputs["hidden"][input_name] options, has_upload, upload_type = extract_options(input_def, "hidden") if options or has_upload: return options, has_upload, upload_type return None, False, None def get_input_description(self, class_type: str, input_name: str) -> str: """Get parameter description from API for tooltips""" if class_type not in self.node_defs: return "" node_def = self.node_defs[class_type] # Try to extract description if available if "input" in node_def: for section in ["required", "optional"]: if ( section in node_def["input"] and input_name in node_def["input"][section] ): input_def = node_def["input"][section][input_name] if isinstance(input_def, list) and len(input_def) > 1: if isinstance(input_def[1], dict) and "tooltip" in input_def[1]: return input_def[1]["tooltip"] return "" def load_workflow(self, file, server_url="http://127.0.0.1:8188") -> Tuple: """Load and analyze workflow with full API integration""" try: logger.info("=" * 60) logger.info( f"đ Loading workflow: {file.name if hasattr(file, 'name') else 'uploaded file'}" ) logger.info(f"đ ComfyUI Server: {server_url}") logger.info("=" * 60) self.server_url = server_url # STEP 1: Fetch API definitions logger.info("STEP 1: Fetching ComfyUI API...") api_success, api_msg = self.fetch_node_defs(server_url) # STEP 2: Load workflow logger.info("STEP 2: Loading workflow JSON...") if hasattr(file, "name"): with open(file.name, "r", encoding="utf-8") as f: self.workflow = json.load(f) else: if file is None: return None, "â No file uploaded", [] self.workflow = json.loads( file.decode("utf-8") if isinstance(file, bytes) else file ) logger.info(f"â Loaded workflow with {len(self.workflow)} nodes") # STEP 3: Extract parameters logger.info("STEP 3: Analyzing parameters...") self.parameters = [] self.nodes_by_category = {} for node_id, node_data in self.workflow.items(): class_type = node_data.get("class_type", "") inputs = node_data.get("inputs", {}) title = node_data.get("_meta", {}).get("title", class_type) logger.debug(f"đ Node {node_id}: {class_type} ({title})") for input_name, input_value in inputs.items(): # Skip connections if isinstance(input_value, list): continue # Get API options and description api_options, has_upload, upload_type = ( self.get_dropdown_options_from_api(class_type, input_name) ) description = self.get_input_description(class_type, input_name) # Log for debugging specific nodes if ( "TextEncode" in class_type or "Audio" in class_type or "language" in input_name.lower() or has_upload ): logger.info( f"đ Processing: {class_type}.{input_name} - Options: {api_options is not None} ({len(api_options) if api_options else 0} items), Upload: {has_upload} ({upload_type})" ) # Fallback to hardcoded if not api_options: api_options = self.get_dropdown_options(input_name) # Detect type - pass upload_type instead of has_image_upload param_type, detected_upload_type = self.smart_detect_type( input_name, input_value, class_type, api_options, has_upload, upload_type, ) category = self.categorize(class_type, input_name, title) # Use detected upload type if available, otherwise keep API value final_upload_type = ( detected_upload_type if detected_upload_type else upload_type ) param = { "node_id": node_id, "node_title": title, "node_class": class_type, "input_name": input_name, "value": input_value, "type": param_type, "options": api_options, "category": category, "description": description, "has_upload": has_upload or (detected_upload_type is not None), "upload_type": final_upload_type, } self.parameters.append(param) if category not in self.nodes_by_category: self.nodes_by_category[category] = set() self.nodes_by_category[category].add(node_id) # Build checkbox options options = [] for i, param in enumerate(self.parameters): opts_info = ( f" [{len(param['options'])} opts]" if param["options"] else "" ) label = f"{param['category']} | {param['node_title']} â {param['input_name']} ({param['type']}{opts_info})" options.append((label, i)) logger.info("=" * 60) logger.info(f"â COMPLETE: {len(self.parameters)} parameters extracted") logger.info(f"đ {len(self.nodes_by_category)} categories detected") logger.info("=" * 60) info = api_msg + "\n" info += f"â Loaded {len(self.parameters)} parameters from {len(self.workflow)} nodes\n" info += f"đ {len(self.nodes_by_category)} categories detected" return ( gr.update(choices=options, value=list(range(len(options)))), info, list(range(len(options))), ) except Exception as e: error_msg = f"â Error: {str(e)}\n{traceback.format_exc()}" logger.error(error_msg) return gr.update(choices=[], value=[]), error_msg, [] def smart_detect_type( self, input_name: str, value, class_type: str, api_options: Optional[List], has_upload: bool = False, upload_type: Optional[str] = None, ): """Enhanced type detection with API priority - returns (param_type, upload_type)""" name_lower = input_name.lower() # Priority #0: Upload fields (image or audio) from API flags if has_upload: if upload_type == "audio": return "audio_upload", "audio" else: return "image_upload", "image" # Priority #0.5: Hardcoded upload fields for nodes without proper API flags if class_type in self.UPLOAD_FIELDS: if input_name in self.UPLOAD_FIELDS[class_type]: field_type = self.UPLOAD_FIELDS[class_type][input_name] return f"{field_type}_upload", field_type # Priority #1: API dropdown if api_options and len(api_options) > 0: return "dropdown", None # Check hardcoded dropdown if input_name in self.DROPDOWN_OPTIONS or name_lower in self.DROPDOWN_OPTIONS: return "dropdown", None # Type detection from value if isinstance(value, bool): return "toggle", None elif isinstance(value, int): if "seed" in name_lower and value > 1000: return "seed", None return "number", None elif isinstance(value, float): return ("slider_small" if value <= 2.0 else "slider"), None elif isinstance(value, str): try: float_val = float(value) return ("slider_small" if float_val <= 2.0 else "slider"), None except: pass if len(value) > 100 or "\n" in value: return "textarea", None if any( ext in value for ext in [".safetensors", ".ckpt", ".gguf", ".pth", ".pt"] ): return "file_selector", None if "/" in value or "\\" in value: return "path", None return "text", None def get_dropdown_options(self, input_name: str) -> Optional[List[str]]: """Fallback to hardcoded dropdown options""" if input_name in self.DROPDOWN_OPTIONS: return self.DROPDOWN_OPTIONS[input_name] name_lower = input_name.lower() if name_lower in self.DROPDOWN_OPTIONS: return self.DROPDOWN_OPTIONS[name_lower] for key in self.DROPDOWN_OPTIONS.keys(): if key in name_lower: return self.DROPDOWN_OPTIONS[key] return None def categorize(self, class_type: str, input_name: str, title: str): """Smart categorization with emoji icons""" class_lower = class_type.lower() input_lower = input_name.lower() title_lower = title.lower() if any(w in class_lower for w in ["loader", "load"]): if ( "model" in class_lower or "checkpoint" in class_lower or "gguf" in class_lower ): return "đ¤ Models" if "vae" in class_lower: return "đ¨ VAE" if "clip" in class_lower: return "đ CLIP" if "lora" in class_lower: return "⨠LoRA" return "đ Loaders" if any( w in class_lower for w in ["sampler", "ksampler", "sample", "scheduler"] ): return "âī¸ Sampling" if "resolution" in class_lower or "resolution" in title_lower: return "đ Resolution" if any(w in class_lower for w in ["condition", "encode"]): if "text" in class_lower or "prompt" in input_lower: return "đ Prompts" return "đ¯ Conditioning" if "seed" in input_lower or "noise" in input_lower: return "đ˛ Seeds" if any(w in input_lower for w in ["steps", "cfg", "denoise"]): return "⥠Generation" if any(w in class_lower for w in ["image", "resize", "scale", "upscale"]): return "đŧī¸ Images" if any(w in class_lower for w in ["video", "frame", "fps"]): return "đŦ Video" if "batch" in class_lower or "batch" in title_lower: return f"đĻ Batch" if any(w in class_lower for w in ["audio", "sound", "tts"]): return "đ Audio" if any(w in class_lower for w in ["save", "output", "preview", "combine"]): return "đž Output" return f"đ§ {title}" def generate_html( self, selected_indices, app_name, server_url, output_path, enable_features ): """Generate enhanced HTML with modern features""" if not selected_indices: return "â Please select at least one parameter", None selected = [ self.parameters[i] for i in selected_indices if i < len(self.parameters) ] # Group by category grouped = {} for param in selected: cat = param["category"] if cat not in grouped: grouped[cat] = [] grouped[cat].append(param) # Parse enabled features features = { "dark_mode": "Dark Mode" in enable_features, "presets": "Preset System" in enable_features, "gallery": "Enhanced Gallery" in enable_features, "progress": "Progress Tracking" in enable_features, "tooltips": "Tooltips" in enable_features, "keyboard": "Keyboard Shortcuts" in enable_features, "autosave": "Auto-save Parameters" in enable_features, "batch": "Batch Generation" in enable_features, } html = self.build_enhanced_html( app_name, server_url, output_path, grouped, self.workflow, features ) filename = f"{app_name.replace(' ', '_').replace('/', '_')}.html" filepath = Path(filename) with open(filepath, "w", encoding="utf-8") as f: f.write(html) logger.info(f"â Generated: {filename} ({len(html)} bytes)") result = f"â Generated: {filename}\n" result += f"đ Parameters: {len(selected)}\n" result += f"đ Categories: {len(grouped)}\n" result += f"đ¨ Features: {', '.join([k for k, v in features.items() if v])}\n" result += f"đĻ Size: {len(html) // 1024}KB\n" result += f"đ Location: {str(filepath.absolute())}" return result, str(filepath.absolute()) def build_enhanced_html( self, app_name, server_url, output_path, grouped, workflow, features ): """Build modern, feature-rich HTML""" # Generate parameter inputs HTML inputs_html = "" param_map = [] bypass_toggles = [] for category, params in sorted(grouped.items()): cat_id = ( category.replace(" ", "_") .replace(":", "_") .replace("đ¤", "") .replace("đ", "") .replace("âī¸", "") .replace("đ¨", "") .replace("â¨", "") .replace("đ", "") .replace("đ", "") .replace("đ¯", "") .replace("đ˛", "") .replace("âĄ", "") .replace("đŧī¸", "") .replace("đŦ", "") .replace("đĻ", "") .replace("đ", "") .replace("đž", "") .replace("đ§", "") .strip() ) bypass_id = f"bypass_{cat_id}" cat_nodes = set(p["node_id"] for p in params) bypass_toggles.append({"id": bypass_id, "nodes": list(cat_nodes)}) inputs_html += f'''
Initializing...
Ctrl + Enter - Generate
Ctrl + R - Reset
Ctrl + S - Save Preset
? - Toggle this help