mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-08-07 14:30:15 -03:00
Compare commits
9 Commits
v1.1.8
...
fe95fae5f2
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fe95fae5f2 | ||
|
|
ce8a95abf7 | ||
|
|
c8e7e543d6 | ||
|
|
a9dbb15ffa | ||
|
|
cf64043f7d | ||
|
|
ccaff92c18 | ||
|
|
585b5c922a | ||
|
|
ea80c2224c | ||
|
|
8b0f56c1a6 |
@@ -17,6 +17,7 @@ try: # pragma: no cover - import fallback for pytest collection
|
||||
from .py.nodes.lora_cycler import LoraCyclerLM
|
||||
from .py.nodes.lora_info import LoraInfoLM
|
||||
from .py.nodes.lora_syntax_to_path import LoraSyntaxToPath
|
||||
from .py.nodes.create_hook_lora import CreateHookLoraLM
|
||||
from .py.metadata_collector import init as init_metadata_collector
|
||||
except (
|
||||
ImportError
|
||||
@@ -62,6 +63,9 @@ except (
|
||||
LoraSyntaxToPath = importlib.import_module(
|
||||
"py.nodes.lora_syntax_to_path"
|
||||
).LoraSyntaxToPath
|
||||
CreateHookLoraLM = importlib.import_module(
|
||||
"py.nodes.create_hook_lora"
|
||||
).CreateHookLoraLM
|
||||
init_metadata_collector = importlib.import_module("py.metadata_collector").init
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
@@ -83,6 +87,7 @@ NODE_CLASS_MAPPINGS = {
|
||||
LoraCyclerLM.NAME: LoraCyclerLM,
|
||||
LoraInfoLM.NAME: LoraInfoLM,
|
||||
LoraSyntaxToPath.NAME: LoraSyntaxToPath,
|
||||
CreateHookLoraLM.NAME: CreateHookLoraLM,
|
||||
}
|
||||
|
||||
WEB_DIRECTORY = "./web/comfyui"
|
||||
|
||||
117
py/nodes/create_hook_lora.py
Normal file
117
py/nodes/create_hook_lora.py
Normal file
@@ -0,0 +1,117 @@
|
||||
"""Create Hook LoRA (LoraManager) — multi-LoRA hook node compatible with ComfyUI's built-in hook pipeline.
|
||||
|
||||
Produces ``("HOOKS",)`` output that chains seamlessly with downstream hook consumers
|
||||
(ConditioningSetProperties, SetHookKeyframes, CombineHooks, SetClipHooks, etc.).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
from ..utils.utils import get_lora_info_absolute
|
||||
from .utils import (
|
||||
FlexibleOptionalInputType,
|
||||
any_type,
|
||||
apply_lora_syntax_format,
|
||||
get_loras_list,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CreateHookLoraLM:
|
||||
NAME = "Create Hook LoRA (LoraManager)"
|
||||
CATEGORY = "Lora Manager/hooks"
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"text": (
|
||||
"AUTOCOMPLETE_TEXT_LORAS",
|
||||
{
|
||||
"placeholder": "Search LoRAs to add...",
|
||||
"tooltip": (
|
||||
"Search and select LoRAs. Each LoRA gets its own "
|
||||
"model/clip strength. Hooks chain with prev_hooks."
|
||||
),
|
||||
},
|
||||
),
|
||||
},
|
||||
"optional": FlexibleOptionalInputType(any_type),
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("HOOKS", "STRING", "STRING")
|
||||
RETURN_NAMES = ("HOOKS", "trigger_words", "active_loras")
|
||||
FUNCTION = "create_hook"
|
||||
|
||||
def create_hook(self, text: str, **kwargs):
|
||||
"""Create a HookGroup from the selected LoRAs, chained with prev_hooks.
|
||||
|
||||
Each active LoRA from the widget is loaded and wrapped in a WeightHook
|
||||
via :func:`comfy.hooks.create_hook_lora`. All hooks are combined into a
|
||||
single group and returned alongside trigger words and a human-readable
|
||||
summary of the active LoRAs.
|
||||
"""
|
||||
del text # used by the frontend widget only
|
||||
|
||||
# Lazy imports: comfy is not available in CI/test environment at module level
|
||||
import comfy.hooks # type: ignore # noqa: C0415
|
||||
import comfy.utils # type: ignore # noqa: C0415
|
||||
|
||||
prev_hooks: comfy.hooks.HookGroup | None = kwargs.get("prev_hooks")
|
||||
|
||||
hook_group = prev_hooks.clone() if prev_hooks is not None else comfy.hooks.HookGroup()
|
||||
|
||||
all_trigger_words: list[str] = []
|
||||
active_loras: list[tuple[str, float, float]] = []
|
||||
|
||||
for lora in get_loras_list(kwargs):
|
||||
if not lora.get("active", False):
|
||||
continue
|
||||
|
||||
lora_name = apply_lora_syntax_format(lora["name"])
|
||||
model_strength = float(lora["strength"])
|
||||
clip_strength = float(lora.get("clipStrength", model_strength))
|
||||
|
||||
# Skip useless no-op entries (both strengths are zero)
|
||||
if model_strength == 0.0 and clip_strength == 0.0:
|
||||
continue
|
||||
|
||||
lora_path, trigger_words = get_lora_info_absolute(lora_name)
|
||||
if not lora_path or not os.path.isfile(lora_path):
|
||||
logger.warning("LoRA '%s' not found — skipping", lora_name)
|
||||
continue
|
||||
|
||||
try:
|
||||
lora_weights = comfy.utils.load_torch_file(lora_path, safe_load=True)
|
||||
|
||||
lora_hooks = comfy.hooks.create_hook_lora(
|
||||
lora=lora_weights,
|
||||
strength_model=model_strength,
|
||||
strength_clip=clip_strength,
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("Failed to load LoRA '%s' — skipping", lora_name)
|
||||
continue
|
||||
hook_group = hook_group.clone_and_combine(lora_hooks)
|
||||
|
||||
active_loras.append((lora_name, model_strength, clip_strength))
|
||||
all_trigger_words.extend(trigger_words)
|
||||
|
||||
# Format trigger words (group mode separator)
|
||||
trigger_words_text = ",, ".join(all_trigger_words) if all_trigger_words else ""
|
||||
|
||||
# Format active LoRAs summary
|
||||
formatted_loras = []
|
||||
for name, model_s, clip_s in active_loras:
|
||||
if abs(model_s - clip_s) > 0.001:
|
||||
formatted_loras.append(
|
||||
f"<lora:{name}:{model_s}:{clip_s}>"
|
||||
)
|
||||
else:
|
||||
formatted_loras.append(f"<lora:{name}:{model_s}>")
|
||||
active_loras_text = " ".join(formatted_loras)
|
||||
|
||||
return (hook_group, trigger_words_text, active_loras_text)
|
||||
@@ -1389,7 +1389,17 @@ class DownloadManager:
|
||||
|
||||
# Update save directory with relative path if provided
|
||||
if relative_path:
|
||||
base_save_dir = save_dir
|
||||
save_dir = os.path.join(save_dir, relative_path)
|
||||
# Security: validate path containment after joining
|
||||
resolved_dir = os.path.realpath(os.path.normpath(save_dir))
|
||||
base_dir = os.path.realpath(os.path.normpath(base_save_dir))
|
||||
if not resolved_dir.startswith(base_dir + os.sep) and resolved_dir != base_dir:
|
||||
logger.warning(
|
||||
"Path traversal detected: %s escapes %s",
|
||||
resolved_dir, base_dir,
|
||||
)
|
||||
return {"success": False, "error": "Download path is outside allowed directory"}
|
||||
# Create directory if it doesn't exist
|
||||
os.makedirs(save_dir, exist_ok=True)
|
||||
|
||||
@@ -1827,6 +1837,9 @@ class DownloadManager:
|
||||
model_tags, model_type
|
||||
)
|
||||
|
||||
if not first_tag:
|
||||
first_tag = "no tags" # Default if no tags available
|
||||
|
||||
# Format the template with available data
|
||||
formatted_path = path_template
|
||||
formatted_path = formatted_path.replace("{base_model}", mapped_base_model)
|
||||
@@ -1842,6 +1855,15 @@ class DownloadManager:
|
||||
if model_type == "embedding":
|
||||
formatted_path = formatted_path.replace(" ", "_")
|
||||
|
||||
# Sanitize the resolved path to prevent path traversal:
|
||||
# - Strip leading slashes (prevents os.path.join from treating path as absolute)
|
||||
# - Collapse double slashes from empty placeholder substitutions
|
||||
# - Strip trailing slashes for cleanliness
|
||||
formatted_path = formatted_path.lstrip("/")
|
||||
while "//" in formatted_path:
|
||||
formatted_path = formatted_path.replace("//", "/")
|
||||
formatted_path = formatted_path.rstrip("/")
|
||||
|
||||
return formatted_path
|
||||
|
||||
async def _execute_download(
|
||||
|
||||
@@ -8,6 +8,7 @@ from abc import ABC, abstractmethod
|
||||
from ..utils.utils import calculate_relative_path_for_model, remove_empty_dirs
|
||||
from ..utils.constants import AUTO_ORGANIZE_BATCH_SIZE
|
||||
from ..services.settings_manager import get_settings_manager
|
||||
from ..services.model_lifecycle_service import _require_path_in_library_roots
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -493,6 +494,9 @@ class ModelMoveService:
|
||||
Dictionary with move result
|
||||
"""
|
||||
try:
|
||||
_require_path_in_library_roots(file_path, self.scanner, label="Source path")
|
||||
_require_path_in_library_roots(target_path, self.scanner, label="Target path")
|
||||
|
||||
if use_default_paths:
|
||||
# Find the model in cache to get metadata
|
||||
cache = await self.scanner.get_cached_data()
|
||||
|
||||
@@ -48,6 +48,35 @@ async def delete_model_artifacts(
|
||||
return deleted
|
||||
|
||||
|
||||
def _require_path_in_library_roots(file_path: str, scanner, *, label: str = "path") -> None:
|
||||
"""Raise ``ValueError`` if *file_path* is not inside a configured model root.
|
||||
|
||||
Uses ``os.path.realpath()`` to resolve symlinks before comparing,
|
||||
so symlink-based escapes are also caught. Skips when the scanner
|
||||
does not expose ``get_model_roots`` or the list is empty.
|
||||
"""
|
||||
|
||||
roots = None
|
||||
if hasattr(scanner, "get_model_roots"):
|
||||
try:
|
||||
roots = scanner.get_model_roots()
|
||||
except NotImplementedError:
|
||||
roots = None
|
||||
if not roots:
|
||||
return
|
||||
|
||||
resolved = os.path.realpath(os.path.normpath(file_path))
|
||||
|
||||
for root in roots:
|
||||
root_resolved = os.path.realpath(os.path.normpath(root))
|
||||
if resolved == root_resolved or resolved.startswith(root_resolved + os.sep):
|
||||
return
|
||||
|
||||
raise ValueError(
|
||||
f"{label} '{file_path}' is outside configured library directories"
|
||||
)
|
||||
|
||||
|
||||
class ModelLifecycleService:
|
||||
"""Co-ordinate destructive and mutating model operations."""
|
||||
|
||||
@@ -74,6 +103,8 @@ class ModelLifecycleService:
|
||||
if not file_path:
|
||||
raise ValueError("Model path is required")
|
||||
|
||||
_require_path_in_library_roots(file_path, self._scanner, label="File path")
|
||||
|
||||
cache = await self._scanner.get_cached_data()
|
||||
|
||||
cached_entry = None
|
||||
@@ -182,6 +213,8 @@ class ModelLifecycleService:
|
||||
if not file_path:
|
||||
raise ValueError("Model path is required")
|
||||
|
||||
_require_path_in_library_roots(file_path, self._scanner, label="File path")
|
||||
|
||||
metadata_path = os.path.splitext(file_path)[0] + ".metadata.json"
|
||||
metadata = await self._metadata_loader(metadata_path)
|
||||
metadata["exclude"] = True
|
||||
@@ -229,6 +262,8 @@ class ModelLifecycleService:
|
||||
if not file_path:
|
||||
raise ValueError("Model path is required")
|
||||
|
||||
_require_path_in_library_roots(file_path, self._scanner, label="File path")
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
raise ValueError("Model file does not exist")
|
||||
|
||||
@@ -270,6 +305,9 @@ class ModelLifecycleService:
|
||||
if not file_paths:
|
||||
raise ValueError("No file paths provided for deletion")
|
||||
|
||||
for path in file_paths:
|
||||
_require_path_in_library_roots(path, self._scanner, label="File path")
|
||||
|
||||
return await self._scanner.bulk_delete_models(file_paths)
|
||||
|
||||
async def rename_model(
|
||||
@@ -280,6 +318,8 @@ class ModelLifecycleService:
|
||||
if not file_path or not new_file_name:
|
||||
raise ValueError("File path and new file name are required")
|
||||
|
||||
_require_path_in_library_roots(file_path, self._scanner, label="File path")
|
||||
|
||||
invalid_chars = {"/", "\\", ":", "*", "?", '"', "<", ">", "|"}
|
||||
if any(char in new_file_name for char in invalid_chars):
|
||||
raise ValueError("Invalid characters in file name")
|
||||
|
||||
@@ -14,7 +14,7 @@ from ..utils.metadata_manager import MetadataManager
|
||||
from ..utils.civitai_utils import resolve_license_info
|
||||
from .model_cache import ModelCache
|
||||
from .model_hash_index import ModelHashIndex
|
||||
from .model_lifecycle_service import delete_model_artifacts
|
||||
from .model_lifecycle_service import delete_model_artifacts, _require_path_in_library_roots
|
||||
from .service_registry import ServiceRegistry
|
||||
from .websocket_manager import ws_manager
|
||||
from .persistent_model_cache import get_persistent_cache
|
||||
@@ -1394,6 +1394,9 @@ class ModelScanner:
|
||||
|
||||
base_name = os.path.splitext(os.path.basename(source_path))[0]
|
||||
source_dir = os.path.dirname(source_path)
|
||||
|
||||
_require_path_in_library_roots(source_path, self, label="Source path")
|
||||
_require_path_in_library_roots(target_path, self, label="Target path")
|
||||
|
||||
os.makedirs(target_path, exist_ok=True)
|
||||
|
||||
@@ -1971,6 +1974,8 @@ class ModelScanner:
|
||||
break
|
||||
|
||||
try:
|
||||
_require_path_in_library_roots(file_path, self, label="File path")
|
||||
|
||||
target_dir = os.path.dirname(file_path)
|
||||
base_name = os.path.basename(file_path)
|
||||
file_name, main_extension = os.path.splitext(base_name)
|
||||
|
||||
@@ -12,6 +12,7 @@ NODE_TYPES = {
|
||||
"Lora Loader (LoraManager)": 1,
|
||||
"Lora Stacker (LoraManager)": 2,
|
||||
"WanVideo Lora Select (LoraManager)": 3,
|
||||
"Create Hook LoRA (LoraManager)": 4,
|
||||
}
|
||||
|
||||
# Default ComfyUI node color when bgcolor is null
|
||||
|
||||
@@ -488,6 +488,12 @@ def calculate_relative_path_for_model(
|
||||
if model_type == "embedding":
|
||||
formatted_path = formatted_path.replace(" ", "_")
|
||||
|
||||
# Sanitize the resolved path to prevent path traversal
|
||||
formatted_path = formatted_path.lstrip("/")
|
||||
while "//" in formatted_path:
|
||||
formatted_path = formatted_path.replace("//", "/")
|
||||
formatted_path = formatted_path.rstrip("/")
|
||||
|
||||
return formatted_path
|
||||
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
[project]
|
||||
name = "comfyui-lora-manager"
|
||||
description = "Revolutionize your workflow with the ultimate LoRA companion for ComfyUI!"
|
||||
version = "1.1.8"
|
||||
version = "1.1.9"
|
||||
license = {file = "LICENSE"}
|
||||
dependencies = [
|
||||
"aiohttp",
|
||||
|
||||
@@ -49,10 +49,6 @@ export const MODEL_CONFIG = {
|
||||
* @returns {Object} Object containing all API endpoints for the model type
|
||||
*/
|
||||
export function getApiEndpoints(modelType) {
|
||||
if (!Object.values(MODEL_TYPES).includes(modelType)) {
|
||||
throw new Error(`Invalid model type: ${modelType}`);
|
||||
}
|
||||
|
||||
return {
|
||||
// Base CRUD operations
|
||||
list: `/api/lm/${modelType}/list`,
|
||||
|
||||
@@ -369,21 +369,24 @@ export function getMatureBlurThreshold(settings = {}) {
|
||||
export const NODE_TYPES = {
|
||||
LORA_LOADER: 1,
|
||||
LORA_STACKER: 2,
|
||||
WAN_VIDEO_LORA_SELECT: 3
|
||||
WAN_VIDEO_LORA_SELECT: 3,
|
||||
HOOK_LORA: 4
|
||||
};
|
||||
|
||||
// Node type names to IDs mapping
|
||||
export const NODE_TYPE_NAMES = {
|
||||
"Lora Loader (LoraManager)": NODE_TYPES.LORA_LOADER,
|
||||
"Lora Stacker (LoraManager)": NODE_TYPES.LORA_STACKER,
|
||||
"WanVideo Lora Select (LoraManager)": NODE_TYPES.WAN_VIDEO_LORA_SELECT
|
||||
"WanVideo Lora Select (LoraManager)": NODE_TYPES.WAN_VIDEO_LORA_SELECT,
|
||||
"Create Hook LoRA (LoraManager)": NODE_TYPES.HOOK_LORA
|
||||
};
|
||||
|
||||
// Node type icons
|
||||
export const NODE_TYPE_ICONS = {
|
||||
[NODE_TYPES.LORA_LOADER]: "fas fa-l",
|
||||
[NODE_TYPES.LORA_STACKER]: "fas fa-s",
|
||||
[NODE_TYPES.WAN_VIDEO_LORA_SELECT]: "fas fa-w"
|
||||
[NODE_TYPES.WAN_VIDEO_LORA_SELECT]: "fas fa-w",
|
||||
[NODE_TYPES.HOOK_LORA]: "fas fa-h"
|
||||
};
|
||||
|
||||
// Default ComfyUI node color when bgcolor is null
|
||||
|
||||
@@ -85,6 +85,7 @@ sys.modules['comfy.utils'] = comfy_mock.utils
|
||||
sys.modules['comfy.sd'] = comfy_mock.sd
|
||||
sys.modules['comfy.model_management'] = comfy_mock.model_management
|
||||
sys.modules['comfy.comfy_types'] = comfy_mock.comfy_types
|
||||
sys.modules['comfy.hooks'] = MockModule("comfy.hooks")
|
||||
|
||||
execution_mock = MockModule("execution")
|
||||
execution_mock.PromptExecutor = mock.MagicMock()
|
||||
|
||||
@@ -1189,6 +1189,65 @@ def test_relative_path_sanitizes_model_and_version_placeholders():
|
||||
assert relative_path == "Fancy_Model/Version_One"
|
||||
|
||||
|
||||
def test_relative_path_empty_first_tag_fallback():
|
||||
"""Test that empty first_tag falls back to 'no tags'."""
|
||||
manager = DownloadManager()
|
||||
settings_manager = get_settings_manager()
|
||||
settings_manager.settings["download_path_templates"]["lora"] = (
|
||||
"{base_model}/{first_tag}"
|
||||
)
|
||||
|
||||
version_info = {
|
||||
"baseModel": "SDXL",
|
||||
"model": {"name": "Test Model", "tags": []},
|
||||
"creator": {"username": "Author"},
|
||||
}
|
||||
|
||||
relative_path = manager._calculate_relative_path(version_info, "lora")
|
||||
|
||||
assert relative_path == "SDXL/no tags"
|
||||
|
||||
|
||||
def test_relative_path_empty_base_model_and_first_tag():
|
||||
"""Test that empty base_model + empty first_tag does NOT produce a leading slash."""
|
||||
manager = DownloadManager()
|
||||
settings_manager = get_settings_manager()
|
||||
settings_manager.settings["download_path_templates"]["lora"] = (
|
||||
"{base_model}/{first_tag}"
|
||||
)
|
||||
|
||||
version_info = {
|
||||
"baseModel": "",
|
||||
"model": {"name": "Test Model", "tags": []},
|
||||
"creator": {"username": "Author"},
|
||||
}
|
||||
|
||||
relative_path = manager._calculate_relative_path(version_info, "lora")
|
||||
|
||||
assert not relative_path.startswith("/")
|
||||
assert relative_path == "no tags"
|
||||
|
||||
|
||||
def test_relative_path_sanitizes_double_slashes():
|
||||
"""Test that empty placeholder substitutions don't produce double slashes."""
|
||||
manager = DownloadManager()
|
||||
settings_manager = get_settings_manager()
|
||||
settings_manager.settings["download_path_templates"]["lora"] = (
|
||||
"{base_model}/{first_tag}/{author}"
|
||||
)
|
||||
|
||||
version_info = {
|
||||
"baseModel": "SDXL",
|
||||
"model": {"name": "Test Model", "tags": []},
|
||||
"creator": {"username": "Author"},
|
||||
}
|
||||
|
||||
relative_path = manager._calculate_relative_path(version_info, "lora")
|
||||
|
||||
assert "//" not in relative_path
|
||||
assert relative_path == "SDXL/no tags/Author"
|
||||
|
||||
|
||||
def test_distribute_preview_to_entries_moves_and_copies(tmp_path):
|
||||
"""Test that preview distribution moves file to first entry and copies to others."""
|
||||
manager = DownloadManager()
|
||||
|
||||
@@ -3,11 +3,164 @@ from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from py.services.model_lifecycle_service import ModelLifecycleService
|
||||
from py.services.model_lifecycle_service import ModelLifecycleService, _require_path_in_library_roots
|
||||
from py.utils.metadata_manager import MetadataManager
|
||||
from py.utils.models import LoraMetadata
|
||||
|
||||
|
||||
class ScannerWithRoots:
|
||||
def __init__(self, roots):
|
||||
self._roots = list(roots)
|
||||
|
||||
def get_model_roots(self):
|
||||
return self._roots
|
||||
|
||||
|
||||
class TestRequirePathInLibraryRoots:
|
||||
def test_accepts_path_within_root(self, tmp_path):
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
model = root / "model.safetensors"
|
||||
model.write_text("")
|
||||
|
||||
scanner = ScannerWithRoots([str(root)])
|
||||
_require_path_in_library_roots(str(model), scanner)
|
||||
|
||||
def test_rejects_path_outside_roots(self, tmp_path):
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
outside = tmp_path / "outside" / "model.safetensors"
|
||||
outside.parent.mkdir(parents=True)
|
||||
outside.write_text("")
|
||||
|
||||
scanner = ScannerWithRoots([str(root)])
|
||||
with pytest.raises(ValueError, match="outside configured library"):
|
||||
_require_path_in_library_roots(str(outside), scanner)
|
||||
|
||||
def test_passes_when_no_roots_configured(self, tmp_path):
|
||||
f = tmp_path / "model.safetensors"
|
||||
f.write_text("")
|
||||
|
||||
scanner = ScannerWithRoots([])
|
||||
_require_path_in_library_roots(str(f), scanner)
|
||||
|
||||
def test_accepts_path_matching_root_exactly(self, tmp_path):
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
|
||||
scanner = ScannerWithRoots([str(root)])
|
||||
_require_path_in_library_roots(str(root), scanner)
|
||||
|
||||
def test_rejects_symlink_escape(self, tmp_path):
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
model = root / "model.safetensors"
|
||||
model.write_text("")
|
||||
|
||||
outside_dir = tmp_path / "outside"
|
||||
outside_dir.mkdir()
|
||||
outside_file = outside_dir / "escaped.safetensors"
|
||||
outside_file.write_text("")
|
||||
|
||||
symlink = root / "link.safetensors"
|
||||
symlink.symlink_to(outside_file)
|
||||
|
||||
scanner = ScannerWithRoots([str(root)])
|
||||
with pytest.raises(ValueError, match="outside configured library"):
|
||||
_require_path_in_library_roots(str(symlink), scanner)
|
||||
|
||||
|
||||
class ScannerForDelete:
|
||||
def __init__(self, raw_data, roots, model_type="lora"):
|
||||
self.model_type = model_type
|
||||
self.cache = DummyCache(raw_data)
|
||||
self._hash_index = DummyHashIndex()
|
||||
self._roots = list(roots)
|
||||
self._persist_calls = []
|
||||
|
||||
def get_model_roots(self):
|
||||
return self._roots
|
||||
|
||||
async def get_cached_data(self):
|
||||
return self.cache
|
||||
|
||||
async def _persist_current_cache(self):
|
||||
self._persist_calls.append(True)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_model_rejects_path_outside_roots(tmp_path: Path):
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
model = root / "model.safetensors"
|
||||
model.write_bytes(b"data")
|
||||
|
||||
scanner = ScannerForDelete(
|
||||
raw_data=[{"file_path": str(model)}],
|
||||
roots=[str(root)],
|
||||
)
|
||||
service = ModelLifecycleService(
|
||||
scanner=scanner,
|
||||
metadata_manager=DummyMetadataManager({"civitai": {"modelId": 1}}),
|
||||
metadata_loader=lambda x: {},
|
||||
)
|
||||
# Path within root should work (model file exists)
|
||||
result = await service.delete_model(str(model))
|
||||
assert result["success"] is True
|
||||
|
||||
# Path outside root should be rejected
|
||||
outside = tmp_path / "outside.safetensors"
|
||||
outside.write_bytes(b"data")
|
||||
scanner2 = ScannerForDelete(
|
||||
raw_data=[],
|
||||
roots=[str(root)],
|
||||
)
|
||||
service2 = ModelLifecycleService(
|
||||
scanner=scanner2,
|
||||
metadata_manager=DummyMetadataManager({}),
|
||||
metadata_loader=lambda x: {},
|
||||
)
|
||||
with pytest.raises(ValueError, match="outside configured library"):
|
||||
await service2.delete_model(str(outside))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rename_model_rejects_path_outside_roots(tmp_path: Path):
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
|
||||
scanner = ScannerWithRoots([str(root)])
|
||||
service = ModelLifecycleService(
|
||||
scanner=scanner,
|
||||
metadata_manager=DummyMetadataManager({}),
|
||||
metadata_loader=lambda x: {},
|
||||
)
|
||||
outside = tmp_path / "outside.safetensors"
|
||||
outside.write_bytes(b"data")
|
||||
|
||||
with pytest.raises(ValueError, match="outside configured library"):
|
||||
await service.rename_model(file_path=str(outside), new_file_name="new_name")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_delete_rejects_any_path_outside_roots(tmp_path: Path):
|
||||
root = tmp_path / "loras"
|
||||
root.mkdir()
|
||||
model_ok = root / "model.safetensors"
|
||||
model_ok.write_bytes(b"data")
|
||||
outside = tmp_path / "outside.safetensors"
|
||||
outside.write_bytes(b"data")
|
||||
|
||||
scanner = ScannerWithRoots([str(root)])
|
||||
service = ModelLifecycleService(
|
||||
scanner=scanner,
|
||||
metadata_manager=DummyMetadataManager({}),
|
||||
metadata_loader=lambda x: {},
|
||||
)
|
||||
with pytest.raises(ValueError, match="outside configured library"):
|
||||
await service.bulk_delete_models([str(model_ok), str(outside)])
|
||||
|
||||
|
||||
class DummyCache:
|
||||
def __init__(self, raw_data):
|
||||
self.raw_data = raw_data
|
||||
|
||||
@@ -114,6 +114,38 @@ def test_calculate_relative_path_sanitizes_model_and_version_names(isolated_sett
|
||||
assert relative_path == "Fancy_Model/Version_One"
|
||||
|
||||
|
||||
def test_calculate_relative_path_sanitizes_leading_slash(isolated_settings):
|
||||
"""Test that empty base_model does NOT produce a leading slash in the path."""
|
||||
isolated_settings["download_path_templates"]["lora"] = "{base_model}/{first_tag}"
|
||||
|
||||
model_data = {
|
||||
"base_model": "",
|
||||
"tags": [],
|
||||
"civitai": {"id": 1, "creator": {"username": "Author"}},
|
||||
}
|
||||
|
||||
relative_path = calculate_relative_path_for_model(model_data, "lora")
|
||||
|
||||
assert not relative_path.startswith("/")
|
||||
assert relative_path == "no tags"
|
||||
|
||||
|
||||
def test_calculate_relative_path_sanitizes_double_slashes(isolated_settings):
|
||||
"""Test that empty substitutions don't produce double slashes."""
|
||||
isolated_settings["download_path_templates"]["lora"] = "{base_model}/{first_tag}/{author}"
|
||||
|
||||
model_data = {
|
||||
"base_model": "",
|
||||
"tags": [],
|
||||
"civitai": {"id": 1, "creator": {"username": "Author"}},
|
||||
}
|
||||
|
||||
relative_path = calculate_relative_path_for_model(model_data, "lora")
|
||||
|
||||
assert "//" not in relative_path
|
||||
assert relative_path == "no tags/Author"
|
||||
|
||||
|
||||
def test_calculate_recipe_fingerprint_filters_and_sorts():
|
||||
loras = [
|
||||
{"hash": "ABC", "strength": 0.1234},
|
||||
|
||||
@@ -16,6 +16,7 @@ export const LORA_PROVIDER_NODE_TYPES = [
|
||||
"Lora Stacker (LoraManager)",
|
||||
"Lora Randomizer (LoraManager)",
|
||||
"Lora Cycler (LoraManager)",
|
||||
"Create Hook LoRA (LoraManager)",
|
||||
] as const;
|
||||
|
||||
/**
|
||||
|
||||
143
web/comfyui/create_hook_lora.js
Normal file
143
web/comfyui/create_hook_lora.js
Normal file
@@ -0,0 +1,143 @@
|
||||
import { app } from "../../scripts/app.js";
|
||||
import {
|
||||
getActiveLorasFromNode,
|
||||
updateConnectedTriggerWords,
|
||||
chainCallback,
|
||||
mergeLoras,
|
||||
getWidgetByName,
|
||||
getWidgetSerializedValue,
|
||||
} from "./utils.js";
|
||||
import { addLorasWidget } from "./loras_widget.js";
|
||||
import { applyLoraValuesToText, debounce } from "./lora_syntax_utils.js";
|
||||
import { applySelectionHighlight } from "./trigger_word_highlight.js";
|
||||
import { updateConnectedLoraInfoNodes } from "./lora_info.js";
|
||||
|
||||
app.registerExtension({
|
||||
name: "LoraManager.CreateHookLora",
|
||||
|
||||
async beforeRegisterNodeDef(nodeType, nodeData, app) {
|
||||
if (nodeType.comfyClass === "Create Hook LoRA (LoraManager)") {
|
||||
chainCallback(nodeType.prototype, "onNodeCreated", function () {
|
||||
// Enable widget serialization so loras widget state is persisted
|
||||
this.serialize_widgets = true;
|
||||
|
||||
this.addInput("prev_hooks", "HOOKS", {
|
||||
shape: 7,
|
||||
});
|
||||
|
||||
// Flags to prevent callback loops between text widget ↔ loras widget
|
||||
let isUpdating = false;
|
||||
let isSyncingInput = false;
|
||||
|
||||
// Get the text input widget (AUTOCOMPLETE_TEXT_LORAS type, created by Vue widgets)
|
||||
const inputWidget = getWidgetByName(this, "text");
|
||||
if (!inputWidget) {
|
||||
console.warn(
|
||||
"LoRA Manager: text widget not found for Create Hook LoRA"
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.inputWidget = inputWidget;
|
||||
|
||||
const scheduleInputSync = debounce((lorasValue) => {
|
||||
if (isSyncingInput) {
|
||||
return;
|
||||
}
|
||||
|
||||
isSyncingInput = true;
|
||||
isUpdating = true;
|
||||
|
||||
try {
|
||||
const nextText = applyLoraValuesToText(
|
||||
inputWidget.value,
|
||||
lorasValue
|
||||
);
|
||||
|
||||
if (inputWidget.value !== nextText) {
|
||||
inputWidget.value = nextText;
|
||||
}
|
||||
} finally {
|
||||
isUpdating = false;
|
||||
isSyncingInput = false;
|
||||
}
|
||||
});
|
||||
|
||||
// Create the LoRA list widget
|
||||
const result = addLorasWidget(
|
||||
this,
|
||||
"loras",
|
||||
{
|
||||
onSelectionChange: (selection) => {
|
||||
applySelectionHighlight(this, selection);
|
||||
updateConnectedLoraInfoNodes(this, selection);
|
||||
},
|
||||
},
|
||||
(value) => {
|
||||
// Prevent recursive calls
|
||||
if (isUpdating) return;
|
||||
isUpdating = true;
|
||||
|
||||
try {
|
||||
// Update connected trigger word toggles with active LoRA names
|
||||
const activeLoraNames = new Set();
|
||||
value.forEach((lora) => {
|
||||
if (lora.active) {
|
||||
activeLoraNames.add(lora.name);
|
||||
}
|
||||
});
|
||||
updateConnectedTriggerWords(this, activeLoraNames);
|
||||
} finally {
|
||||
isUpdating = false;
|
||||
}
|
||||
|
||||
scheduleInputSync(value);
|
||||
}
|
||||
);
|
||||
|
||||
this.lorasWidget = result.widget;
|
||||
|
||||
// Set up callback for the text input widget to trigger merge logic
|
||||
inputWidget.callback = (value) => {
|
||||
if (isUpdating) return;
|
||||
isUpdating = true;
|
||||
|
||||
try {
|
||||
const currentLoras = this.lorasWidget?.value || [];
|
||||
const mergedLoras = mergeLoras(value, currentLoras);
|
||||
if (this.lorasWidget) {
|
||||
this.lorasWidget.value = mergedLoras;
|
||||
}
|
||||
|
||||
// Update connected trigger word toggles
|
||||
const activeLoraNames = getActiveLorasFromNode(this);
|
||||
updateConnectedTriggerWords(this, activeLoraNames);
|
||||
} finally {
|
||||
isUpdating = false;
|
||||
}
|
||||
};
|
||||
});
|
||||
}
|
||||
},
|
||||
|
||||
async loadedGraphNode(node) {
|
||||
if (node.comfyClass === "Create Hook LoRA (LoraManager)") {
|
||||
// Restore saved loras widget values on workflow load
|
||||
let existingLoras = [];
|
||||
if (node.widgets_values && node.widgets_values.length > 0) {
|
||||
const savedValue = getWidgetSerializedValue(node, "loras");
|
||||
existingLoras = savedValue || [];
|
||||
}
|
||||
// Merge the loras data from text widget with saved values
|
||||
const inputWidget =
|
||||
node.inputWidget || getWidgetByName(node, "text");
|
||||
if (!inputWidget) {
|
||||
console.warn(
|
||||
"LoRA Manager: text widget not found while restoring Create Hook LoRA"
|
||||
);
|
||||
return;
|
||||
}
|
||||
const mergedLoras = mergeLoras(inputWidget.value, existingLoras);
|
||||
node.lorasWidget.value = mergedLoras;
|
||||
}
|
||||
},
|
||||
});
|
||||
@@ -37,22 +37,28 @@ app.registerExtension({
|
||||
|
||||
// Handle broadcast mode (for Desktop/non-browser support)
|
||||
if (numericNodeId === -1) {
|
||||
// Find all Lora Loader nodes in the current graph
|
||||
const loraLoaderNodes = getAllGraphNodes(app.graph)
|
||||
// Find all compatible nodes in the current graph
|
||||
const compatibleClasses = new Set([
|
||||
"Lora Loader (LoraManager)",
|
||||
"Lora Stacker (LoraManager)",
|
||||
"WanVideo Lora Select (LoraManager)",
|
||||
"Create Hook LoRA (LoraManager)",
|
||||
]);
|
||||
const targetNodes = getAllGraphNodes(app.graph)
|
||||
.map(({ node }) => node)
|
||||
.filter((node) => node?.comfyClass === "Lora Loader (LoraManager)");
|
||||
.filter((node) => compatibleClasses.has(node?.comfyClass));
|
||||
|
||||
// Update each Lora Loader node found
|
||||
if (loraLoaderNodes.length > 0) {
|
||||
loraLoaderNodes.forEach((node) => {
|
||||
// Update each node found
|
||||
if (targetNodes.length > 0) {
|
||||
targetNodes.forEach((node) => {
|
||||
this.updateNodeLoraCode(node, loraCode, mode);
|
||||
});
|
||||
console.log(
|
||||
`Updated ${loraLoaderNodes.length} Lora Loader nodes in broadcast mode`
|
||||
`Updated ${targetNodes.length} nodes in broadcast mode`
|
||||
);
|
||||
} else {
|
||||
console.warn(
|
||||
"No Lora Loader nodes found in the workflow for broadcast update"
|
||||
"No compatible LoRA nodes found in the workflow for broadcast update"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -65,10 +71,11 @@ app.registerExtension({
|
||||
!node ||
|
||||
(node.comfyClass !== "Lora Loader (LoraManager)" &&
|
||||
node.comfyClass !== "Lora Stacker (LoraManager)" &&
|
||||
node.comfyClass !== "WanVideo Lora Select (LoraManager)")
|
||||
node.comfyClass !== "WanVideo Lora Select (LoraManager)" &&
|
||||
node.comfyClass !== "Create Hook LoRA (LoraManager)")
|
||||
) {
|
||||
console.warn(
|
||||
"Node not found or not a LoraLoader:",
|
||||
"Node not found or not a compatible LoRA node:",
|
||||
graphId ?? "root",
|
||||
nodeId
|
||||
);
|
||||
|
||||
@@ -12,6 +12,7 @@ const LORA_NODE_CLASSES = new Set([
|
||||
"Lora Loader (LoraManager)",
|
||||
"Lora Stacker (LoraManager)",
|
||||
"WanVideo Lora Select (LoraManager)",
|
||||
"Create Hook LoRA (LoraManager)",
|
||||
]);
|
||||
|
||||
function normalizeTriggerWordList(triggerWords) {
|
||||
|
||||
@@ -8,6 +8,7 @@ export const LORA_PROVIDER_NODE_TYPES = [
|
||||
"Lora Stacker (LoraManager)",
|
||||
"Lora Randomizer (LoraManager)",
|
||||
"Lora Cycler (LoraManager)",
|
||||
"Create Hook LoRA (LoraManager)",
|
||||
];
|
||||
|
||||
export const LORA_STACK_AGGREGATOR_NODE_TYPES = [
|
||||
|
||||
@@ -15656,7 +15656,8 @@ function createVueWidgetCleanup(vueApp, onCleanup) {
|
||||
const LORA_PROVIDER_NODE_TYPES$1 = [
|
||||
"Lora Stacker (LoraManager)",
|
||||
"Lora Randomizer (LoraManager)",
|
||||
"Lora Cycler (LoraManager)"
|
||||
"Lora Cycler (LoraManager)",
|
||||
"Create Hook LoRA (LoraManager)"
|
||||
];
|
||||
const LORA_STACK_AGGREGATOR_NODE_TYPES$1 = [
|
||||
"Lora Stack Combiner (LoraManager)"
|
||||
@@ -15781,7 +15782,8 @@ const ROOT_GRAPH_ID = "root";
|
||||
const LORA_PROVIDER_NODE_TYPES = [
|
||||
"Lora Stacker (LoraManager)",
|
||||
"Lora Randomizer (LoraManager)",
|
||||
"Lora Cycler (LoraManager)"
|
||||
"Lora Cycler (LoraManager)",
|
||||
"Create Hook LoRA (LoraManager)"
|
||||
];
|
||||
const LORA_STACK_AGGREGATOR_NODE_TYPES = [
|
||||
"Lora Stack Combiner (LoraManager)"
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -9,6 +9,7 @@ const LORA_NODE_CLASSES = new Set([
|
||||
"Lora Loader (LoraManager)",
|
||||
"Lora Stacker (LoraManager)",
|
||||
"WanVideo Lora Select (LoraManager)",
|
||||
"Create Hook LoRA (LoraManager)",
|
||||
]);
|
||||
|
||||
const TARGET_WIDGET_NAMES = new Set(["ckpt_name", "unet_name"]);
|
||||
|
||||
Reference in New Issue
Block a user