1322 lines
60 KiB
Python
1322 lines
60 KiB
Python
import torch
|
|
import os
|
|
import folder_paths
|
|
from PIL import Image, ImageOps
|
|
import numpy as np
|
|
import hashlib
|
|
import json
|
|
import re
|
|
|
|
class SimpleReadableMetadataSG:
|
|
"""Load image with drag-and-drop, automatically extract properties and metadata"""
|
|
|
|
CATEGORY = "image/analysis"
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
input_dir = folder_paths.get_input_directory()
|
|
files = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))]
|
|
return {
|
|
"required": {
|
|
"image": (sorted(files), {"image_upload": True}),
|
|
"emoji_in_readable_text": ("BOOLEAN", {"default": True}),
|
|
"show_info": (["both", "properties", "metadata", "none"], {"default": "both"}),
|
|
},
|
|
"ui": {
|
|
"image": {"min_width": 450},
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING", "IMAGE", "MASK", "STRING", "STRING", "STRING", "INT", "STRING")
|
|
RETURN_NAMES = ("Simple_Readable_Metadata", "image", "mask", "metadata_raw", "Positive_Prompt", "Negative_Prompt", "seed", "file_name_text")
|
|
FUNCTION = "load_analyze_extract"
|
|
OUTPUT_NODE = True
|
|
|
|
@classmethod
|
|
def IS_CHANGED(cls, image, emoji_in_readable_text, show_info="both"):
|
|
image_path = folder_paths.get_annotated_filepath(image)
|
|
m = hashlib.sha256()
|
|
with open(image_path, 'rb') as f:
|
|
m.update(f.read())
|
|
return m.digest().hex()
|
|
|
|
@classmethod
|
|
def VALIDATE_INPUTS(cls, image, emoji_in_readable_text, show_info="both"):
|
|
|
|
if not folder_paths.exists_annotated_filepath(image):
|
|
return "Invalid image file: {}".format(image)
|
|
return True
|
|
|
|
def is_node_reference(self, value):
|
|
"""Check if a value is a node reference like [node_id, output_index]"""
|
|
try:
|
|
if isinstance(value, list):
|
|
if len(value) == 2:
|
|
if isinstance(value[0], (str, int)) and isinstance(value[1], int):
|
|
return True
|
|
except:
|
|
pass
|
|
return False
|
|
|
|
def safely_convert_to_string(self, value):
|
|
"""Safely convert any value to string - GUARANTEED TO RETURN STRING"""
|
|
try:
|
|
if self.is_node_reference(value):
|
|
return "N/A"
|
|
if isinstance(value, str):
|
|
return value
|
|
elif isinstance(value, list):
|
|
if len(value) > 0:
|
|
return self.safely_convert_to_string(value[0])
|
|
return "N/A"
|
|
elif value is None:
|
|
return "N/A"
|
|
else:
|
|
return str(value)
|
|
except:
|
|
return "N/A"
|
|
|
|
def safely_process_value(self, value):
|
|
"""Master wrapper - ensures value is ALWAYS a string before any operations"""
|
|
return self.safely_convert_to_string(value)
|
|
|
|
def _get_prompt_data_from_image(self, img):
|
|
"""Helper method to extract prompt data from both PNG and WebP formats"""
|
|
try:
|
|
# First check direct img.info (works for PNG)
|
|
if hasattr(img, 'info') and img.info:
|
|
if "prompt" in img.info:
|
|
try:
|
|
prompt_str = img.info["prompt"]
|
|
if isinstance(prompt_str, str):
|
|
return json.loads(prompt_str)
|
|
return prompt_str
|
|
except:
|
|
pass
|
|
|
|
# Check EXIF data for WebP
|
|
if "exif" in img.info:
|
|
try:
|
|
exif_bytes = img.info["exif"]
|
|
if isinstance(exif_bytes, bytes):
|
|
exif_string = exif_bytes.decode('utf-8', errors='ignore')
|
|
# Look for prompt: marker
|
|
if "prompt:" in exif_string:
|
|
prompt_start = exif_string.find("prompt:")
|
|
if prompt_start != -1:
|
|
prompt_data = exif_string[prompt_start + 7:]
|
|
prompt_data = prompt_data.split('\x00')[0]
|
|
try:
|
|
return json.loads(prompt_data)
|
|
except:
|
|
pass
|
|
except Exception as e:
|
|
print(f"Error parsing WebP EXIF for prompt: {e}")
|
|
except Exception as e:
|
|
print(f"Error extracting prompt data: {e}")
|
|
return None
|
|
|
|
def extract_model_name(self, img):
|
|
"""Extract model name from image metadata"""
|
|
model_name = "N/A"
|
|
try:
|
|
if not hasattr(img, 'info') or not img.info:
|
|
return model_name
|
|
|
|
# Try to get prompt data (works for both PNG and WebP)
|
|
prompt_data = self._get_prompt_data_from_image(img)
|
|
if prompt_data:
|
|
try:
|
|
for node_id, node_data in prompt_data.items():
|
|
class_type = node_data.get('class_type', '')
|
|
inputs = node_data.get('inputs', {})
|
|
|
|
if 'CheckpointLoader' in class_type and 'ckpt_name' in inputs:
|
|
model_name = inputs['ckpt_name']
|
|
break
|
|
if 'UNETLoader' in class_type and 'unet_name' in inputs:
|
|
model_name = f"{inputs['unet_name']} (UNET)"
|
|
break
|
|
if 'Loader' in class_type:
|
|
if 'ckpt_name' in inputs:
|
|
model_name = inputs['ckpt_name']
|
|
break
|
|
elif 'unet_name' in inputs:
|
|
model_name = f"{inputs['unet_name']} (UNET)"
|
|
break
|
|
elif 'model_name' in inputs:
|
|
model_name = inputs['model_name']
|
|
break
|
|
except Exception as e:
|
|
print(f"Error parsing prompt metadata: {e}")
|
|
|
|
# Try workflow format (PNG direct access)
|
|
if model_name == "N/A" and 'workflow' in img.info:
|
|
try:
|
|
workflow_data = json.loads(img.info['workflow'])
|
|
for node in workflow_data.get('nodes', []):
|
|
node_type = node.get('type', '')
|
|
if 'Checkpoint' in node_type or 'Loader' in node_type:
|
|
widgets = node.get('widgets_values', [])
|
|
if widgets and len(widgets) > 0:
|
|
model_name = widgets[0]
|
|
break
|
|
except Exception as e:
|
|
print(f"Error parsing workflow metadata: {e}")
|
|
|
|
# Try A1111 format (PNG direct access)
|
|
if model_name == "N/A" and 'parameters' in img.info:
|
|
try:
|
|
params = img.info['parameters']
|
|
model_pattern = r'Model:\s*([^,\n]+)'
|
|
match = re.search(model_pattern, params)
|
|
if match:
|
|
model_name = match.group(1).strip()
|
|
except Exception as e:
|
|
print(f"Error parsing A1111 metadata: {e}")
|
|
|
|
except Exception as e:
|
|
print(f"Error extracting model metadata: {e}")
|
|
|
|
return model_name
|
|
|
|
def extract_generation_params(self, img):
|
|
"""Extract generation parameters (seed, steps, cfg, sampler, scheduler) from image metadata"""
|
|
params = {
|
|
'seed': 'N/A',
|
|
'steps': 'N/A',
|
|
'cfg': 'N/A',
|
|
'sampler': 'N/A',
|
|
'scheduler': 'N/A'
|
|
}
|
|
|
|
try:
|
|
if not hasattr(img, 'info') or not img.info:
|
|
return params
|
|
|
|
# Try to get prompt data (works for both PNG and WebP)
|
|
prompt_data = self._get_prompt_data_from_image(img)
|
|
if prompt_data:
|
|
try:
|
|
for node_id, node_data in prompt_data.items():
|
|
class_type = node_data.get('class_type', '')
|
|
inputs = node_data.get('inputs', {})
|
|
|
|
# KSampler node has all the info we need
|
|
if class_type == "KSampler":
|
|
params['seed'] = inputs.get('seed', 'N/A')
|
|
params['steps'] = inputs.get('steps', 'N/A')
|
|
params['cfg'] = inputs.get('cfg', 'N/A')
|
|
params['sampler'] = inputs.get('sampler_name', 'N/A')
|
|
params['scheduler'] = inputs.get('scheduler', 'N/A')
|
|
return params
|
|
|
|
# Check individual nodes for distributed sampler setup
|
|
if 'seed' in inputs or 'noise_seed' in inputs:
|
|
params['seed'] = inputs.get('seed', inputs.get('noise_seed', params['seed']))
|
|
if 'steps' in inputs:
|
|
params['steps'] = inputs.get('steps', params['steps'])
|
|
if 'cfg' in inputs:
|
|
params['cfg'] = inputs.get('cfg', params['cfg'])
|
|
if 'sampler_name' in inputs:
|
|
params['sampler'] = inputs.get('sampler_name', params['sampler'])
|
|
if 'scheduler' in inputs:
|
|
params['scheduler'] = inputs.get('scheduler', params['scheduler'])
|
|
|
|
except Exception as e:
|
|
print(f"Error parsing ComfyUI generation params: {e}")
|
|
|
|
# Try A1111/Forge format (PNG direct access)
|
|
if 'parameters' in img.info and any(v == 'N/A' for v in params.values()):
|
|
try:
|
|
metadata_text = img.info['parameters']
|
|
|
|
seed_match = re.search(r'Seed:\s*(\d+)', metadata_text)
|
|
if seed_match:
|
|
params['seed'] = int(seed_match.group(1))
|
|
|
|
steps_match = re.search(r'Steps:\s*(\d+)', metadata_text)
|
|
if steps_match:
|
|
params['steps'] = int(steps_match.group(1))
|
|
|
|
cfg_match = re.search(r'CFG scale:\s*([\d.]+)', metadata_text)
|
|
if cfg_match:
|
|
params['cfg'] = float(cfg_match.group(1))
|
|
|
|
sampler_match = re.search(r'Sampler:\s*([^,\n]+)', metadata_text)
|
|
if sampler_match:
|
|
params['sampler'] = sampler_match.group(1).strip()
|
|
|
|
scheduler_match = re.search(r'Schedule type:\s*([^,\n]+)', metadata_text)
|
|
if scheduler_match:
|
|
params['scheduler'] = scheduler_match.group(1).strip()
|
|
|
|
except Exception as e:
|
|
print(f"Error parsing A1111 generation params: {e}")
|
|
|
|
except Exception as e:
|
|
print(f"Error extracting generation parameters: {e}")
|
|
|
|
return params
|
|
|
|
def load_analyze_extract(self, image, emoji_in_readable_text=True, show_info="both"):
|
|
"""Combined function that loads image, analyzes properties, and extracts metadata"""
|
|
try:
|
|
image_path = folder_paths.get_annotated_filepath(image)
|
|
img = Image.open(image_path)
|
|
|
|
model_name = self.extract_model_name(img)
|
|
gen_params = self.extract_generation_params(img)
|
|
|
|
img = ImageOps.exif_transpose(img)
|
|
|
|
metadata_raw = self.extract_raw_metadata(img)
|
|
|
|
if img.mode == 'I':
|
|
img = img.point(lambda i: i * (1 / 255))
|
|
|
|
original_img = img
|
|
if img.mode != 'RGB':
|
|
img = img.convert('RGB')
|
|
|
|
image_tensor = torch.from_numpy(np.array(img).astype(np.float32) / 255.0).unsqueeze(0)
|
|
|
|
if 'A' in original_img.getbands():
|
|
mask = np.array(original_img.getchannel('A')).astype(np.float32) / 255.0
|
|
mask = 1. - torch.from_numpy(mask)
|
|
else:
|
|
mask = torch.zeros((original_img.size[1], original_img.size[0]), dtype=torch.float32, device="cpu")
|
|
|
|
# Build display info for UI
|
|
batch_size, height, width, channels = image_tensor.shape
|
|
total_pixels = width * height
|
|
resolution_mp = float(total_pixels / 1_000_000)
|
|
|
|
# Get actual file size
|
|
try:
|
|
file_size_bytes = os.path.getsize(image_path)
|
|
file_size_mb = float(file_size_bytes) / (1024 * 1024)
|
|
self._current_image_path = os.path.basename(image_path)
|
|
self._current_file_size = file_size_mb
|
|
file_name_text_without_ext = os.path.splitext(os.path.basename(image_path))[0]
|
|
except:
|
|
file_size_mb = 0.0
|
|
file_name_text_without_ext = "unknown"
|
|
|
|
def gcd(a, b):
|
|
while b:
|
|
a, b = b, a % b
|
|
return a
|
|
|
|
def find_closest_standard_ratio(decimal_ratio):
|
|
standard_ratios = [
|
|
(1.0, '1:1'), (1.25, '5:4'), (1.33333, '4:3'),
|
|
(1.5, '3:2'), (1.6, '16:10'), (1.66667, '5:3'),
|
|
(1.77778, '16:9'), (1.88889, '17:9'), (2.0, '2:1'),
|
|
(2.33333, '21:9'), (2.35, '2.35:1'), (2.39, '2.39:1'),
|
|
(2.4, '12:5'),
|
|
]
|
|
|
|
closest_diff = float('inf')
|
|
closest_ratio = None
|
|
error_threshold = 0.05
|
|
|
|
for std_value, std_string in standard_ratios:
|
|
diff = abs(std_value - decimal_ratio)
|
|
if diff < closest_diff:
|
|
closest_diff = diff
|
|
closest_ratio = std_string
|
|
|
|
if closest_diff <= error_threshold:
|
|
return closest_ratio
|
|
return None
|
|
|
|
divisor = gcd(width, height)
|
|
width_ratio = float(width // divisor)
|
|
height_ratio = float(height // divisor)
|
|
aspect_ratio_decimal = width / height
|
|
closest_standard = find_closest_standard_ratio(aspect_ratio_decimal)
|
|
|
|
# Store image/tensor info for readable format reuse
|
|
self._last_width = width
|
|
self._last_height = height
|
|
self._last_resolution_mp = resolution_mp
|
|
self._last_width_ratio = width_ratio
|
|
self._last_height_ratio = height_ratio
|
|
self._last_aspect_ratio_decimal = aspect_ratio_decimal
|
|
self._last_closest_standard = closest_standard
|
|
self._last_file_size_mb = file_size_mb
|
|
|
|
line1 = f"{width}x{height} | {resolution_mp:.2f}MP "
|
|
|
|
if closest_standard and closest_standard != f"{int(width_ratio)}:{int(height_ratio)}":
|
|
line2 = f"Ratio: {int(width_ratio)}:{int(height_ratio)} or {aspect_ratio_decimal:.2f}:1 or ~{closest_standard}"
|
|
else:
|
|
line2 = f"Ratio: {int(width_ratio)}:{int(height_ratio)} or {aspect_ratio_decimal:.2f}:1"
|
|
|
|
line3 = f"File Size: {file_size_mb:.2f}MB"
|
|
|
|
lines = [line1, line2, line3]
|
|
lines.append("")
|
|
|
|
line4 = f"Model: {model_name}"
|
|
lines.append(line4)
|
|
|
|
line5 = f"Seed: {gen_params['seed']} | Steps: {gen_params['steps']} | CFG: {gen_params['cfg']}"
|
|
lines.append(line5)
|
|
|
|
line6 = f"Sampler: {gen_params['sampler']} | Scheduler: {gen_params['scheduler']}"
|
|
lines.append(line6)
|
|
|
|
Simple_Readable_Metadata, positive, negative = self.parse_metadata(metadata_raw, emoji_in_readable_text)
|
|
|
|
# Convert seed to integer, use 0 if N/A
|
|
seed_value = gen_params['seed']
|
|
if seed_value == 'N/A' or seed_value is None:
|
|
seed_int = 0
|
|
else:
|
|
try:
|
|
seed_int = int(seed_value)
|
|
except (ValueError, TypeError):
|
|
seed_int = 0
|
|
|
|
return {
|
|
"ui": {"text": lines},
|
|
"result": (Simple_Readable_Metadata, image_tensor, mask, metadata_raw, positive, negative, seed_int, file_name_text_without_ext)
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error in load_analyze_extract: {e}")
|
|
raise
|
|
|
|
def extract_raw_metadata(self, img):
|
|
"""Extract raw metadata in format compatible with conversion - supports PNG and WebP"""
|
|
png_info = img.info if hasattr(img, 'info') else {}
|
|
|
|
if not png_info:
|
|
return "No metadata found in image"
|
|
|
|
# Check for ComfyUI prompt metadata (works for PNG)
|
|
if "prompt" in png_info:
|
|
try:
|
|
prompt_data = png_info["prompt"]
|
|
if isinstance(prompt_data, str):
|
|
json.loads(prompt_data) # Validate JSON
|
|
return prompt_data
|
|
else:
|
|
return json.dumps(prompt_data)
|
|
except:
|
|
pass
|
|
|
|
# Check for A1111/Forge parameters (works for PNG)
|
|
if "parameters" in png_info:
|
|
return png_info["parameters"]
|
|
|
|
# Check for EXIF data in WebP
|
|
if "exif" in png_info:
|
|
try:
|
|
exif_bytes = png_info["exif"]
|
|
if isinstance(exif_bytes, bytes):
|
|
# Decode the EXIF bytes to string
|
|
exif_string = exif_bytes.decode('utf-8', errors='ignore')
|
|
|
|
# Look for prompt: or workflow: markers in the EXIF data
|
|
if "prompt:" in exif_string:
|
|
# Extract the JSON after "prompt:"
|
|
prompt_start = exif_string.find("prompt:")
|
|
if prompt_start != -1:
|
|
# Extract everything after "prompt:"
|
|
prompt_data = exif_string[prompt_start + 7:] # Skip "prompt:"
|
|
|
|
# Find the end of the JSON (look for the null terminator or end)
|
|
# Try to extract valid JSON
|
|
try:
|
|
# Remove any trailing null bytes or extra data
|
|
prompt_data = prompt_data.split('\x00')[0]
|
|
# Validate it's proper JSON
|
|
json.loads(prompt_data)
|
|
return prompt_data
|
|
except:
|
|
pass
|
|
|
|
# If no valid prompt found, try to extract workflow
|
|
if "workflow:" in exif_string:
|
|
workflow_start = exif_string.find("workflow:")
|
|
if workflow_start != -1:
|
|
workflow_data = exif_string[workflow_start + 9:] # Skip "workflow:"
|
|
try:
|
|
workflow_data = workflow_data.split('\x00')[0]
|
|
workflow_json = json.loads(workflow_data)
|
|
# Return workflow as a fallback
|
|
return json.dumps({"workflow": workflow_json})
|
|
except:
|
|
pass
|
|
|
|
except Exception as e:
|
|
print(f"Error parsing WebP EXIF metadata: {e}")
|
|
|
|
# Fallback: Try using PIL's getexif() method for standard EXIF tags
|
|
if hasattr(img, 'getexif'):
|
|
try:
|
|
exif_data = img.getexif()
|
|
if exif_data:
|
|
# Try to find UserComment tag (0x9286)
|
|
user_comment = exif_data.get(0x9286)
|
|
if user_comment:
|
|
if isinstance(user_comment, bytes):
|
|
user_comment = user_comment.decode('utf-8', errors='ignore')
|
|
try:
|
|
json.loads(user_comment)
|
|
return user_comment
|
|
except:
|
|
return user_comment
|
|
except Exception as e:
|
|
print(f"Error reading EXIF via getexif(): {e}")
|
|
|
|
return "No ComfyUI or WebUI format metadata found. Image may be from a different source."
|
|
|
|
def parse_metadata(self, metadata_raw, include_emojis=True):
|
|
"""Main parsing function that detects format and routes to appropriate converter"""
|
|
try:
|
|
# Handle bytes (WebP EXIF data can be bytes)
|
|
if isinstance(metadata_raw, (bytes, bytearray)):
|
|
try:
|
|
metadata_raw = metadata_raw.decode("utf-8", errors="ignore")
|
|
except Exception:
|
|
metadata_raw = str(metadata_raw)
|
|
|
|
# Convert to string safely (handles dict/list/node references)
|
|
metadata_raw = self.safely_process_value(metadata_raw)
|
|
|
|
# Now detect format and parse
|
|
format_type = self.detect_format(metadata_raw)
|
|
|
|
if format_type == "comfyui":
|
|
Simple_Readable_Metadata = self.parse_comfyui_format(
|
|
metadata_raw,
|
|
include_emojis,
|
|
image_path=getattr(self, '_current_image_path', None),
|
|
file_size_mb=getattr(self, '_current_file_size', None)
|
|
)
|
|
elif format_type == "webui":
|
|
Simple_Readable_Metadata = self.parse_webui_format(metadata_raw, include_emojis)
|
|
else:
|
|
Simple_Readable_Metadata = "Unable to detect metadata format. Please ensure the input is from ComfyUI or WebUI Forge/A1111."
|
|
|
|
positive, negative = self.extract_individual_params(metadata_raw, format_type)
|
|
|
|
positive = self.safely_convert_to_string(positive)
|
|
negative = self.safely_convert_to_string(negative)
|
|
|
|
return (Simple_Readable_Metadata, positive, negative)
|
|
|
|
except Exception as e:
|
|
print(f"Error in parse_metadata: {e}")
|
|
return (f"Error parsing metadata: {str(e)}", "N/A", "N/A")
|
|
|
|
def detect_format(self, text):
|
|
"""Detect whether the input is ComfyUI JSON or WebUI text format"""
|
|
try:
|
|
text = self.safely_process_value(text).strip()
|
|
if text.startswith("Prompt: "):
|
|
text = text[8:]
|
|
|
|
# Try JSON first
|
|
try:
|
|
json.loads(text)
|
|
return "comfyui"
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Then try A1111/WebUI pattern detection
|
|
if (re.search(r'Steps:\s*\d+', text) or
|
|
re.search(r'Sampler:\s*\w+', text) or
|
|
re.search(r'CFG scale:\s*[\d.]+', text)):
|
|
return "webui"
|
|
|
|
return "unknown"
|
|
except Exception as e:
|
|
print(f"Error detecting format: {e}")
|
|
return "unknown"
|
|
|
|
def parse_webui_format(self, text, include_emojis=True):
|
|
"""Parse A1111/WebUI Forge text format metadata"""
|
|
try:
|
|
text = self.safely_process_value(text)
|
|
|
|
emoji_map = {
|
|
"sampling": "🎯",
|
|
"dimensions": "📏",
|
|
"prompts": "📝",
|
|
"models": "🧠",
|
|
"lora": "🎨",
|
|
"advanced": "⚙️"
|
|
} if include_emojis else {k: "" for k in ["sampling", "dimensions", "prompts", "models", "lora", "advanced"]}
|
|
|
|
output = []
|
|
output.append("=== WebUI Forge/A1111 Generation Parameters ===\n")
|
|
|
|
# Parse lines first to extract metadata_line
|
|
lines = text.strip().split('\n')
|
|
positive_prompt = ""
|
|
negative_prompt = ""
|
|
metadata_line = ""
|
|
|
|
for i, line in enumerate(lines):
|
|
if line.startswith("Negative prompt:"):
|
|
negative_prompt = line.replace("Negative prompt:", "").strip()
|
|
elif re.search(r'Steps:\s*\d+', line):
|
|
metadata_line = line
|
|
elif not metadata_line and not line.startswith("Negative prompt:"):
|
|
positive_prompt += line + " "
|
|
|
|
positive_prompt = positive_prompt.strip()
|
|
|
|
# Extract model name from metadata for display at the top
|
|
model_name_display = "N/A"
|
|
if metadata_line:
|
|
model_match = re.search(r'Model:\s*([^,\n]+)', metadata_line)
|
|
if model_match:
|
|
model_name_display = model_match.group(1).strip()
|
|
|
|
# Display model at the top
|
|
output.append(f"{emoji_map['models']} MODEL: {model_name_display}\n")
|
|
|
|
output.append(f"{emoji_map['prompts']} PROMPTS: |If empty, Check fail-safe below|\n")
|
|
output.append(f" Positive:\n {positive_prompt if positive_prompt else '(empty)'}\n")
|
|
if negative_prompt:
|
|
output.append(f" Negative:\n {negative_prompt}")
|
|
output.append("")
|
|
|
|
if metadata_line:
|
|
params = {}
|
|
patterns = {
|
|
'steps': r'Steps:\s*(\d+)',
|
|
'sampler': r'Sampler:\s*([^,]+)',
|
|
'cfg': r'CFG scale:\s*([\d.]+)',
|
|
'seed': r'Seed:\s*(\d+)',
|
|
'size': r'Size:\s*(\d+x\d+)',
|
|
'model': r'Model:\s*([^,]+)',
|
|
'model_hash': r'Model hash:\s*([^,]+)',
|
|
'denoising': r'Denoising strength:\s*([\d.]+)',
|
|
'clip_skip': r'Clip skip:\s*(\d+)',
|
|
'scheduler': r'Schedule type:\s*([^,]+)',
|
|
'version': r'Version:\s*([^,]+)',
|
|
}
|
|
|
|
for key, pattern in patterns.items():
|
|
match = re.search(pattern, metadata_line)
|
|
if match:
|
|
params[key] = match.group(1).strip()
|
|
|
|
output.append(f"{emoji_map['sampling']} SAMPLING SETTINGS:")
|
|
if 'seed' in params:
|
|
output.append(f" Seed: {params['seed']}")
|
|
if 'steps' in params:
|
|
output.append(f" Steps: {params['steps']}")
|
|
if 'cfg' in params:
|
|
output.append(f" CFG Scale: {params['cfg']}")
|
|
if 'sampler' in params:
|
|
output.append(f" Sampler: {params['sampler']}")
|
|
if 'scheduler' in params:
|
|
output.append(f" Scheduler: {params['scheduler']}")
|
|
if 'denoising' in params:
|
|
output.append(f" Denoise: {params['denoising']}")
|
|
output.append("")
|
|
|
|
if 'size' in params:
|
|
output.append(f"{emoji_map['dimensions']} IMAGE DIMENSIONS:")
|
|
output.append(f" Resolution: {params['size']}")
|
|
output.append("")
|
|
|
|
if 'model' in params or 'model_hash' in params:
|
|
output.append(f"{emoji_map['models']} MODELS & COMPONENTS:")
|
|
if 'model' in params:
|
|
output.append(f" Checkpoint: {params['model']}")
|
|
if 'model_hash' in params:
|
|
output.append(f" Model Hash: {params['model_hash']}")
|
|
output.append("")
|
|
|
|
# Extract LoRAs
|
|
lora_pattern = r'<lora:([^:]+):([\d.]+)>'
|
|
lora_matches = re.findall(lora_pattern, text)
|
|
if lora_matches:
|
|
output.append(f"{emoji_map['lora']} LORA MODELS:")
|
|
for lora_name, lora_strength in lora_matches:
|
|
output.append(f" {lora_name} (Strength: {lora_strength})")
|
|
output.append("")
|
|
|
|
# Advanced settings
|
|
if metadata_line:
|
|
params_check = {}
|
|
patterns = {
|
|
'clip_skip': r'Clip skip:\s*(\d+)',
|
|
'version': r'Version:\s*([^,]+)',
|
|
}
|
|
|
|
for key, pattern in patterns.items():
|
|
match = re.search(pattern, metadata_line)
|
|
if match:
|
|
params_check[key] = match.group(1).strip()
|
|
|
|
if 'clip_skip' in params_check or 'version' in params_check:
|
|
output.append(f"{emoji_map['advanced']} ADVANCED SETTINGS:")
|
|
if 'clip_skip' in params_check:
|
|
output.append(f" Clip Skip: {params_check['clip_skip']}")
|
|
if 'version' in params_check:
|
|
output.append(f" WebUI Version: {params_check['version']}")
|
|
output.append("")
|
|
|
|
return "\n".join(output)
|
|
|
|
except Exception as e:
|
|
print(f"Error parsing WebUI format: {e}")
|
|
return f"Error parsing WebUI metadata: {str(e)}"
|
|
|
|
def parse_comfyui_format(self, metadata_raw, include_emojis=True, image_path=None, file_size_mb=None):
|
|
"""Parse ComfyUI JSON format metadata"""
|
|
try:
|
|
# Safely convert to string FIRST
|
|
metadata_raw = self.safely_process_value(metadata_raw)
|
|
|
|
clean_text = metadata_raw.strip()
|
|
if clean_text.startswith("Prompt: "):
|
|
clean_text = clean_text[8:]
|
|
|
|
try:
|
|
data = json.loads(clean_text)
|
|
except json.JSONDecodeError as e:
|
|
return f"Error parsing JSON: {str(e)}\n\nPlease ensure the input is valid JSON format."
|
|
|
|
# Create output and emoji_map
|
|
output = []
|
|
|
|
emoji_map = {
|
|
"sampling": "🎯",
|
|
"dimensions": "📏",
|
|
"prompts": "📝",
|
|
"models": "🧠",
|
|
"lora": "🎨",
|
|
"advanced": "⚙️"
|
|
} if include_emojis else {k: "" for k in ["sampling", "dimensions", "prompts", "models", "lora", "advanced"]}
|
|
|
|
output.append("=== ComfyUI Generation Parameters ===\n")
|
|
|
|
# Extract latent dimensions
|
|
latent_data = None
|
|
width = 'N/A'
|
|
height = 'N/A'
|
|
batch_size = 'N/A'
|
|
|
|
for node_id, node_data in data.items():
|
|
try:
|
|
class_type = node_data.get("class_type", "")
|
|
if "LatentImage" in class_type or "EmptyLatent" in class_type:
|
|
latent_data = node_data["inputs"]
|
|
width = self.resolve_node_reference(data, latent_data.get('width', 'N/A'))
|
|
height = self.resolve_node_reference(data, latent_data.get('height', 'N/A'))
|
|
batch_size = latent_data.get('batch_size', 'N/A')
|
|
break
|
|
except Exception as e:
|
|
continue
|
|
|
|
if latent_data:
|
|
# Calculate megapixels
|
|
try:
|
|
width_int = int(width) if width != "N/A" else 0
|
|
height_int = int(height) if height != "N/A" else 0
|
|
total_pixels = width_int * height_int
|
|
resolution_mp = float(total_pixels) / 1_000_000
|
|
except:
|
|
resolution_mp = 0.0
|
|
|
|
# Calculate aspect ratio
|
|
def gcd(a, b):
|
|
while b:
|
|
a, b = b, a % b
|
|
return a
|
|
|
|
def find_closest_standard_ratio(decimal_ratio):
|
|
standard_ratios = [
|
|
(1.0, "1:1"), (1.25, "5:4"), (1.33333, "4:3"),
|
|
(1.5, "3:2"), (1.6, "16:10"), (1.66667, "5:3"),
|
|
(1.77778, "16:9"), (1.88889, "17:9"), (2.0, "2:1"),
|
|
(2.33333, "21:9"), (2.35, "2.35:1"), (2.39, "2.39:1"),
|
|
(2.4, "12:5"),
|
|
]
|
|
|
|
closest_diff = float('inf')
|
|
closest_ratio = None
|
|
error_threshold = 0.05
|
|
|
|
for std_value, std_string in standard_ratios:
|
|
diff = abs(std_value - decimal_ratio)
|
|
if diff < closest_diff:
|
|
closest_diff = diff
|
|
closest_ratio = std_string
|
|
|
|
if closest_diff <= error_threshold:
|
|
return closest_ratio
|
|
return None
|
|
|
|
try:
|
|
divisor = gcd(width_int, height_int)
|
|
width_ratio = float(width_int) / divisor
|
|
height_ratio = float(height_int) / divisor
|
|
aspect_ratio_decimal = width_int / height_int
|
|
closest_standard = find_closest_standard_ratio(aspect_ratio_decimal)
|
|
|
|
# Build ratio string
|
|
exact_ratio = f"{int(width_ratio)}:{int(height_ratio)}"
|
|
decimal_ratio = f"{aspect_ratio_decimal:.2f}:1"
|
|
|
|
if closest_standard and closest_standard != exact_ratio:
|
|
ratio_display = f"{exact_ratio} or {decimal_ratio} or ~{closest_standard}"
|
|
else:
|
|
ratio_display = f"{exact_ratio} or {decimal_ratio}"
|
|
except:
|
|
ratio_display = "N/A"
|
|
|
|
# Dimensions, resolution, ratio, and file size
|
|
file_size_str = f" | {file_size_mb:.2f}MB" if file_size_mb is not None else ""
|
|
output.append(f"{width}x{height} | {resolution_mp:.2f}MP | Ratio: {ratio_display}{file_size_str}")
|
|
output.append("")
|
|
|
|
# Extract model name for display at the top
|
|
model_name_display = "N/A"
|
|
for node_id, node_data in data.items():
|
|
class_type = node_data.get("class_type", "")
|
|
inputs = node_data.get("inputs", {})
|
|
|
|
if 'CheckpointLoader' in class_type and 'ckpt_name' in inputs:
|
|
model_name_display = inputs['ckpt_name']
|
|
break
|
|
elif 'UNETLoader' in class_type and 'unet_name' in inputs:
|
|
model_name_display = f"{inputs['unet_name']} (UNET)"
|
|
break
|
|
elif 'Loader' in class_type:
|
|
if 'ckpt_name' in inputs:
|
|
model_name_display = inputs['ckpt_name']
|
|
break
|
|
elif 'unet_name' in inputs:
|
|
model_name_display = f"{inputs['unet_name']} (UNET)"
|
|
break
|
|
|
|
output.append(f"{emoji_map['models']} MODEL: {model_name_display}")
|
|
output.append("")
|
|
|
|
# Extract sampling parameters
|
|
sampling_params = {
|
|
'seed': 'N/A',
|
|
'steps': 'N/A',
|
|
'cfg': 'N/A',
|
|
'sampler': 'N/A',
|
|
'scheduler': 'N/A',
|
|
'denoise': 'N/A'
|
|
}
|
|
|
|
ksampler_data = None
|
|
for node_id, node_data in data.items():
|
|
try:
|
|
if node_data.get("class_type") == "KSampler":
|
|
ksampler_data = node_data.get("inputs", {})
|
|
sampling_params['seed'] = self.safely_process_value(ksampler_data.get('seed', 'N/A'))
|
|
sampling_params['steps'] = self.safely_process_value(ksampler_data.get('steps', 'N/A'))
|
|
sampling_params['cfg'] = self.safely_process_value(ksampler_data.get('cfg', 'N/A'))
|
|
sampling_params['sampler'] = self.safely_process_value(ksampler_data.get('sampler_name', 'N/A'))
|
|
sampling_params['scheduler'] = self.safely_process_value(ksampler_data.get('scheduler', 'N/A'))
|
|
sampling_params['denoise'] = self.safely_process_value(ksampler_data.get('denoise', 'N/A'))
|
|
break
|
|
except Exception as e:
|
|
continue
|
|
|
|
# If no KSampler found, look for distributed nodes
|
|
if not ksampler_data:
|
|
for node_id, node_data in data.items():
|
|
try:
|
|
class_type = node_data.get("class_type", "")
|
|
inputs = node_data.get("inputs", {})
|
|
|
|
if "Noise" in class_type and "noise_seed" in inputs:
|
|
sampling_params['seed'] = self.safely_process_value(inputs.get('noise_seed', 'N/A'))
|
|
elif "Noise" in class_type and "seed" in inputs:
|
|
sampling_params['seed'] = self.safely_process_value(inputs.get('seed', 'N/A'))
|
|
|
|
if "Scheduler" in class_type:
|
|
if 'steps' in inputs:
|
|
sampling_params['steps'] = self.safely_process_value(inputs.get('steps', 'N/A'))
|
|
if 'scheduler' in inputs:
|
|
sampling_params['scheduler'] = self.safely_process_value(inputs.get('scheduler', 'N/A'))
|
|
if 'denoise' in inputs:
|
|
sampling_params['denoise'] = self.safely_process_value(inputs.get('denoise', 'N/A'))
|
|
|
|
if "CFG" in class_type and 'cfg' in inputs:
|
|
sampling_params['cfg'] = self.safely_process_value(inputs.get('cfg', 'N/A'))
|
|
|
|
if "KSamplerSelect" in class_type and 'sampler_name' in inputs:
|
|
sampling_params['sampler'] = self.safely_process_value(inputs.get('sampler_name', 'N/A'))
|
|
elif "Sampler" in class_type and 'sampler_name' in inputs:
|
|
sampling_params['sampler'] = self.safely_process_value(inputs.get('sampler_name', 'N/A'))
|
|
except Exception as e:
|
|
continue
|
|
|
|
if any(v != 'N/A' for v in sampling_params.values()):
|
|
output.append(f"{emoji_map['sampling']} SAMPLING SETTINGS:")
|
|
output.append(f" Seed: {sampling_params['seed']}")
|
|
output.append(f" Steps: {sampling_params['steps']}")
|
|
output.append(f" CFG Scale: {sampling_params['cfg']}")
|
|
output.append(f" Sampler: {sampling_params['sampler']}")
|
|
output.append(f" Scheduler: {sampling_params['scheduler']}")
|
|
output.append(f" Denoise: {sampling_params['denoise']}")
|
|
output.append("")
|
|
|
|
# Extract prompts
|
|
positive_prompt = ""
|
|
negative_prompt = ""
|
|
positive_candidates = []
|
|
negative_candidates = []
|
|
|
|
for node_id, node_data in data.items():
|
|
try:
|
|
class_type = node_data.get("class_type", "")
|
|
inputs = node_data.get("inputs", {})
|
|
|
|
if "CLIPTextEncode" in class_type or "TextEncode" in class_type or "Prompt" in class_type:
|
|
title = node_data.get("_meta", {}).get("title", "").lower()
|
|
text_value = inputs.get("text", "")
|
|
|
|
if self.is_node_reference(text_value):
|
|
continue
|
|
|
|
text_value = self.safely_process_value(text_value)
|
|
if not text_value or text_value == "NA" or not text_value.strip():
|
|
continue
|
|
|
|
# --- Smart content-based detection ---
|
|
negative_keywords = ["watermark", "bad anatomy", "ugly", "deformed", "disfigured", "blurry", "low quality", "worst quality"]
|
|
|
|
# Check explicit title indicators first
|
|
is_explicit_negative = "negative" in title or "neg" in title
|
|
is_explicit_positive = "positive" in title or "pos" in title
|
|
|
|
if is_explicit_negative:
|
|
is_negative = True
|
|
elif is_explicit_positive:
|
|
is_negative = False
|
|
else:
|
|
# Only use content detection if title is ambiguous
|
|
is_negative = any(keyword in text_value.lower() for keyword in negative_keywords)
|
|
|
|
is_positive = not is_negative
|
|
|
|
if is_negative:
|
|
negative_candidates.append(text_value)
|
|
elif is_positive:
|
|
# --- Priority for custom nodes ---
|
|
is_custom_save_node = "SavePositivePromptSG" in class_type or "SaveNegativePromptSG" in class_type
|
|
|
|
if is_custom_save_node:
|
|
positive_candidates.insert(0, text_value) # Put custom nodes first
|
|
else:
|
|
positive_candidates.append(text_value)
|
|
except Exception as e:
|
|
continue
|
|
|
|
if positive_candidates:
|
|
positive_prompt = positive_candidates[0]
|
|
if negative_candidates:
|
|
negative_prompt = negative_candidates[0]
|
|
|
|
output.append(f"{emoji_map['prompts']} PROMPTS: |If empty, Check fail-safe below|\n")
|
|
output.append(f" Positive:\n {positive_prompt if positive_prompt else '(empty)'}\n")
|
|
if negative_prompt:
|
|
output.append(f" Negative:\n {negative_prompt}")
|
|
output.append("")
|
|
|
|
# Extract LoRA models
|
|
loras = []
|
|
lora_files = set()
|
|
processed_keys = set()
|
|
|
|
for node_id, node_data in data.items():
|
|
try:
|
|
class_type = node_data.get("class_type", "")
|
|
inputs = node_data.get("inputs", {})
|
|
|
|
if "lora" in class_type.lower():
|
|
for key in inputs:
|
|
node_key = f"{node_id}_{key}"
|
|
if node_key in processed_keys:
|
|
continue
|
|
|
|
key_lower = key.lower()
|
|
if 'lora' in key_lower and inputs.get(key) not in [None, "", "None"]:
|
|
lora_value = inputs.get(key, "")
|
|
|
|
# Handle dict-wrapped LoRA values
|
|
if isinstance(lora_value, dict):
|
|
if 'on' in lora_value and not lora_value.get('on'):
|
|
processed_keys.add(node_key)
|
|
continue
|
|
if 'lora' in lora_value:
|
|
actual_lora_name = lora_value.get('lora', '')
|
|
actual_strength = lora_value.get('strength', 1.0)
|
|
if actual_lora_name and actual_lora_name != "None":
|
|
display_name = os.path.basename(self.safely_process_value(actual_lora_name))
|
|
loras.append(f" {display_name} (Strength: {actual_strength})")
|
|
lora_files.add(actual_lora_name)
|
|
processed_keys.add(node_key)
|
|
continue
|
|
|
|
# Skip numeric values
|
|
if isinstance(lora_value, (int, float)):
|
|
continue
|
|
if isinstance(lora_value, str) and lora_value.replace('.', '').replace('-', '').replace('_', '').isdigit():
|
|
continue
|
|
if isinstance(lora_value, dict) and lora_value.get('type'):
|
|
processed_keys.add(node_key)
|
|
continue
|
|
|
|
# Find corresponding strength value
|
|
strength = 1.0
|
|
|
|
if '_' in key:
|
|
parts = key.rsplit('_', 1)
|
|
if len(parts) == 2:
|
|
prefix, suffix = parts
|
|
strength_patterns = [
|
|
f"strength_{suffix}",
|
|
f"strength{suffix}",
|
|
f"{prefix}_strength_{suffix}",
|
|
f"str_{suffix}",
|
|
]
|
|
|
|
for pattern in strength_patterns:
|
|
if pattern in inputs:
|
|
strength = inputs.get(pattern, 1.0)
|
|
processed_keys.add(pattern)
|
|
break
|
|
|
|
if strength == 1.0:
|
|
strength_patterns = [
|
|
"strength_model",
|
|
"strength",
|
|
"model_strength",
|
|
"lora_strength"
|
|
]
|
|
|
|
for pattern in strength_patterns:
|
|
if pattern in inputs:
|
|
strength = inputs.get(pattern, 1.0)
|
|
processed_keys.add(pattern)
|
|
break
|
|
|
|
numbers = re.findall(r'\d+', key)
|
|
if numbers and strength == 1.0:
|
|
num = numbers[-1]
|
|
possible_keys = [
|
|
f"strength_{num}",
|
|
f"strength{num}",
|
|
f"str_{num}",
|
|
f"lora_strength_{num}"
|
|
]
|
|
|
|
for possible_key in possible_keys:
|
|
if possible_key in inputs:
|
|
strength = inputs.get(possible_key, 1.0)
|
|
processed_keys.add(possible_key)
|
|
break
|
|
|
|
if lora_value and lora_value != "None":
|
|
display_name = os.path.basename(self.safely_process_value(lora_value))
|
|
loras.append(f" {display_name} (Strength: {strength})")
|
|
lora_files.add(lora_value)
|
|
|
|
processed_keys.add(node_key)
|
|
|
|
except Exception as e:
|
|
print(f"Error processing LoRA: {e}")
|
|
continue
|
|
|
|
# Extract models
|
|
models = {}
|
|
model_keywords = {
|
|
'checkpoint': ['ckpt_name', 'checkpoint_name', 'model_name', 'checkpoint'],
|
|
'unet': ['unet_name', 'unet', 'diffusion_model'],
|
|
'clip': ['clip_name', 'clip_name1', 'clip_name2', 'clip', 'text_encoder'],
|
|
'vae': ['vae_name', 'vae', 'autoencoder'],
|
|
't5': ['t5_name', 't5', 't5xxl'],
|
|
'controlnet': ['control_net_name', 'controlnet_name', 'controlnet'],
|
|
'upscaler': ['upscale_model', 'upscaler_name', 'upscaler'],
|
|
'embeddings': ['embedding_name', 'embedding'],
|
|
'hypernetwork': ['hypernetwork_name', 'hypernetwork'],
|
|
}
|
|
|
|
for node_id, node_data in data.items():
|
|
try:
|
|
class_type = node_data.get("class_type", "")
|
|
inputs = node_data.get("inputs", {})
|
|
|
|
# Skip LoRA nodes
|
|
if "lora" in class_type.lower():
|
|
continue
|
|
|
|
if "loader" in class_type.lower() or "load" in class_type.lower() or any(keyword in class_type.lower() for keyword in ['checkpoint', 'unet', 'clip', 'vae', 'model']):
|
|
for model_type, param_names in model_keywords.items():
|
|
for param_name in param_names:
|
|
if param_name in inputs:
|
|
model_value = inputs.get(param_name)
|
|
|
|
# Skip node references
|
|
if isinstance(model_value, list) and len(model_value) == 2:
|
|
continue
|
|
|
|
# Handle dict-wrapped values
|
|
if isinstance(model_value, dict):
|
|
if 'on' in model_value and not model_value.get('on'):
|
|
continue
|
|
model_value = model_value.get('model', model_value.get('name', model_value.get('value', '')))
|
|
|
|
if isinstance(model_value, list):
|
|
continue
|
|
|
|
# Don't add LoRA files to models
|
|
if model_value in lora_files:
|
|
continue
|
|
|
|
if model_value and model_value != "None":
|
|
display_type = model_type.upper()
|
|
|
|
# Special handling for CLIP
|
|
if model_type == 'clip':
|
|
if param_name == 'clip_name1':
|
|
display_type = "CLIP-1"
|
|
elif param_name == 'clip_name2':
|
|
display_type = "CLIP-2"
|
|
elif param_name.startswith('clip_name') and param_name[-1].isdigit():
|
|
num = param_name.replace('clip_name', '')
|
|
display_type = f"CLIP-{num}"
|
|
else:
|
|
display_type = "CLIP"
|
|
|
|
if display_type not in models:
|
|
models[display_type] = self.safely_process_value(model_value)
|
|
|
|
if model_type != 'clip':
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f"Error processing model node: {e}")
|
|
continue
|
|
|
|
# LORA MODELS section
|
|
if loras:
|
|
output.append(f"{emoji_map['lora']} LORA MODELS:")
|
|
seen = set()
|
|
unique_loras = []
|
|
for lora in loras:
|
|
if lora not in seen:
|
|
seen.add(lora)
|
|
unique_loras.append(lora)
|
|
output.extend(unique_loras)
|
|
output.append("")
|
|
|
|
# MODELS & COMPONENTS section
|
|
if models:
|
|
output.append(f"{emoji_map['models']} MODELS & COMPONENTS:")
|
|
sorted_models = sorted(models.items(), key=lambda x: (
|
|
0 if x[0] == "Checkpoint" else
|
|
1 if x[0] == "UNET" else
|
|
2 if x[0].startswith("CLIP") else
|
|
3 if x[0] == "VAE" else
|
|
4, x[0]
|
|
))
|
|
|
|
for model_type, model_name in sorted_models:
|
|
output.append(f" {model_type}: {model_name}")
|
|
output.append("")
|
|
|
|
|
|
# --- FAIL-SAFE SECTION ---
|
|
output.append("")
|
|
output.append("=== ALL DETECTED TEXT IN WORKFLOW ===")
|
|
output.append("(Fail-safe dump of all text-like values)")
|
|
output.append("")
|
|
|
|
all_text_dump = self.extract_all_text_content(data)
|
|
output.append(all_text_dump)
|
|
|
|
return "\n".join(output)
|
|
|
|
|
|
|
|
except Exception as e:
|
|
print(f"Error in parse_comfyui_format: {e}")
|
|
return f"Error processing parameters: {str(e)}"
|
|
|
|
def resolve_node_reference(self, data, reference):
|
|
"""Helper function to resolve node references like ['124', 0]"""
|
|
if isinstance(reference, list) and len(reference) == 2:
|
|
node_id = str(reference[0])
|
|
if node_id in data:
|
|
node_data = data[node_id]
|
|
if node_data.get("class_type") == "easy int":
|
|
return node_data["inputs"].get("value")
|
|
# If we can't resolve the reference, return a string representation
|
|
return f"[Node Reference: {reference[0]}]"
|
|
return reference
|
|
|
|
def extract_individual_params(self, text, format_type):
|
|
"""Extract individual parameters for output connections"""
|
|
positive = ""
|
|
negative = ""
|
|
|
|
try:
|
|
text = self.safely_process_value(text)
|
|
|
|
if format_type == "comfyui":
|
|
clean_text = text.strip()
|
|
if clean_text.startswith("Prompt: "):
|
|
clean_text = clean_text[8:]
|
|
|
|
try:
|
|
data = json.loads(clean_text)
|
|
|
|
positive_candidates = []
|
|
negative_candidates = []
|
|
|
|
negative_keywords = ["watermark", "bad anatomy", "ugly", "deformed", "disfigured", "blurry", "low quality", "worst quality"]
|
|
|
|
for node_id, node_data in data.items():
|
|
try:
|
|
class_type = node_data.get("class_type", "")
|
|
|
|
if "CLIPTextEncode" in class_type or "TextEncode" in class_type or "Prompt" in class_type:
|
|
title = node_data.get("_meta", {}).get("title", "").lower()
|
|
|
|
text_content = None
|
|
for text_key in ["text", "prompt", "conditioning", "string"]:
|
|
if text_key in node_data["inputs"]:
|
|
text_content = node_data["inputs"].get(text_key)
|
|
break
|
|
|
|
if text_content is None or self.is_node_reference(text_content):
|
|
continue
|
|
|
|
text_content = self.safely_process_value(text_content)
|
|
if not text_content or not text_content.strip():
|
|
continue
|
|
|
|
# --- SMART LOGIC ---
|
|
is_negative_content = any(keyword in text_content.lower() for keyword in negative_keywords)
|
|
is_negative_title = any(neg_word in title for neg_word in ["negative", "neg"])
|
|
is_positive_title = any(pos_word in title for pos_word in ["positive", "pos"])
|
|
|
|
is_negative = is_negative_title or is_negative_content
|
|
if is_positive_title: is_negative = False
|
|
|
|
# --- PRIORITY LOGIC ---
|
|
# Check if this is one of your special "Save" nodes
|
|
is_custom_save_node = "SavePositivePromptSG" in class_type or "SaveNegativePromptSG" in class_type
|
|
|
|
if is_negative:
|
|
if is_custom_save_node:
|
|
negative_candidates.insert(0, text_content) # High Priority
|
|
else:
|
|
negative_candidates.append(text_content) # Normal Priority
|
|
else:
|
|
if is_custom_save_node:
|
|
positive_candidates.insert(0, text_content) # High Priority
|
|
else:
|
|
positive_candidates.append(text_content) # Normal Priority
|
|
|
|
except Exception as e:
|
|
continue
|
|
|
|
if positive_candidates:
|
|
positive = positive_candidates[0]
|
|
if negative_candidates:
|
|
negative = negative_candidates[0]
|
|
|
|
except Exception as e:
|
|
print(f"Error parsing JSON in extract_individual: {e}")
|
|
|
|
elif format_type == "webui":
|
|
lines = text.strip().split('\n')
|
|
metadata_line = ""
|
|
for line in lines:
|
|
if line.startswith("Negative prompt:"):
|
|
negative = line.replace("Negative prompt:", "").strip()
|
|
elif re.search(r'Steps:\s*\d+', line):
|
|
metadata_line = line
|
|
elif not metadata_line and not line.startswith("Negative prompt:"):
|
|
positive += line + " "
|
|
positive = positive.strip()
|
|
|
|
except Exception as e:
|
|
print(f"Error in extract_individual_params: {e}")
|
|
|
|
return positive, negative
|
|
|
|
|
|
|
|
def extract_all_text_content(self, data):
|
|
"""Extract ALL text strings from the workflow as a fail-safe"""
|
|
all_texts = []
|
|
seen_texts = set()
|
|
|
|
try:
|
|
for node_id, node_data in data.items():
|
|
inputs = node_data.get("inputs", {})
|
|
class_type = node_data.get("class_type", "")
|
|
|
|
# Check for common text keys
|
|
candidates = []
|
|
for key in ["text", "string", "prompt", "value", "positive", "negative"]:
|
|
if key in inputs:
|
|
val = inputs[key]
|
|
# Try to resolve if reference
|
|
if self.is_node_reference(val):
|
|
val = self.resolve_node_reference(data, val)
|
|
|
|
val = self.safely_process_value(val)
|
|
if val and val != "N/A" and isinstance(val, str):
|
|
candidates.append(val)
|
|
|
|
# Filter valid text
|
|
for text in candidates:
|
|
clean_text = text.strip()
|
|
# Skip short/irrelevant text or numeric-looking strings if desired
|
|
# But for fail-safe, keep almost everything except empty/N/A
|
|
if not clean_text or clean_text == "N/A" or clean_text.startswith("[Node Reference"):
|
|
continue
|
|
|
|
# Skip generic filenames or internal IDs if they look like it
|
|
if len(clean_text) < 2: continue
|
|
|
|
if clean_text not in seen_texts:
|
|
seen_texts.add(clean_text)
|
|
# Add a label based on node type
|
|
label = f"[{class_type} (ID {node_id})]"
|
|
all_texts.append(f"{label}\n{clean_text}")
|
|
except Exception as e:
|
|
return f"Error extracting all text: {e}"
|
|
|
|
if not all_texts:
|
|
return "No text content found in workflow."
|
|
|
|
return "\n\n------\n\n".join(all_texts)
|
|
|
|
|
|
|
|
# Node registration
|
|
NODE_CLASS_MAPPINGS = {
|
|
"SimpleReadableMetadataSG": SimpleReadableMetadataSG
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"SimpleReadableMetadataSG": "Simple Readable Metadata-SG"
|
|
}
|
|
|
|
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
|