Upload 2 files

salia_console_get_node.py

@@ -0,0 +1,175 @@
+import os
+import shlex
+import subprocess
+from urllib.parse import urlparse
+
+
+NODES_PATH = "/ComfyUI/custom_nodes/comfyui-salia_online/nodes"
+DEFAULT_REPO_PATH = "https://huggingface.co/saliacoel/x/resolve/main/"
+
+
+class Salia_Console_Get_Node:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "items": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": True,
+                        "placeholder": "salia_example_node_1, salia_example_node_2.py, https://huggingface.co/other_hf/other_repo/resolve/main/salia_example_node_3.py, salia_example_4_nodepack.zip",
+                    },
+                ),
+            },
+        }
+
+    RETURN_TYPES = ("STRING",)
+    RETURN_NAMES = ("status",)
+    FUNCTION = "get_nodes"
+    CATEGORY = "salia/online"
+    OUTPUT_NODE = True
+
+    @classmethod
+    def IS_CHANGED(cls, items):
+        return float("nan")
+
+    def _log(self, message: str):
+        print(f"[Salia_Console_Get_Node] {message}")
+
+    def _split_items(self, items: str):
+        parts = items.replace("\n", ",").split(",")
+        cleaned = []
+
+        for part in parts:
+            part = part.strip().strip('"').strip("'")
+            if part:
+                cleaned.append(part)
+
+        return cleaned
+
+    def _is_url(self, value: str) -> bool:
+        # Accept both https:// and http:// as already-complete URLs.
+        return value.startswith(("https://", "http://"))
+
+    def _has_supported_extension(self, value: str) -> bool:
+        target = urlparse(value).path if self._is_url(value) else value
+        target = target.lower()
+        return target.endswith(".py") or target.endswith(".zip")
+
+    def _normalize_item(self, item: str) -> str:
+        item = item.strip().strip('"').strip("'")
+
+        # 1) Add .py if it is neither .py nor .zip
+        if not self._has_supported_extension(item):
+            item += ".py"
+
+        # 2) Prefix default Hugging Face repo if it is not already a URL
+        if not self._is_url(item):
+            item = DEFAULT_REPO_PATH + item.lstrip("/")
+
+        return item
+
+    def _filename_from_url(self, url: str) -> str:
+        filename = os.path.basename(urlparse(url).path)
+        if not filename:
+            raise ValueError(f"Could not determine filename from URL: {url}")
+        return filename
+
+    def _run_command(self, args, cwd=None):
+        command_text = " ".join(shlex.quote(str(arg)) for arg in args)
+        if cwd:
+            self._log(f"$ cd {cwd} && {command_text}")
+        else:
+            self._log(f"$ {command_text}")
+
+        completed = subprocess.run(
+            args,
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+        )
+
+        if completed.stdout:
+            print(completed.stdout)
+        if completed.stderr:
+            print(completed.stderr)
+
+        return completed
+
+    def get_nodes(self, items: str):
+        entries = self._split_items(items)
+
+        if not entries:
+            self._log("No items provided.")
+            return ("OK: 0\nERROR: 0",)
+
+        total_count = len(entries)
+        ok_count = total_count
+        error_count = 0
+
+        try:
+            os.makedirs(NODES_PATH, exist_ok=True)
+        except Exception as e:
+            self._log(f"Could not create/access nodes path: {e}")
+            return (f"OK: 0\nERROR: {total_count}",)
+
+        for original_item in entries:
+            try:
+                url = self._normalize_item(original_item)
+                filename = self._filename_from_url(url)
+                lower_filename = filename.lower()
+
+                self._log(f"Resolved: {original_item} -> {url}")
+
+                if lower_filename.endswith(".py"):
+                    completed = self._run_command(
+                        ["wget", "-O", filename, url],
+                        cwd=NODES_PATH,
+                    )
+                    if completed.returncode != 0:
+                        raise RuntimeError(f"wget failed for {filename}")
+
+                elif lower_filename.endswith(".zip"):
+                    completed = self._run_command(["apt-get", "update"])
+                    if completed.returncode != 0:
+                        raise RuntimeError("apt-get update failed")
+
+                    completed = self._run_command(["apt-get", "install", "-y", "unzip"])
+                    if completed.returncode != 0:
+                        raise RuntimeError("apt-get install -y unzip failed")
+
+                    completed = self._run_command(
+                        ["wget", "-O", filename, url],
+                        cwd=NODES_PATH,
+                    )
+                    if completed.returncode != 0:
+                        raise RuntimeError(f"wget failed for {filename}")
+
+                    completed = self._run_command(
+                        ["unzip", "-o", filename],
+                        cwd=NODES_PATH,
+                    )
+                    if completed.returncode != 0:
+                        raise RuntimeError(f"unzip failed for {filename}")
+
+                else:
+                    raise RuntimeError(f"Unsupported file type: {filename}")
+
+            except Exception as e:
+                error_count += 1
+                ok_count -= 1
+                self._log(f"ERROR for '{original_item}': {e}")
+
+        summary = f"OK: {ok_count}\nERROR: {error_count}"
+        self._log(summary.replace("\n", " | "))
+        return (summary,)
+
+
+NODE_CLASS_MAPPINGS = {
+    "Salia_Console_Get_Node": Salia_Console_Get_Node,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "Salia_Console_Get_Node": "Salia_Console_Get_Node",
+}

salia_get_lora_x.py

@@ -0,0 +1,157 @@
+import logging
+import os
+import shutil
+import urllib.parse
+import urllib.request
+
+import comfy.sd
+import comfy.utils
+import folder_paths
+
+
+HF_REPO_BASE = "https://huggingface.co/saliacoel/x/resolve/main"
+
+
+class Salia_Get_Lora_X:
+    """
+    Loads a single LoRA from the public Hugging Face repo `saliacoel/x`.
+
+    Behavior:
+      - Takes a single base name, e.g. `Fade_to_Black`
+      - Normalizes it to `Fade_to_Black.safetensors`
+      - Checks if it already exists in ComfyUI's loras folders
+      - If missing, downloads it
+      - Loads it and applies it model-only to one MODEL input
+    """
+
+    CATEGORY = "loaders/saliacoel"
+    FUNCTION = "load_lora"
+    RETURN_TYPES = ("MODEL",)
+    RETURN_NAMES = ("loaded_model",)
+    DESCRIPTION = (
+        "Loads a single LoRA from saliacoel/x. If missing locally, it is "
+        "downloaded first, then applied model-only."
+    )
+    OUTPUT_TOOLTIPS = (
+        "model_in with the LoRA applied.",
+    )
+
+    def __init__(self):
+        self.loaded_lora = None
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "filename": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": False,
+                        "placeholder": "Fade_to_Black",
+                        "tooltip": "LoRA name without .safetensors",
+                    },
+                ),
+                "model_in": (
+                    "MODEL",
+                    {"tooltip": "The MODEL input that will receive the LoRA."},
+                ),
+                "strength": (
+                    "FLOAT",
+                    {
+                        "default": 1.0,
+                        "min": -100.0,
+                        "max": 100.0,
+                        "step": 0.01,
+                        "tooltip": "Strength used when applying the LoRA.",
+                    },
+                ),
+            }
+        }
+
+    @staticmethod
+    def _normalize_lora_name(name: str) -> str:
+        base = os.path.basename((name or "").strip())
+        if not base:
+            raise ValueError("filename cannot be empty.")
+
+        if base.lower().endswith(".safetensors"):
+            base = base[: -len(".safetensors")]
+
+        if not base:
+            raise ValueError("filename resolves to an empty name.")
+
+        return f"{base}.safetensors"
+
+    @staticmethod
+    def _download_file(url: str, target_path: str) -> None:
+        os.makedirs(os.path.dirname(target_path), exist_ok=True)
+        tmp_path = target_path + ".download"
+
+        if os.path.exists(tmp_path):
+            os.remove(tmp_path)
+
+        try:
+            request = urllib.request.Request(
+                url,
+                headers={"User-Agent": "ComfyUI-SaliacoelSingleRepoLora/1.0"},
+            )
+            with urllib.request.urlopen(request) as response, open(tmp_path, "wb") as out_file:
+                shutil.copyfileobj(response, out_file)
+
+            os.replace(tmp_path, target_path)
+        except Exception:
+            if os.path.exists(tmp_path):
+                os.remove(tmp_path)
+            raise
+
+    @classmethod
+    def _ensure_lora_available(cls, lora_name: str) -> str:
+        existing_path = folder_paths.get_full_path("loras", lora_name)
+        if existing_path is not None:
+            return existing_path
+
+        lora_dirs = folder_paths.get_folder_paths("loras")
+        if not lora_dirs:
+            raise RuntimeError("No ComfyUI 'loras' folder is configured.")
+
+        target_dir = lora_dirs[0]
+        target_path = os.path.join(target_dir, lora_name)
+        url = f"{HF_REPO_BASE}/{urllib.parse.quote(lora_name)}"
+
+        logging.info("[SaliacoelSingleRepoLora] Downloading missing LoRA: %s", url)
+        cls._download_file(url, target_path)
+
+        resolved_path = folder_paths.get_full_path("loras", lora_name)
+        return resolved_path if resolved_path is not None else target_path
+
+    def _get_or_load_lora(self, lora_path: str):
+        if self.loaded_lora is not None and self.loaded_lora[0] == lora_path:
+            return self.loaded_lora[1]
+
+        lora = comfy.utils.load_torch_file(lora_path, safe_load=True)
+        self.loaded_lora = (lora_path, lora)
+        return lora
+
+    def _apply_lora_model_only(self, model, lora_path: str, strength: float):
+        if strength == 0:
+            return model
+
+        lora = self._get_or_load_lora(lora_path)
+        model_lora, _ = comfy.sd.load_lora_for_models(model, None, lora, strength, 0)
+        return model_lora
+
+    def load_lora(self, filename, model_in, strength):
+        lora_name = self._normalize_lora_name(filename)
+        lora_path = self._ensure_lora_available(lora_name)
+        loaded_model = self._apply_lora_model_only(model_in, lora_path, strength)
+        return (loaded_model,)
+
+
+NODE_CLASS_MAPPINGS = {
+    "Salia_Get_Lora_X": Salia_Get_Lora_X,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "Salia_Get_Lora_X": "Salia Get LoRa and Load LoRa X (1)",
+}