ComfyUI-Lora-Manager/py/utils/sidecar_paths.py

"""Resolution of sidecar metadata and preview storage paths.

All code that needs the on-disk location of a model's ``.metadata.json``
sidecar or preview assets MUST go through these helpers instead of deriving
paths inline (``splitext(model_path)[0] + ".metadata.json"`` and friends).

Two storage modes are supported, selected by the ``sidecar_storage_mode``
setting:

- ``alongside`` (default): sidecars and previews live next to the model
  file, the historical layout other tools may rely on.
- ``centralized``: sidecars and previews live under a configurable root
  (``sidecar_storage_path`` setting, default ``<settings_dir>/sidecars``),
  mirroring the library-relative directory structure:
  ``<root>/<library>/<root_basename-roothash>/<rel_dir>/<name>.metadata.json``.

All helpers are pure path computations: no directory scans and no file I/O
on the hot path. Settings lookups go through ``SettingsManager.get`` (a dict
read); config roots come from the already-initialized ``config`` singleton.
"""

from __future__ import annotations

import hashlib
import logging
import os
import re
from typing import List, Optional

logger = logging.getLogger(__name__)

METADATA_SUFFIX = ".metadata.json"

STORAGE_MODE_ALONGSIDE = "alongside"
STORAGE_MODE_CENTRALIZED = "centralized"

_VALID_MODES = frozenset({STORAGE_MODE_ALONGSIDE, STORAGE_MODE_CENTRALIZED})


def _get_settings_value(key: str, default=None):
    """Read a setting defensively; never fail path resolution on settings errors."""

    try:
        from ..services.settings_manager import get_settings_manager

        value = get_settings_manager().get(key)
    except Exception as exc:  # pragma: no cover - defensive fallback
        logger.debug("sidecar_paths: settings lookup for %r failed: %s", key, exc)
        return default
    return default if value is None else value


def get_storage_mode() -> str:
    """Return the active sidecar storage mode (``alongside`` unless configured)."""

    mode = _get_settings_value("sidecar_storage_mode", STORAGE_MODE_ALONGSIDE)
    if mode not in _VALID_MODES:
        return STORAGE_MODE_ALONGSIDE
    return mode


def is_centralized() -> bool:
    """Return True when centralized sidecar storage is active and resolvable."""

    return get_storage_mode() == STORAGE_MODE_CENTRALIZED and bool(get_sidecar_root())


def _resolve_root_from_settings() -> str:
    """Resolve the configured/default centralized root, ignoring the active mode."""

    configured = _get_settings_value("sidecar_storage_path", "")
    if configured and isinstance(configured, str):
        root = os.path.abspath(os.path.expanduser(configured.strip()))
        if root:
            return root

    # Default: <settings_dir>/sidecars
    try:
        from .settings_paths import get_settings_dir

        return os.path.join(get_settings_dir(), "sidecars")
    except Exception as exc:  # pragma: no cover - defensive fallback
        logger.warning("sidecar_paths: cannot resolve default sidecar root: %s", exc)
        return ""


def get_sidecar_root() -> str:
    """Return the absolute root directory for centralized sidecar storage.

    Empty string when centralized storage is not usable (mode alongside or an
    unresolvable configured path).
    """

    if get_storage_mode() != STORAGE_MODE_CENTRALIZED:
        return ""

    return _resolve_root_from_settings()


def get_configured_sidecar_root() -> str:
    """Return the centralized sidecar root regardless of the active mode.

    Unlike :func:`get_sidecar_root`, this resolves the configured
    ``sidecar_storage_path`` (or the ``<settings_dir>/sidecars`` default) even
    when the storage mode is ``alongside``. Migration tooling needs both
    layouts at once and must not depend on which mode is currently active.
    """

    return _resolve_root_from_settings()


def _installation_root() -> str:
    """Return the plugin installation directory (repository root)."""

    return os.path.dirname(
        os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    )


def _path_contains(base: str, path: str) -> bool:
    """Containment check tolerant of symlinked installs (custom_nodes links)."""

    for candidate in (os.path.abspath(path), os.path.realpath(path)):
        normalized = os.path.normcase(os.path.normpath(candidate))
        for root_variant in (os.path.abspath(base), os.path.realpath(base)):
            root_normalized = os.path.normcase(os.path.normpath(root_variant))
            if (
                normalized == root_normalized
                or normalized.startswith(root_normalized + os.sep)
            ):
                return True
    return False


def describe_sidecar_root() -> dict:
    """Describe the effective centralized sidecar root for UI display.

    ``inside_repo`` flags the portable-mode hazard: when settings live in the
    repository, the default root lands inside the plugin folder, where a
    reinstall or ``git clean`` would silently delete every sidecar.
    """

    configured = _get_settings_value("sidecar_storage_path", "")
    is_default = not (isinstance(configured, str) and configured.strip())
    root = _resolve_root_from_settings()
    return {
        "root": root,
        "is_default": is_default,
        "inside_repo": bool(root) and _path_contains(_installation_root(), root),
    }


def sanitize_path_component(name: str) -> str:
    """Return a filesystem-safe single path component."""

    safe = re.sub(r"[^A-Za-z0-9_.-]", "_", name or "")
    return safe or "_"


def _iter_model_roots() -> List[str]:
    """Return every configured model root for the active library."""

    try:
        from ..config import config
    except Exception as exc:  # pragma: no cover - defensive fallback
        logger.debug("sidecar_paths: config unavailable: %s", exc)
        return []

    roots: List[str] = []
    for attr in (
        "loras_roots",
        "base_models_roots",
        "embeddings_roots",
        "other_roots",
        "extra_loras_roots",
        "extra_checkpoints_roots",
        "extra_unet_roots",
        "extra_embeddings_roots",
    ):
        value = getattr(config, attr, None)
        if value:
            roots.extend(value)
    return roots


def _normalize_for_match(path: str) -> str:
    return os.path.normpath(os.path.abspath(path))


def root_mirror_component(root_path: str) -> str:
    """Return the mirror path component identifying a model root.

    ``<sanitized basename>-<hash>`` where the hash is a short digest of the
    normalized absolute root path. Two roots sharing a basename (e.g.
    ``/mnt/a/loras`` and ``/mnt/b/loras``) would otherwise map to the same
    mirror directory and overwrite each other's sidecars.
    """

    normalized = _normalize_for_match(root_path)
    digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:8]
    return f"{sanitize_path_component(os.path.basename(normalized))}-{digest}"


def resolve_centralized_dir(model_path: str) -> Optional[str]:
    """Return the centralized mirror directory for ``model_path``.

    The mirror layout is
    ``<sidecar_root>/<library>/<root_basename-roothash>/<rel_dir>``
    where ``rel_dir`` is the model's directory relative to the model root that
    contains it. The longest matching root wins so nested roots resolve to the
    most specific mirror. Returns ``None`` when centralized storage is inactive
    or the path is not under any configured model root.
    """

    return resolve_centralized_dir_for_dir(
        os.path.dirname(_normalize_for_match(model_path))
    )


def resolve_centralized_dir_for_dir(
    model_dir: str, *, sidecar_root: Optional[str] = None
) -> Optional[str]:
    """Return the centralized mirror directory for a model *directory*.

    Same layout as :func:`resolve_centralized_dir`, but accepts the directory
    itself. Used by folder-level operations (folder rename, mirror-tree walks)
    that have no model file path to derive from. Passing a configured model
    root returns the mirror base for that root.

    ``sidecar_root`` overrides the root lookup; pass
    :func:`get_configured_sidecar_root` to resolve mirror paths independently
    of the active storage mode (migration tooling).
    """

    root = sidecar_root if sidecar_root is not None else get_sidecar_root()
    if not root:
        return None

    normalized_dir = _normalize_for_match(model_dir)

    best_root: Optional[str] = None
    for candidate in _iter_model_roots():
        if not candidate:
            continue
        normalized = _normalize_for_match(candidate)
        if normalized_dir == normalized or normalized_dir.startswith(normalized + os.sep):
            if best_root is None or len(normalized) > len(best_root):
                best_root = normalized

    if best_root is None:
        return None

    try:
        from ..services.settings_manager import get_settings_manager

        library = get_settings_manager().get_active_library_name() or "default"
    except Exception:  # pragma: no cover - defensive fallback
        library = "default"

    rel_dir = os.path.relpath(normalized_dir, best_root)
    parts = [root, sanitize_path_component(library), root_mirror_component(best_root)]
    if rel_dir and rel_dir != os.curdir:
        parts.extend(sanitize_path_component(part) for part in rel_dir.split(os.sep) if part not in ("", os.curdir))
    return os.path.join(*parts)


def get_sidecar_dir(model_path: str) -> str:
    """Return the directory holding the model's sidecar/preview assets.

    Centralized mode falls back to the model's own directory (with a warning)
    when the path lies outside every configured model root.
    """

    if get_storage_mode() == STORAGE_MODE_CENTRALIZED:
        mirror = resolve_centralized_dir(model_path)
        if mirror:
            return mirror
        logger.warning(
            "sidecar_paths: %s is outside configured model roots; storing sidecar alongside",
            model_path,
        )
    return os.path.dirname(os.path.abspath(model_path))


def get_metadata_path(model_path: str) -> str:
    """Return the ``.metadata.json`` sidecar path for a model file."""

    base_name = os.path.splitext(os.path.basename(model_path))[0] + METADATA_SUFFIX
    return os.path.join(get_sidecar_dir(model_path), base_name)


def is_metadata_path(path: str) -> bool:
    """Return True when ``path`` already points at a metadata sidecar file."""

    return path.endswith(METADATA_SUFFIX)


def resolve_metadata_path(path: str) -> str:
    """Accept either a model path or a sidecar path and return the sidecar path."""

    if is_metadata_path(path):
        return path
    return get_metadata_path(path)


def get_preview_dir(model_path: str) -> str:
    """Return the directory holding the model's preview assets."""

    return get_sidecar_dir(model_path)