""" Path Utilities Module Provides helper functions for path normalization, symlink resolution, and directory matching to be shared across all modules. """ import hashlib import json import os import re from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple from .type_utils import normalize_alphanumeric_key _log = None MODEL_RESOLVER_METADATA_SUFFIX = ".modelresolver.json" MODEL_RESOLVER_METADATA_SCHEMA = "comfyui-model-resolver" MODEL_RESOLVER_METADATA_SCHEMA_VERSION = 1 _EXTERNAL_METADATA_PREFERRED_FIELDS = { "notes", "favorite", "exclude", "skip_metadata_refresh", "db_checked", "description", "model_description", "modelDescription", } _MERGED_METADATA_LIST_FIELDS = { "tags", "trained_words", "trainedWords", } def _get_log(): """Return a lazily-created module logger (avoids circular imports).""" global _log if _log is None: from .log_system import create_module_logger _log = create_module_logger(__name__) return _log def get_path_identity(path: str) -> str: """Return a stable path identity across symlinks/junctions.""" if not path: return "" try: return os.path.normcase(os.path.realpath(os.path.abspath(path))) except (OSError, ValueError): return os.path.normcase(os.path.abspath(path)) def get_model_path_identity(path_value: Any) -> str: """Return a stable filesystem identity for a model path value.""" path = str(path_value or "").strip() return get_path_identity(path) def normalize_absolute_path(path_value: Any, path_module=os.path) -> str: """Return an absolute, normalized and symlink-resolved path.""" return path_module.realpath( path_module.abspath(path_module.normpath(str(path_value))) ) def normalize_metadata_file_path(path_value: Any) -> str: """Normalize metadata paths to portable forward-slash separators.""" return str(path_value or "").replace(os.sep, "/") def get_path_abs(path_value: Any) -> str: """Return absolute path safely, handling OSError/ValueError.""" try: return os.path.abspath(str(path_value or "")) except (OSError, ValueError): return str(path_value or "") def get_path_key(path_value: Any) -> str: """Return a normalized absolute path key for comparison.""" text = str(path_value or "").strip() if not text: return "" try: return os.path.normcase(os.path.abspath(os.path.normpath(text))) except (OSError, ValueError): return os.path.normcase(os.path.normpath(text)) def is_path_within(path_value: Any, root_value: Any) -> bool: """Check if a path is located within a root directory.""" if not path_value or not root_value: return False try: path_key = get_path_identity(str(path_value)) root_key = get_path_identity(str(root_value)) return os.path.commonpath([path_key, root_key]) == root_key except Exception: return False def _get_folder_paths_module(folder_paths_module: Optional[Any] = None) -> Optional[Any]: if folder_paths_module is not None: return folder_paths_module try: import folder_paths return folder_paths except ImportError: return None def normalize_string_values(value: Any) -> List[str]: """Normalize an external string collection without iterating scalar text.""" if isinstance(value, str): return [value] if value else [] if not isinstance(value, (list, tuple, set, frozenset)): return [] return [item for item in value if isinstance(item, str) and item] def normalize_folder_path_values(value: Any) -> List[str]: """Normalize ComfyUI folder-path values across supported registry shapes.""" if isinstance(value, (list, tuple)) and value: first_item = value[0] if isinstance(first_item, (list, tuple, set, frozenset)): value = first_item return normalize_string_values(value) def get_configured_model_roots( folder_paths_module: Optional[Any] = None, *, skip_keys: Optional[Iterable[str]] = None, ) -> List[str]: """Return all configured ComfyUI model roots, including external model folders.""" fp = _get_folder_paths_module(folder_paths_module) if fp is None: return [] skipped = {str(key).strip().lower() for key in (skip_keys or {"custom_nodes", "configs"})} names = [] folder_names_and_paths = getattr(fp, "folder_names_and_paths", None) if isinstance(folder_names_and_paths, dict): names = list(folder_names_and_paths.keys()) else: try: get_folder_names = getattr(fp, "get_folder_names", None) if callable(get_folder_names): names = list(get_folder_names() or []) except Exception: names = [] roots: List[str] = [] seen = set() for raw_name in names: name = str(raw_name or "").strip() if not name or name.lower() in skipped: continue paths = [] try: get_folder_paths = getattr(fp, "get_folder_paths", None) if callable(get_folder_paths): paths = normalize_folder_path_values(get_folder_paths(name)) except Exception: paths = [] if not paths and isinstance(folder_names_and_paths, dict): paths = normalize_folder_path_values(folder_names_and_paths.get(name)) for path in paths: if not path: continue absolute_path = os.path.abspath(os.path.normpath(str(path))) identity = get_path_identity(absolute_path) if identity in seen: continue seen.add(identity) roots.append(absolute_path) return roots def is_path_in_configured_model_roots( path_value: Any, folder_paths_module: Optional[Any] = None, ) -> bool: """Return True when a path is inside a configured ComfyUI model root.""" if not path_value: return False return any( is_path_within(path_value, root) for root in get_configured_model_roots(folder_paths_module) ) def prefer_local_base_directory( candidate: str, current: str, preferred_directory: str = "", comfy_root: str = "", ) -> bool: """Determine if candidate directory should be preferred over current directory.""" if not current: return True if not candidate: return False candidate_key = get_path_key(candidate) current_key = get_path_key(current) preferred_key = get_path_key(preferred_directory) if preferred_key: if candidate_key == preferred_key and current_key != preferred_key: return True if current_key == preferred_key and candidate_key != preferred_key: return False if comfy_root: candidate_is_external = not is_path_within(candidate, comfy_root) current_is_external = not is_path_within(current, comfy_root) if candidate_is_external != current_is_external: return candidate_is_external candidate_is_canonical = candidate_key == get_path_identity(candidate) current_is_canonical = current_key == get_path_identity(current) if candidate_is_canonical != current_is_canonical: return candidate_is_canonical return False def dedupe_local_base_directories( paths: Optional[list | tuple] = None, preferred_directory: str = "", comfy_root: str = "", ) -> list: """Deduplicate a list of directory paths, resolving symlinks/junctions.""" if paths is None: path_values = () elif isinstance(paths, (list, tuple)): path_values = paths else: raise TypeError("paths must be a list or tuple") by_identity = {} ordered_identities = [] for path in path_values: if not path or not os.path.isdir(path): continue path_abs = get_path_abs(path) path_identity = get_path_identity(path_abs) if not path_identity: continue current = by_identity.get(path_identity) if not current: by_identity[path_identity] = path_abs ordered_identities.append(path_identity) elif prefer_local_base_directory( path_abs, current, preferred_directory, comfy_root, ): by_identity[path_identity] = path_abs return [by_identity[key] for key in ordered_identities] def write_json_atomic( file_path: str, data: Any, indent: Optional[int] = None, separators: Optional[Tuple[str, str]] = None, ) -> None: """Safely write data to a JSON file using a temporary file and atomic replace.""" import json import tempfile dir_name = os.path.dirname(file_path) if dir_name: os.makedirs(dir_name, exist_ok=True) fd, tmp_path = tempfile.mkstemp( prefix=f".{os.path.basename(file_path)}.", suffix=".tmp", dir=dir_name or None, text=True, ) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, indent=indent, separators=separators, ensure_ascii=False) if indent is not None: f.write("\n") os.replace(tmp_path, file_path) except Exception: try: os.remove(tmp_path) except OSError: pass raise def get_filename_from_path(path: Any) -> str: """Return the filename/basename of a path, handling both forward and backward slashes.""" if not path: return "" return str(path).replace("\\", "/").split("/")[-1] # Path to metadata directory METADATA_DIR = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "metadata" ) class HashCalculationCancelled(Exception): """Exception raised when file SHA256 calculation is cancelled by the caller.""" pass SAFETENSORS_HEADER_MAX_BYTES = 100 * 1024 * 1024 _NON_SHA256_HASH_KEY_PARTS = ( "blake3", "crc32", "md5", "sha1", "ss_new_sd_model_hash", "ss_sd_model_hash", "sshs_model_hash", "sshs_legacy_hash", "autov3", "autov1", "autov2", ) _BASE_MODEL_ALIAS_CACHE: Optional[List[Tuple[str, str]]] = None SAFETENSORS_METADATA_TAG_LIMIT = 40 SAFETENSORS_METADATA_KEY_LIMIT = 80 SAFETENSORS_METADATA_TEXT_MAX_CHARS = 20000 SAFETENSORS_METADATA_THUMBNAIL_MAX_CHARS = 512 * 1024 def read_safetensors_header( file_path: str, max_header_size: int = SAFETENSORS_HEADER_MAX_BYTES, ) -> Optional[Dict[str, Any]]: """Read and parse a safetensors JSON header without touching tensor payloads.""" if not file_path or not str(file_path).lower().endswith(".safetensors"): return None try: file_size = os.path.getsize(file_path) except (OSError, ValueError): return None if file_size < 8: return None try: with open(file_path, "rb") as f: header_size_bytes = f.read(8) if len(header_size_bytes) != 8: return None header_size = int.from_bytes( header_size_bytes, byteorder="little", signed=False, ) if ( header_size <= 0 or header_size > max_header_size or header_size > file_size - 8 ): return None header_bytes = f.read(header_size) if len(header_bytes) != header_size: return None header_json = json.loads(header_bytes.decode("utf-8")) if isinstance(header_json, dict): return header_json except Exception as e: _get_log().debug(f"Could not read safetensors header for {file_path}: {e}") return None def extract_safetensors_header_sha256( file_path: str, max_header_size: int = SAFETENSORS_HEADER_MAX_BYTES, ) -> Optional[str]: """Return an embedded SHA256 from a safetensors metadata header when present.""" from .type_utils import normalize_sha256 header_json = read_safetensors_header(file_path, max_header_size=max_header_size) if not isinstance(header_json, dict): return None metadata = header_json.get("__metadata__") if not isinstance(metadata, dict): return None for key in ( "modelspec.hash.sha256", "modelspec.hash_sha256", "sha256", "SHA256", ): sha256 = normalize_sha256(metadata.get(key)) if sha256: return sha256 for key, value in metadata.items(): key_lower = str(key).lower() if not ( "sha256" in key_lower or "hash" in key_lower or "civitai" in key_lower ): continue if any(part in key_lower for part in _NON_SHA256_HASH_KEY_PARTS): continue sha256 = normalize_sha256(value) if sha256: return sha256 return None def _load_base_model_aliases() -> List[Tuple[str, str]]: """Return normalized base-model aliases ordered from most specific first.""" global _BASE_MODEL_ALIAS_CACHE if _BASE_MODEL_ALIAS_CACHE is not None: return _BASE_MODEL_ALIAS_CACHE aliases: List[Tuple[str, str]] = [] try: from .sources.popular import load_raw_base_models base_models = load_raw_base_models() for entry in base_models: if not isinstance(entry, dict): continue name = str(entry.get("name") or "").strip() if not name: continue raw_aliases = entry.get("aliases") or [] if isinstance(raw_aliases, str): raw_aliases = [raw_aliases] if not isinstance(raw_aliases, (list, tuple, set)): raw_aliases = [] for raw_alias in [name, *raw_aliases]: alias = normalize_alphanumeric_key(raw_alias) if alias: aliases.append((alias, name)) except Exception as e: _get_log().debug(f"Could not load base model aliases: {e}") aliases.sort(key=lambda item: len(item[0]), reverse=True) _BASE_MODEL_ALIAS_CACHE = aliases return aliases def clear_base_model_alias_cache() -> None: """Clear the base model alias cache.""" global _BASE_MODEL_ALIAS_CACHE _BASE_MODEL_ALIAS_CACHE = None def _canonical_base_model_from_text(value: Any) -> str: normalized = normalize_alphanumeric_key(value) if not normalized: return "" for alias, name in _load_base_model_aliases(): if normalized == alias or (len(alias) >= 4 and alias in normalized): return name return "" def infer_safetensors_base_model( file_path: str, max_tensor_keys: int = 500, ) -> str: """Infer a likely base model from safetensors metadata and tensor names.""" header_json = read_safetensors_header(file_path) if not isinstance(header_json, dict): return "" metadata = header_json.get("__metadata__") if not isinstance(metadata, dict): metadata = {} architecture = str( metadata.get("modelspec.architecture") or metadata.get("architecture") or metadata.get("ss_base_model_version") or "" ).lower() if architecture: if "stable-diffusion-xl" in architecture or "sdxl" in architecture: return "SDXL 1.0" if ( "stable-diffusion-v1" in architecture or "stable-diffusion-1" in architecture or "runwayml/stable-diffusion-v1-5" in architecture or "sd1.5" in architecture or "sd 1.5" in architecture ): return "SD 1.5" if "stable-diffusion-v2" in architecture or "sd2" in architecture: return "SD 2.1" if ( "stable-diffusion-3" in architecture or "stable-diffusion-v3" in architecture or "sd3" in architecture ): return "SD 3" if "flux" in architecture: if "krea" in architecture: return "Flux.1 Krea" if "schnell" in architecture: return "Flux.1 S" if "kontext" in architecture: return "Flux.1 Kontext" return "Flux.1 D" alias_match = _canonical_base_model_from_text(architecture) if alias_match: return alias_match tensor_keys = [ str(key) for key in header_json if key != "__metadata__" ][:max_tensor_keys] keys_text = " ".join(tensor_keys) if "double_blocks.0.img_attn" in keys_text or "img_in.weight" in keys_text: return "Flux.1 D" if "joint_blocks.0.x_block" in keys_text: return "SD 3" if ( "conditioner.embedders.1.model" in keys_text or "label_emb.0.0.weight" in keys_text ): return "SDXL 1.0" if ( "cond_stage_model.transformer.text_model" in keys_text or "model.diffusion_model.input_blocks.0.0.weight" in keys_text ): return "SD 1.5" return "" def _parse_safetensors_metadata_value(value: Any) -> Any: if not isinstance(value, str): return value text = value.strip() if not text: return "" if text[:1] in {"{", "["}: try: return json.loads(text) except Exception: return text return text def _metadata_get(metadata: Dict[str, Any], *keys: str) -> Any: if not isinstance(metadata, dict): return None for key in keys: if key in metadata: return metadata.get(key) lowered = {str(key).lower(): key for key in metadata} for key in keys: original_key = lowered.get(str(key).lower()) if original_key is not None: return metadata.get(original_key) return None def _metadata_text( metadata: Dict[str, Any], *keys: str, max_chars: int = SAFETENSORS_METADATA_TEXT_MAX_CHARS, ) -> str: value = _parse_safetensors_metadata_value(_metadata_get(metadata, *keys)) if value is None or isinstance(value, (dict, list, tuple, set)): return "" text = str(value).strip() if not text or text.lower() in {"none", "null", "undefined"}: return "" if max_chars and len(text) > max_chars: return text[:max_chars].rstrip() return text def _dedupe_strings(values: Iterable[Any], limit: int = 0) -> List[str]: result: List[str] = [] seen = set() for value in values: text = str(value or "").strip() if not text: continue key = text.lower() if key in seen: continue seen.add(key) result.append(text) if limit and len(result) >= limit: break return result def _metadata_string_list(value: Any, limit: int = SAFETENSORS_METADATA_TAG_LIMIT) -> List[str]: parsed = _parse_safetensors_metadata_value(value) items: List[Any] = [] def collect(item: Any) -> None: item = _parse_safetensors_metadata_value(item) if item is None: return if isinstance(item, dict): for key in ( "tags", "model_tags", "modelTags", "trained_words", "trainedWords", "trigger", "trigger_phrase", "values", "words", ): if key in item: collect(item.get(key)) return for value_item in item.values(): collect(value_item) return if isinstance(item, (list, tuple, set)): for value_item in item: collect(value_item) return if isinstance(item, str): parts = re.split(r"[\n,|;]+", item) items.extend(part.strip() for part in parts if part.strip()) return items.append(item) collect(parsed) return _dedupe_strings(items, limit=limit) def _collect_tag_frequency_tags(value: Any, limit: int = 20) -> List[str]: parsed = _parse_safetensors_metadata_value(value) counts: Dict[str, float] = {} def add_tag(tag: Any, count: Any) -> None: text = str(tag or "").strip() if ( not text or len(text) > 80 or "/" in text or "\\" in text or text.lower().endswith((".safetensors", ".ckpt", ".pt")) ): return try: amount = float(count) except (TypeError, ValueError): amount = 1.0 counts[text] = counts.get(text, 0.0) + max(amount, 0.0) def walk(item: Any) -> None: item = _parse_safetensors_metadata_value(item) if isinstance(item, dict): numeric_values = True for value_item in item.values(): try: float(value_item) except (TypeError, ValueError): numeric_values = False break if numeric_values: for key, value_item in item.items(): add_tag(key, value_item) return for value_item in item.values(): walk(value_item) elif isinstance(item, (list, tuple, set)): for value_item in item: walk(value_item) walk(parsed) ordered = sorted(counts.items(), key=lambda item: (-item[1], item[0].lower())) return [tag for tag, _count in ordered[:limit]] def _limited_metadata_value(value: Any, *, max_chars: int = 1000, max_items: int = 8) -> Any: parsed = _parse_safetensors_metadata_value(value) if isinstance(parsed, str): text = parsed.strip() return text[:max_chars].rstrip() if len(text) > max_chars else text if isinstance(parsed, (bool, int, float)) or parsed is None: return parsed if isinstance(parsed, dict): limited: Dict[str, Any] = {} for index, (key, item_value) in enumerate(parsed.items()): if index >= max_items: limited["_truncated"] = True break limited[str(key)] = _limited_metadata_value( item_value, max_chars=max_chars, max_items=max_items, ) return limited if isinstance(parsed, (list, tuple, set)): values = list(parsed) limited_list = [ _limited_metadata_value(item, max_chars=max_chars, max_items=max_items) for item in values[:max_items] ] if len(values) > max_items: limited_list.append({"_truncated": True}) return limited_list return str(parsed) def _metadata_preview_url(metadata: Dict[str, Any]) -> str: value = _metadata_get( metadata, "modelspec.thumbnail", "thumbnail", "thumbnail_url", "thumbnailUrl", "preview_url", "previewUrl", "ssmd_cover_images", ) parsed = _parse_safetensors_metadata_value(value) def extract(item: Any) -> str: item = _parse_safetensors_metadata_value(item) if isinstance(item, str): text = item.strip() if not text or len(text) > SAFETENSORS_METADATA_THUMBNAIL_MAX_CHARS: return "" if text.startswith(("http://", "https://", "data:image/")): return text return "" if isinstance(item, dict): for key in ("url", "src", "image", "thumbnail", "preview", "data"): preview = extract(item.get(key)) if preview: return preview if isinstance(item, (list, tuple)): for value_item in item: preview = extract(value_item) if preview: return preview return "" return extract(parsed) def _metadata_number_or_text(metadata: Dict[str, Any], *keys: str) -> Any: value = _metadata_text(metadata, *keys, max_chars=100) if not value: return None try: number = int(value) if str(number) == value: return number except (TypeError, ValueError): pass return value def extract_safetensors_header_metadata(file_path: str) -> Dict[str, Any]: """Extract displayable local model metadata from a safetensors header.""" header_json = read_safetensors_header(file_path) if not isinstance(header_json, dict): return {} metadata = header_json.get("__metadata__") if not isinstance(metadata, dict): metadata = {} result: Dict[str, Any] = {} metadata_keys = sorted(str(key) for key in metadata) if metadata_keys: result["header_metadata_keys"] = metadata_keys[:SAFETENSORS_METADATA_KEY_LIMIT] title = _metadata_text( metadata, "modelspec.title", "ssmd_display_name", "ss_output_name", "title", "name", "model_name", max_chars=300, ) if title: result["model_name"] = title description = _metadata_text( metadata, "modelspec.description", "ssmd_description", "description", "model_description", "ss_training_comment", ) if description: result["description"] = description result["model_description"] = description author = _metadata_text( metadata, "modelspec.author", "ssmd_author", "author", "creator", max_chars=200, ) if author: result["author"] = author result["creator"] = {"username": author, "name": author} license_value = _metadata_text(metadata, "modelspec.license", "license", max_chars=200) if license_value: result["license"] = license_value usage_hint = _metadata_text( metadata, "modelspec.usage_hint", "usage_hint", "usage_tips", max_chars=2000, ) if usage_hint: result["usage_hint"] = usage_hint result["usage_tips"] = usage_hint explicit_tags: List[str] = [] for key in ("modelspec.tags", "ssmd_tags", "tags", "model_tags"): explicit_tags.extend(_metadata_string_list(_metadata_get(metadata, key))) explicit_tags.extend(_collect_tag_frequency_tags(_metadata_get(metadata, "ss_tag_frequency"))) tags = _dedupe_strings(explicit_tags, limit=SAFETENSORS_METADATA_TAG_LIMIT) if tags: result["tags"] = tags trained_words: List[str] = [] for key in ( "modelspec.trigger_phrase", "modelspec.trigger_phrases", "trigger_phrase", "trigger_phrases", "trigger", "trained_words", "trainedWords", "ssmd_trained_words", ): trained_words.extend(_metadata_string_list(_metadata_get(metadata, key), limit=20)) trained_words = _dedupe_strings(trained_words, limit=20) if trained_words: result["trained_words"] = trained_words preview_url = _metadata_preview_url(metadata) if preview_url: result["preview_url"] = preview_url result["images"] = [ { "url": preview_url, "metadata": {"source": "safetensors_header"}, } ] clip_skip = _metadata_number_or_text(metadata, "ss_clip_skip", "clip_skip", "clipSkip") if clip_skip not in (None, ""): result["clip_skip"] = clip_skip sha256 = extract_safetensors_header_sha256(file_path) if sha256: result["sha256"] = sha256 result["hash"] = sha256 result["hashes"] = {"SHA256": sha256} result["sha256_source"] = "safetensors_header" result["hash_status"] = "completed" base_model = infer_safetensors_base_model(file_path) if base_model: result["base_model"] = base_model result["base_model_source"] = "safetensors_header" result["base_model_inferred"] = True raw_base = _metadata_text( metadata, "ss_base_model_version", "modelspec.architecture", "architecture", max_chars=300, ) if raw_base: result["base_model_raw"] = raw_base if _metadata_get(metadata, "ss_network_module"): result["model_type"] = "LORA" summary_fields = { "architecture": ("modelspec.architecture", "architecture"), "base_model_version": ("ss_base_model_version",), "resolution": ("modelspec.resolution", "ss_resolution"), "prediction_type": ("modelspec.prediction_type", "prediction_type"), "date": ("modelspec.date", "date"), "implementation": ("modelspec.implementation",), "network_module": ("ss_network_module",), "network_dim": ("ss_network_dim",), "network_alpha": ("ss_network_alpha",), "model_hash": ("sshs_model_hash",), "legacy_model_hash": ("sshs_legacy_hash",), "training_model": ("ss_sd_model_name", "ss_new_sd_model_name"), "vae": ("ss_vae_name",), "encoder_layer": ("modelspec.encoder_layer",), "merged_from": ("modelspec.merged_from",), "merge_recipe": ("sd_merge_recipe",), "merge_models": ("sd_merge_models",), } metadata_summary: Dict[str, Any] = {} for output_key, source_keys in summary_fields.items(): value = _metadata_get(metadata, *source_keys) if value in (None, ""): continue limited_value = _limited_metadata_value(value) if limited_value not in (None, "", [], {}): metadata_summary[output_key] = limited_value if _metadata_get(metadata, "workflow") not in (None, ""): metadata_summary["has_embedded_workflow"] = True if _metadata_get(metadata, "prompt") not in (None, ""): metadata_summary["has_embedded_prompt"] = True if metadata_summary: result["metadata_summary"] = metadata_summary meaningful_keys = ( "model_name", "description", "author", "license", "usage_hint", "tags", "trained_words", "preview_url", "sha256", "base_model", "metadata_summary", ) if any(result.get(key) for key in meaningful_keys): result["from_safetensors_header"] = True result["metadata_source"] = "safetensors_header" result["details_source"] = "safetensors_header" result["local_metadata_available"] = True return result return {} def calculate_file_sha256( file_path: str, chunk_size: int = 131072, on_progress: Optional[Callable[[int, int], None]] = None, is_cancelled: Optional[Callable[[], bool]] = None, *, use_safetensors_header: bool = True, on_hash_source: Optional[Callable[[str], None]] = None, ) -> Optional[str]: """Return a SHA256 for a local file, preferring safetensors header metadata.""" if not file_path or not os.path.exists(file_path): return None if use_safetensors_header: embedded_sha256 = extract_safetensors_header_sha256(file_path) if embedded_sha256: if on_hash_source: on_hash_source("safetensors_header") return embedded_sha256 sha256_hash = hashlib.sha256() try: total_bytes = os.path.getsize(file_path) if on_progress else 0 bytes_read = 0 with open(file_path, "rb") as f: for byte_block in iter(lambda: f.read(chunk_size), b""): if is_cancelled and is_cancelled(): raise HashCalculationCancelled("Hash calculation cancelled") if byte_block: sha256_hash.update(byte_block) bytes_read += len(byte_block) if on_progress: on_progress(bytes_read, total_bytes) if on_hash_source: on_hash_source("file") return sha256_hash.hexdigest() except HashCalculationCancelled: raise except Exception as e: _get_log().error(f"Error computing hash for {file_path}: {e}") return None def save_catalog_with_backup( data_path: str, data: Any, meta_path: str, meta: Any, indent: int = 2 ) -> None: """Backup an existing catalog file to .bak, then atomically write the new data and its metadata.""" import shutil # Ensure directory exists dir_name = os.path.dirname(data_path) if dir_name: os.makedirs(dir_name, exist_ok=True) # Backup existing data file if os.path.exists(data_path): shutil.copy2(data_path, f"{data_path}.bak") # Write data file atomically write_json_atomic(data_path, data, indent=indent) # Write metadata file atomically write_json_atomic(meta_path, meta, indent=indent) def read_json_safe(file_path: str, default: Any = None) -> Any: """Safely read and parse a JSON file, returning a default value on error.""" if not file_path or not os.path.exists(file_path): return default try: with open(file_path, encoding="utf-8") as f: return json.load(f) except Exception as e: _get_log().warning(f"Error reading JSON from {os.path.basename(file_path)}: {e}") return default def get_comfy_root_path(folder_paths_module: Optional[Any] = None) -> str: """Return the absolute path to the ComfyUI root directory.""" fp = folder_paths_module if fp is None: try: import folder_paths fp = folder_paths except ImportError: return "" try: module_file = getattr(fp, "__file__", "") if module_file: return os.path.dirname(os.path.abspath(module_file)) except Exception: pass return "" def get_directory_entry_set( directory: str, cache: Optional[Dict[str, Set[str]]] = None, ) -> Set[str]: """ Return the set of case-normalized filenames in a directory. If ``cache`` is provided, it will be checked and updated with the result keyed by normalized directory path. """ if not directory: return set() normalized_dir = os.path.normcase(os.path.abspath(directory)) if cache is not None and normalized_dir in cache: return cache[normalized_dir] try: entries = {os.path.normcase(name) for name in os.listdir(normalized_dir)} except OSError: entries = set() if cache is not None: cache[normalized_dir] = entries return entries def find_external_metadata_sidecar_path( model_path: str, dir_entries: Optional[Set[str]] = None, ) -> str: """ Find metadata owned by another tool (for example .metadata.json, .civitai.info, or .json). These files are read-only to Model Resolver. """ if not model_path: return "" directory = os.path.dirname(model_path) filename = get_filename_from_path(model_path) normalized_model_path = os.path.normcase(os.path.abspath(model_path)) # Base name without extension base_name = filename.rsplit(".", 1)[0] if "." in filename else filename has_stem_collision = has_model_sidecar_name_collision( model_path, dir_entries=dir_entries, ) # Name patterns based on civitai.py and resolver.py. exact_names = [ filename + ".metadata.json", filename + ".civitai.info", filename + ".json", ] legacy_names = [ base_name + ".metadata.json", base_name + ".civitai.info", base_name + ".json", filename.replace("_", " ").split()[0] + ".metadata.json" if "_" in base_name else None, ] possible_names = _ordered_sidecar_names( model_path, exact_names, legacy_names, has_stem_collision=has_stem_collision, dir_entries=dir_entries, ) for name in possible_names: if name: if dir_entries is not None and os.path.normcase(name) not in dir_entries: continue path = os.path.join(directory, name) if os.path.normcase(os.path.abspath(path)) == normalized_model_path: continue if dir_entries is None and not os.path.exists(path): continue if ( has_stem_collision and name in legacy_names and not _metadata_sidecar_matches_model(path, model_path) ): continue return path return "" def get_model_resolver_sidecar_path(model_path: str) -> str: """Return the sidecar path exclusively owned by Model Resolver.""" directory = os.path.dirname(model_path) filename = get_filename_from_path(model_path) return os.path.join(directory, f"{filename}{MODEL_RESOLVER_METADATA_SUFFIX}") def find_model_resolver_sidecar_path( model_path: str, dir_entries: Optional[Set[str]] = None, ) -> str: """Return the existing Model Resolver sidecar path, if present.""" if not model_path: return "" filename = get_filename_from_path(model_path) expected_name = f"{filename}{MODEL_RESOLVER_METADATA_SUFFIX}" if dir_entries is not None and os.path.normcase(expected_name) not in dir_entries: return "" metadata_path = get_model_resolver_sidecar_path(model_path) normalized_model_path = os.path.normcase(os.path.abspath(model_path)) if ( os.path.normcase(os.path.abspath(metadata_path)) != normalized_model_path and os.path.isfile(metadata_path) ): return metadata_path return "" def find_metadata_sidecar_path( model_path: str, dir_entries: Optional[Set[str]] = None, ) -> str: """ Return the preferred readable sidecar path. Model Resolver metadata takes precedence. External metadata remains available as a read-only fallback. """ return ( find_model_resolver_sidecar_path(model_path, dir_entries=dir_entries) or find_external_metadata_sidecar_path(model_path, dir_entries=dir_entries) ) def _metadata_list_values(value: Any) -> List[Any]: if isinstance(value, (list, tuple, set)): return list(value) if value in (None, ""): return [] return [value] def _merge_metadata_lists(external_value: Any, resolver_value: Any) -> List[Any]: result: List[Any] = [] seen = set() for item in _metadata_list_values(external_value) + _metadata_list_values(resolver_value): try: key = json.dumps(item, sort_keys=True, ensure_ascii=False) except (TypeError, ValueError): key = str(item) if key in seen: continue seen.add(key) result.append(item) return result def read_merged_model_metadata(model_path: str, default: Any = None) -> Any: """ Read external and Model Resolver sidecars without modifying either file. Resolver-owned technical/provider fields take precedence. User-editable external fields remain authoritative, while common list fields are merged. """ external_path = find_external_metadata_sidecar_path(model_path) resolver_path = find_model_resolver_sidecar_path(model_path) external = read_json_safe(external_path, {}) if external_path else {} resolver = read_json_safe(resolver_path, {}) if resolver_path else {} external = external if isinstance(external, dict) else {} resolver = resolver if isinstance(resolver, dict) else {} if not external and not resolver: return default if not external: return dict(resolver) if not resolver: return dict(external) merged = {**external, **resolver} for field in _EXTERNAL_METADATA_PREFERRED_FIELDS: if field in external: merged[field] = external[field] for field in _MERGED_METADATA_LIST_FIELDS: if field in external or field in resolver: merged[field] = _merge_metadata_lists( external.get(field), resolver.get(field), ) return merged def split_path_segments(path_value: Any, filter_dots: bool = True) -> list[str]: """Split path into standard segments, replacing backslashes and filtering empty segments.""" text = str(path_value or "").replace("\\", "/").strip() if not text: return [] parts = [] for raw_part in text.split("/"): part = raw_part.strip() if not part: continue if filter_dots and part in {".", ".."}: continue parts.append(part) return parts def has_model_sidecar_name_collision( model_path: str, dir_entries: Optional[Set[str]] = None, ) -> bool: """Return whether another model file shares this file's stem.""" if not model_path: return False directory = os.path.dirname(model_path) filename = get_filename_from_path(model_path) base_name, extension = os.path.splitext(filename) if not base_name or not extension: return False from .type_utils import MODEL_EXTENSIONS normalized_model_path = os.path.normcase(os.path.abspath(model_path)) normalized_filename = os.path.normcase(filename) for candidate_extension in MODEL_EXTENSIONS: candidate_filename = f"{base_name}{candidate_extension}" if os.path.normcase(candidate_filename) == normalized_filename: continue if dir_entries is not None: if os.path.normcase(candidate_filename) in dir_entries: return True else: candidate_path = os.path.join(directory, candidate_filename) if os.path.normcase(os.path.abspath(candidate_path)) == normalized_model_path: continue if os.path.isfile(candidate_path): return True return False def _metadata_sidecar_matches_model(metadata_path: str, model_path: str) -> bool: """Return whether an ambiguous legacy sidecar names the selected model file.""" metadata = read_json_safe(metadata_path, {}) if not isinstance(metadata, dict): return False candidate_values = [ metadata.get("filename"), metadata.get("file_path"), metadata.get("model_path"), metadata.get("path"), ] for key in ("file", "file_info", "selected_file", "selectedFile"): nested = metadata.get(key) if isinstance(nested, dict): candidate_values.extend( [ nested.get("name"), nested.get("filename"), nested.get("file_name"), nested.get("fileName"), nested.get("path"), ] ) model_filename = os.path.normcase(get_filename_from_path(model_path)) return any( os.path.normcase(get_filename_from_path(value)) == model_filename for value in candidate_values if value ) def _metadata_sidecar_paths(model_path: str) -> List[str]: directory = os.path.dirname(model_path) filename = get_filename_from_path(model_path) base_name = os.path.splitext(filename)[0] candidate_names = _ordered_sidecar_names( model_path, [f"{filename}.metadata.json"], [f"{base_name}.metadata.json"], ) candidates = [os.path.join(directory, name) for name in candidate_names] result: List[str] = [] seen = set() for candidate in candidates: try: key = os.path.normcase(os.path.abspath(candidate)) except (OSError, ValueError): key = os.path.normcase(candidate) if key in seen: continue seen.add(key) result.append(candidate) return result def _ordered_sidecar_names( model_path: str, exact_names: List[Optional[str]], legacy_names: List[Optional[str]], *, has_stem_collision: Optional[bool] = None, dir_entries: Optional[Set[str]] = None, ) -> List[str]: """Return sidecar names in the established exact/legacy preference order.""" directory = os.path.dirname(model_path) exact_names = [name for name in exact_names if name] legacy_names = [name for name in legacy_names if name] if has_stem_collision is None: has_stem_collision = has_model_sidecar_name_collision( model_path, dir_entries=dir_entries, ) exact_exists = False if exact_names: if dir_entries is not None: exact_exists = os.path.normcase(exact_names[0]) in dir_entries else: exact_path = os.path.join(directory, exact_names[0]) exact_exists = bool(exact_path) and os.path.isfile(exact_path) prefer_exact_names = has_stem_collision or exact_exists if prefer_exact_names: return exact_names + legacy_names candidates: List[str] = [] for index in range(max(len(exact_names), len(legacy_names))): if index < len(legacy_names): candidates.append(legacy_names[index]) if index < len(exact_names): candidates.append(exact_names[index]) return candidates def get_metadata_sidecar_path(model_path: str) -> str: """Return the legacy external .metadata.json path for read compatibility.""" return _metadata_sidecar_paths(model_path)[0] def _get_safe_sidecar_path( file_path: str, sidecar_factory: Callable[[str], str], ) -> str: """Return a validated sidecar path produced by ``sidecar_factory``.""" raw_model_path = str(file_path or "").strip() if not raw_model_path: raise ValueError("A model path is required") model_path = os.path.realpath(os.path.abspath(raw_model_path)) model_dir = os.path.realpath(os.path.dirname(model_path)) metadata_path = os.path.realpath(sidecar_factory(model_path)) if ( not model_dir or os.path.dirname(metadata_path) != model_dir or metadata_path == model_path or not is_path_within(metadata_path, model_dir) ): raise ValueError("Metadata path is outside the model directory") return metadata_path def get_safe_metadata_sidecar_path(file_path: str) -> str: """Return a validated legacy external sidecar path for compatibility.""" return _get_safe_sidecar_path(file_path, get_metadata_sidecar_path) def get_safe_model_resolver_sidecar_path(file_path: str) -> str: """Return a validated sidecar path owned exclusively by Model Resolver.""" return _get_safe_sidecar_path(file_path, get_model_resolver_sidecar_path)