372 lines
17 KiB
Python
372 lines
17 KiB
Python
import torch
|
|
import os
|
|
import folder_paths
|
|
import numpy as np
|
|
import cv2
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from types import SimpleNamespace
|
|
from PIL import Image, ImageSequence
|
|
|
|
class SimpleReadableMetadataVideoSG:
|
|
"""
|
|
Load video, extract frames.
|
|
- Output 1 (String): Full detailed metadata (Prompts, Lora, etc.)
|
|
- Node Display: Concise metadata (Res, Ratio, Model, Seed, Sampler)
|
|
"""
|
|
|
|
CATEGORY = "image/video"
|
|
|
|
@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": {
|
|
"video": (sorted(files), {"video_upload": True}),
|
|
"force_rate": ("INT", {"default": 0, "min": 0, "max": 60, "step": 1, "display": "number", "tooltip": "Target FPS. 0 = Original."}),
|
|
"max_frames": ("INT", {"default": 0, "min": 0, "max": 10000, "step": 1, "display": "number", "tooltip": "Limit total frames. 0 = All."}),
|
|
"resize_long_edge": ("INT", {"default": 0, "min": 0, "max": 4096, "step": 64, "display": "number", "tooltip": "Resize longest side. 0 = Original."}),
|
|
"emoji_in_readable_text": ("BOOLEAN", {"default": True})
|
|
},
|
|
"ui": {
|
|
"text": {"min_width": 450},
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING", "IMAGE", "MASK", "INT", "INT", "STRING", "STRING", "STRING", "STRING", "INT")
|
|
RETURN_NAMES = ("Simple_Readable_Metadata", "frames", "mask", "frame_count", "fps", "filename_text", "metadata_raw", "Positive_Prompt", "Negative_Prompt", "seed")
|
|
FUNCTION = "load_video_analyze"
|
|
OUTPUT_NODE = True
|
|
|
|
@classmethod
|
|
def IS_CHANGED(cls, video, force_rate, max_frames, resize_long_edge, emoji_in_readable_text):
|
|
video_path = folder_paths.get_annotated_filepath(video)
|
|
if os.path.exists(video_path):
|
|
stat = os.stat(video_path)
|
|
return f"{video}_{stat.st_mtime}_{stat.st_size}_{force_rate}_{max_frames}_{resize_long_edge}"
|
|
return "N/A"
|
|
|
|
@classmethod
|
|
def VALIDATE_INPUTS(cls, video, **kwargs):
|
|
if not folder_paths.exists_annotated_filepath(video):
|
|
return "Invalid video file: {}".format(video)
|
|
return True
|
|
|
|
# ============================METADATA EXTRACTION UTILS====================================
|
|
|
|
def extract_raw_video_metadata(self, video_path):
|
|
"""Smart metadata extraction based on file type."""
|
|
ext = os.path.splitext(video_path)[1].lower()
|
|
|
|
# STRATEGY A: PIL (WebP, APNG, GIF)
|
|
if ext in ['.webp', '.png', '.gif']:
|
|
try:
|
|
with Image.open(video_path) as img:
|
|
if 'prompt' in img.info: return img.info['prompt']
|
|
if 'workflow' in img.info: return json.dumps({"workflow_only": img.info['workflow']})
|
|
if 'exif' in img.info:
|
|
try:
|
|
exif_data = img.info['exif']
|
|
if isinstance(exif_data, bytes):
|
|
exif_str = exif_data.decode('utf-8', errors='ignore')
|
|
if "prompt:" in exif_str: return exif_str.split("prompt:")[1].split("\x00")[0]
|
|
except: pass
|
|
except: pass
|
|
|
|
# STRATEGY B: FFPROBE
|
|
try:
|
|
command = ['ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_format', '-show_streams', video_path]
|
|
result = subprocess.run(command, capture_output=True, text=True, encoding='utf-8')
|
|
if result.returncode == 0:
|
|
data = json.loads(result.stdout)
|
|
sources = []
|
|
if 'format' in data and 'tags' in data['format']: sources.append(data['format']['tags'])
|
|
if 'streams' in data:
|
|
for stream in data['streams']:
|
|
if stream.get('codec_type') == 'video' and 'tags' in stream: sources.append(stream['tags'])
|
|
|
|
for tags in sources:
|
|
for key in ['comment', 'prompt', 'workflow', 'description', 'user_data']:
|
|
for t_key, t_val in tags.items():
|
|
if t_key.lower() == key:
|
|
clean_val = t_val.strip()
|
|
if clean_val.startswith('{') or "Prompt:" in clean_val: return clean_val
|
|
except: pass
|
|
return None
|
|
|
|
def get_concise_display_info(self, metadata_raw):
|
|
"""Extracts specific fields for the CONCISE NODE DISPLAY (UI)."""
|
|
info = {
|
|
"model": "N/A",
|
|
"seed": "N/A",
|
|
"steps": "N/A",
|
|
"cfg": "N/A",
|
|
"sampler": "N/A",
|
|
"scheduler": "N/A"
|
|
}
|
|
|
|
if not metadata_raw: return info
|
|
|
|
try:
|
|
if metadata_raw.strip().startswith("Prompt:"): metadata_raw = metadata_raw.strip()[7:]
|
|
data = json.loads(metadata_raw)
|
|
|
|
# 1. MODEL EXTRACTION
|
|
|
|
# A) Workflow format (Nodes list)
|
|
if "nodes" in data and isinstance(data["nodes"], list):
|
|
for node in data["nodes"]:
|
|
nt = node.get("type", "").lower()
|
|
|
|
# SKIP undesired loaders
|
|
if "lora" in nt or "clip" in nt or "control" in nt or "vae" in nt:
|
|
continue
|
|
|
|
# Look for Checkpoint or Loader
|
|
if "checkpoint" in nt or "loader" in nt:
|
|
vals = node.get("widgets_values")
|
|
if vals and isinstance(vals, list) and len(vals) > 0:
|
|
val = str(vals[0]).lower()
|
|
if ".safetensors" in val or ".ckpt" in val or ".gguf" in val or ".pt" in val:
|
|
info["model"] = vals[0]
|
|
break
|
|
|
|
# B) Prompt format (Dict of nodes)
|
|
if info["model"] == "N/A":
|
|
for k, v in data.items():
|
|
ct = v.get("class_type", "").lower()
|
|
|
|
# SKIP undesired loaders
|
|
if "lora" in ct or "clip" in ct or "control" in ct or "vae" in ct:
|
|
continue
|
|
|
|
inputs = v.get("inputs", {})
|
|
|
|
# CheckpointLoader (Standard)
|
|
if "ckpt_name" in inputs:
|
|
info["model"] = inputs["ckpt_name"]
|
|
break
|
|
|
|
# UNETLoader
|
|
if "unet_name" in inputs:
|
|
info["model"] = f"{inputs['unet_name']} (UNET)"
|
|
break
|
|
|
|
# GGUF Loader
|
|
if "gguf_name" in inputs:
|
|
info["model"] = f"{inputs['gguf_name']} (GGUF)"
|
|
break
|
|
|
|
# Generic Model/Checkpoint keys (fallback)
|
|
if "model_name" in inputs:
|
|
info["model"] = inputs["model_name"]
|
|
break
|
|
|
|
if "checkpoint" in inputs and isinstance(inputs["checkpoint"], str):
|
|
info["model"] = inputs["checkpoint"]
|
|
break
|
|
|
|
# 2. SAMPLING
|
|
if isinstance(data, dict) and "nodes" not in data:
|
|
for k, v in data.items():
|
|
if "Sampler" in v.get("class_type", ""):
|
|
inputs = v.get("inputs", {})
|
|
info["seed"] = inputs.get("seed", inputs.get("noise_seed", "N/A"))
|
|
info["steps"] = inputs.get("steps", "N/A")
|
|
info["cfg"] = inputs.get("cfg", "N/A")
|
|
info["sampler"] = inputs.get("sampler_name", "N/A")
|
|
info["scheduler"] = inputs.get("scheduler", "N/A")
|
|
break
|
|
except: pass
|
|
return info
|
|
|
|
def extract_full_readable_text(self, metadata_raw, include_emojis=True):
|
|
"""Generates the FULL DETAILED text output for the STRING output."""
|
|
if not metadata_raw: return "No metadata found."
|
|
try:
|
|
if metadata_raw.strip().startswith("Prompt:"): metadata_raw = metadata_raw.strip()[7:]
|
|
data = json.loads(metadata_raw)
|
|
|
|
lines = []
|
|
emoji_map = {"models": "🧠", "sampling": "🎯", "prompts": "📝", "lora": "🎨"} if include_emojis else {k: "" for k in ["models", "sampling", "prompts", "lora"]}
|
|
|
|
# Model (Reuse concise logic to get the MAIN model name)
|
|
info = self.get_concise_display_info(metadata_raw)
|
|
lines.append(f"{emoji_map['models']} MODEL: {info['model']}\n")
|
|
|
|
# Sampling
|
|
if isinstance(data, dict) and "nodes" not in data:
|
|
for k, v in data.items():
|
|
if "Sampler" in v.get("class_type", ""):
|
|
inputs = v["inputs"]
|
|
lines.append(f"{emoji_map['sampling']} SAMPLING SETTINGS:")
|
|
lines.append(f" Seed : {inputs.get('seed', inputs.get('noise_seed', 'N/A'))}")
|
|
lines.append(f" Steps : {inputs.get('steps', 'N/A')}")
|
|
lines.append(f" CFG Scale : {inputs.get('cfg', 'N/A')}")
|
|
lines.append(f" Sampler : {inputs.get('sampler_name', 'N/A')}")
|
|
lines.append(f" Scheduler : {inputs.get('scheduler', 'N/A')}\n")
|
|
break
|
|
|
|
# Prompts
|
|
lines.append(f"{emoji_map['prompts']} PROMPTS:")
|
|
pos, neg = [], []
|
|
if isinstance(data, dict) and "nodes" not in data:
|
|
for k, v in data.items():
|
|
if "CLIPTextEncode" in v.get("class_type", ""):
|
|
t = v["inputs"].get("text", "").strip()
|
|
if "negative" in v.get("_meta", {}).get("title", "").lower(): neg.append(t)
|
|
else: pos.append(t)
|
|
lines.append(f" Positive: {', '.join(pos) if pos else '(empty)'}")
|
|
lines.append(f" Negative: {', '.join(neg) if neg else '(empty)'}\n")
|
|
|
|
# Components
|
|
lines.append(f"{emoji_map['models']} MODELS & COMPONENTS:")
|
|
if isinstance(data, dict) and "nodes" not in data:
|
|
for k, v in data.items():
|
|
ct = v.get("class_type", "")
|
|
inputs = v.get("inputs", {})
|
|
if "CheckpointLoader" in ct: lines.append(f" Checkpoint: {inputs.get('ckpt_name')}")
|
|
if "LoaderGGUF" in ct: lines.append(f" GGUF: {inputs.get('gguf_name')}")
|
|
if "LoraLoader" in ct: lines.append(f" LoRA: {inputs.get('lora_name')} (Str: {inputs.get('strength_model')})")
|
|
if "VAELoader" in ct: lines.append(f" VAE: {inputs.get('vae_name')}")
|
|
|
|
return "\n".join(lines)
|
|
except:
|
|
return metadata_raw
|
|
|
|
def extract_individual_params(self, metadata_raw):
|
|
pos, neg, seed = "", "", 0
|
|
try:
|
|
if metadata_raw and metadata_raw.strip().startswith("{"):
|
|
data = json.loads(metadata_raw)
|
|
if isinstance(data, dict) and "nodes" not in data:
|
|
for k, v in data.items():
|
|
if "Sampler" in v.get("class_type", ""):
|
|
seed = int(v["inputs"].get("seed", v["inputs"].get("noise_seed", 0)))
|
|
break
|
|
for k, v in data.items():
|
|
if "CLIPTextEncode" in v.get("class_type", ""):
|
|
inputs = v.get("inputs", {})
|
|
title = v.get("_meta", {}).get("title", "").lower()
|
|
text = inputs.get("text", "")
|
|
if "negative" in title: neg += text + " "
|
|
else: pos += text + " "
|
|
except: pass
|
|
return pos.strip(), neg.strip(), seed
|
|
|
|
def gcd(self, a, b):
|
|
while b: a, b = b, a % b
|
|
return a
|
|
|
|
def find_closest_standard_ratio(self, 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.77778, '16:9'), (2.33333, '21:9')]
|
|
closest, min_diff = None, float('inf')
|
|
for val, lbl in standard_ratios:
|
|
diff = abs(val - decimal_ratio)
|
|
if diff < min_diff: min_diff, closest = diff, lbl
|
|
return closest if min_diff <= 0.05 else None
|
|
|
|
|
|
# ==================================== MAIN EXECUTION========================================
|
|
|
|
def load_video_analyze(self, video, force_rate, max_frames, resize_long_edge, emoji_in_readable_text=True):
|
|
video_path = folder_paths.get_annotated_filepath(video)
|
|
cap = cv2.VideoCapture(video_path)
|
|
if not cap.isOpened(): raise RuntimeError(f"Could not open video: {video_path}")
|
|
|
|
original_fps = cap.get(cv2.CAP_PROP_FPS)
|
|
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
|
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
|
|
|
frames = []
|
|
step = 1
|
|
if force_rate > 0 and force_rate < original_fps: step = max(1, int(original_fps / force_rate))
|
|
effective_fps = original_fps / step if step > 1 else original_fps
|
|
if force_rate > 0: effective_fps = force_rate
|
|
|
|
count = 0
|
|
output_count = 0
|
|
while True:
|
|
ret, frame = cap.read()
|
|
if not ret: break
|
|
if count % step == 0:
|
|
if resize_long_edge > 0:
|
|
h, w = frame.shape[:2]
|
|
if max(h, w) > resize_long_edge:
|
|
scale = resize_long_edge / max(h, w)
|
|
frame = cv2.resize(frame, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA)
|
|
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
|
|
frames.append(torch.from_numpy(frame))
|
|
output_count += 1
|
|
if max_frames > 0 and output_count >= max_frames: break
|
|
count += 1
|
|
cap.release()
|
|
|
|
if not frames: raise RuntimeError("No frames extracted.")
|
|
output_frames = torch.stack(frames)
|
|
mask = torch.ones((output_frames.shape[0], output_frames.shape[1], output_frames.shape[2]), dtype=torch.float32)
|
|
|
|
# --- METADATA LOGIC ---
|
|
|
|
# 1. Physical Stats
|
|
resolution_mp = (width * height) / 1_000_000
|
|
try: file_size_mb = os.path.getsize(video_path) / (1024 * 1024)
|
|
except: file_size_mb = 0.0
|
|
|
|
divisor = self.gcd(width, height)
|
|
ar_dec = width / height
|
|
std_ratio = self.find_closest_standard_ratio(ar_dec)
|
|
ratio_str = f"{width//divisor}:{height//divisor}"
|
|
if std_ratio and std_ratio != ratio_str: ratio_str += f" or {std_ratio}"
|
|
else: ratio_str += f" or {ar_dec:.2f}:1"
|
|
|
|
# 2. Extract Raw Metadata
|
|
metadata_raw = self.extract_raw_video_metadata(video_path)
|
|
|
|
# 3. GENERATE UI DISPLAY TEXT
|
|
gen_info = self.get_concise_display_info(metadata_raw)
|
|
ui_lines = []
|
|
ui_lines.append(f"{width}x{height} | {resolution_mp:.2f}MP")
|
|
ui_lines.append(f"Ratio: {ratio_str}")
|
|
ui_lines.append(f"File Size: {file_size_mb:.2f}MB")
|
|
ui_lines.append("")
|
|
ui_lines.append(f"Model: {gen_info['model']}")
|
|
ui_lines.append(f"Seed: {gen_info['seed']} | Steps: {gen_info['steps']} | CFG: {gen_info['cfg']}")
|
|
ui_lines.append(f"Sampler: {gen_info['sampler']} | Scheduler: {gen_info['scheduler']}")
|
|
|
|
# 4. GENERATE OUTPUT STRING (Full/Rich)
|
|
full_readable_text = f"=== Video Information ===\nFilename: {os.path.basename(video_path)}\n{width}x{height} | {resolution_mp:.2f}MP | {file_size_mb:.2f}MB\nFPS: {int(effective_fps)} | Duration: {(count/original_fps if original_fps else 0):.1f}s\n\n"
|
|
if metadata_raw:
|
|
full_readable_text += self.extract_full_readable_text(metadata_raw, emoji_in_readable_text)
|
|
else:
|
|
full_readable_text += "(No embedded ComfyUI generation metadata detected in file)"
|
|
|
|
# Return Values
|
|
pos, neg, seed = self.extract_individual_params(metadata_raw)
|
|
|
|
return {
|
|
"ui": {"text": ui_lines},
|
|
"result": (
|
|
full_readable_text,
|
|
output_frames,
|
|
mask,
|
|
len(frames),
|
|
int(effective_fps),
|
|
os.path.basename(video_path),
|
|
metadata_raw if metadata_raw else "",
|
|
pos,
|
|
neg,
|
|
seed
|
|
)
|
|
}
|
|
|
|
NODE_CLASS_MAPPINGS = {
|
|
"SimpleReadableMetadataVideoSG": SimpleReadableMetadataVideoSG
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"SimpleReadableMetadataVideoSG": "Simple Readable Metadata (Video)-SG"
|
|
}
|