From 297d8787bd80adfabdad9dc2aaabab059e13bf76 Mon Sep 17 00:00:00 2001 From: Will Miao Date: Sat, 26 Sep 2026 09:02:37 +0800 Subject: [PATCH] refactor: route sidecar/preview path derivation through sidecar_paths helpers Phase 1 of #1045 (optional centralized sidecar storage): introduce py/utils/sidecar_paths.py as the single place that resolves .metadata.json and preview locations, and replace all inline splitext-based derivations across scanners, services, download manager, and route handlers. No behavior change: the default 'alongside' storage mode resolves every path exactly as before. .civitai.info (third-party sidecar) derivation is intentionally left co-located. --- py/metadata_ops/__init__.py | 3 +- py/routes/handlers/misc_handlers.py | 29 ++- py/routes/handlers/model_handlers.py | 3 +- py/services/autov3_backfill_service.py | 4 +- py/services/download_manager.py | 30 ++- py/services/metadata_sync_service.py | 11 +- py/services/model_lifecycle_service.py | 17 +- py/services/model_scanner.py | 9 +- py/services/pending_delete_service.py | 6 +- py/services/preview_asset_service.py | 5 +- py/services/tag_update_service.py | 6 +- .../use_cases/filename_template_use_case.py | 3 +- py/utils/metadata_manager.py | 18 +- py/utils/sidecar_paths.py | 215 ++++++++++++++++++ 14 files changed, 302 insertions(+), 57 deletions(-) create mode 100644 py/utils/sidecar_paths.py diff --git a/py/metadata_ops/__init__.py b/py/metadata_ops/__init__.py index a83356d8..40781ff4 100644 --- a/py/metadata_ops/__init__.py +++ b/py/metadata_ops/__init__.py @@ -172,12 +172,13 @@ async def download_preview( """ from ..services.downloader import get_downloader from ..utils.exif_utils import ExifUtils + from ..utils.sidecar_paths import get_preview_dir if not url or not url.strip(): return None base_name = os.path.splitext(os.path.basename(model_path))[0] - preview_dir = os.path.dirname(model_path) + preview_dir = get_preview_dir(model_path) output_path = os.path.join(preview_dir, base_name + ".webp") downloader = await get_downloader() diff --git a/py/routes/handlers/misc_handlers.py b/py/routes/handlers/misc_handlers.py index f3021935..8e156f6a 100644 --- a/py/routes/handlers/misc_handlers.py +++ b/py/routes/handlers/misc_handlers.py @@ -68,6 +68,7 @@ from ...utils.example_images_paths import ( ) from ...utils.lora_metadata import extract_trained_words from ...utils.session_logging import get_standalone_session_log_snapshot +from ...utils.sidecar_paths import get_metadata_path, get_preview_dir from ...utils.usage_stats import UsageStats from .base_model_handlers import BaseModelHandlerSet @@ -943,15 +944,24 @@ class DoctorHandler: os.rename(path, new_path) - for suffix in (".metadata.json", ".civitai.info"): - old_sidecar = old_base_no_ext + suffix - new_sidecar = new_base_no_ext + suffix - if os.path.exists(old_sidecar): - os.rename(old_sidecar, new_sidecar) + old_metadata_path = get_metadata_path(path) + new_metadata_path = get_metadata_path(new_path) + if os.path.exists(old_metadata_path): + os.rename(old_metadata_path, new_metadata_path) + + old_sidecar = old_base_no_ext + ".civitai.info" + new_sidecar = new_base_no_ext + ".civitai.info" + if os.path.exists(old_sidecar): + os.rename(old_sidecar, new_sidecar) for preview_ext in PREVIEW_EXTENSIONS: - old_preview = old_base_no_ext + preview_ext - new_preview = new_base_no_ext + preview_ext + old_preview = os.path.join( + get_preview_dir(path), base_name + preview_ext + ) + new_preview = os.path.join( + get_preview_dir(new_path), + candidate_base + preview_ext, + ) if os.path.exists(old_preview): os.rename(old_preview, new_preview) @@ -963,7 +973,10 @@ class DoctorHandler: old_preview_url = entry["preview_url"].replace("\\", "/") preview_ext = os.path.splitext(old_preview_url)[1] if preview_ext: - entry["preview_url"] = (new_base_no_ext + preview_ext).replace(os.sep, "/") + entry["preview_url"] = os.path.join( + get_preview_dir(new_path), + candidate_base + preview_ext, + ).replace(os.sep, "/") await scanner.update_single_model_cache( path, new_path, entry ) diff --git a/py/routes/handlers/model_handlers.py b/py/routes/handlers/model_handlers.py index 2968fbf8..43d7cbc2 100644 --- a/py/routes/handlers/model_handlers.py +++ b/py/routes/handlers/model_handlers.py @@ -50,6 +50,7 @@ from ...services.errors import RateLimitError, ResourceNotFoundError from ...utils.civitai_utils import resolve_license_payload from ...utils.file_utils import calculate_sha256 from ...utils.metadata_manager import MetadataManager +from ...utils.sidecar_paths import get_metadata_path from ...utils.url_utils import relative_root_prefix LICENSE_FIELDS = ( @@ -676,7 +677,7 @@ class ModelManagementHandler: status=400, ) - metadata_path = os.path.splitext(file_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(file_path) local_metadata = await self._metadata_sync.load_local_metadata( metadata_path ) diff --git a/py/services/autov3_backfill_service.py b/py/services/autov3_backfill_service.py index 60ea965d..b6778992 100644 --- a/py/services/autov3_backfill_service.py +++ b/py/services/autov3_backfill_service.py @@ -27,6 +27,8 @@ import os import threading from typing import TYPE_CHECKING, Optional +from ..utils.sidecar_paths import get_metadata_path + if TYPE_CHECKING: # pragma: no cover - type-check only; runtime imports are local from .model_scanner import ModelScanner @@ -41,7 +43,7 @@ def _resolve_autov3(file_path: str) -> str: safetensors header hash. Returns ``''`` when neither is available. """ try: - metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json" + metadata_path = get_metadata_path(file_path) if os.path.exists(metadata_path): with open(metadata_path, "r", encoding="utf-8") as handle: payload = json.load(handle) diff --git a/py/services/download_manager.py b/py/services/download_manager.py index 82164a5d..52518ec7 100644 --- a/py/services/download_manager.py +++ b/py/services/download_manager.py @@ -38,6 +38,7 @@ from ..utils.preview_selection import resolve_mature_threshold, select_preview_m from ..utils.utils import calculate_filename_for_model, sanitize_folder_name from ..utils.exif_utils import ExifUtils from ..utils.metadata_manager import MetadataManager +from ..utils.sidecar_paths import get_metadata_path, get_preview_dir from .service_registry import ServiceRegistry from .download_routing import is_diffusion_model_download, resolve_other_download_sub_type from .settings_manager import get_settings_manager @@ -831,7 +832,7 @@ class DownloadManager: ) for file_path in target_files: - metadata_path = os.path.splitext(file_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(file_path) deleted = await self._delete_file_with_retries(metadata_path) if not deleted and os.path.exists(metadata_path): logger.error(f"Error deleting metadata file: {metadata_path}") @@ -2447,7 +2448,7 @@ class DownloadManager: return {"success": False, "error": save_path} part_path = save_path + ".part" - metadata_path = os.path.splitext(save_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(save_path) pause_control = self._pause_events.get(download_id) if download_id else None @@ -2503,7 +2504,10 @@ class DownloadManager: if media_type == "video": preview_ext = _extension_from_url(preview_url, ".mp4") - preview_path = os.path.splitext(save_path)[0] + preview_ext + preview_path = os.path.join( + get_preview_dir(save_path), + os.path.splitext(os.path.basename(save_path))[0] + preview_ext, + ) rewritten_url, rewritten = rewrite_preview_url( preview_url, media_type="video" ) @@ -2530,7 +2534,10 @@ class DownloadManager: ) if rewritten and rewritten_url: preview_ext = _extension_from_url(preview_url, ".png") - preview_path = os.path.splitext(save_path)[0] + preview_ext + preview_path = os.path.join( + get_preview_dir(save_path), + os.path.splitext(os.path.basename(save_path))[0] + preview_ext, + ) success, _ = await downloader.download_file( rewritten_url, preview_path, use_auth=False ) @@ -2557,8 +2564,9 @@ class DownloadManager: temp_file_handle.write( content if isinstance(content, bytes) else content.encode("utf-8") ) - preview_path = ( - os.path.splitext(save_path)[0] + ".webp" + preview_path = os.path.join( + get_preview_dir(save_path), + os.path.splitext(os.path.basename(save_path))[0] + ".webp", ) optimized_data, _ = ExifUtils.optimize_image( @@ -2788,9 +2796,7 @@ class DownloadManager: entry = cast(Any, adjusted_entry) metadata_entries[index] = entry - metadata_file_path = ( - os.path.splitext(entry.file_path)[0] + ".metadata.json" - ) + metadata_file_path = get_metadata_path(entry.file_path) metadata_files_for_cleanup.append(metadata_file_path) await MetadataManager.save_metadata(entry.file_path, entry) @@ -3049,7 +3055,11 @@ class DownloadManager: extension = os.path.splitext(preview_path)[1] or ".webp" targets = [ - os.path.splitext(entry.file_path)[0] + extension for entry in entries + os.path.join( + get_preview_dir(entry.file_path), + os.path.splitext(os.path.basename(entry.file_path))[0] + extension, + ) + for entry in entries ] if not targets: diff --git a/py/services/metadata_sync_service.py b/py/services/metadata_sync_service.py index 92262a0e..337ec381 100644 --- a/py/services/metadata_sync_service.py +++ b/py/services/metadata_sync_service.py @@ -12,6 +12,7 @@ from ..services.settings_manager import SettingsManager from ..utils.civitai_utils import resolve_license_payload from ..utils.model_utils import determine_base_model from ..utils.models import autov3_from_civitai_files +from ..utils.sidecar_paths import get_metadata_path from .connectivity_guard import OFFLINE_FRIENDLY_MESSAGE, is_expected_offline_error from .errors import RateLimitError from .model_sources import has_external_source @@ -216,7 +217,7 @@ class MetadataSyncService: logger.error(error) return False, error - metadata_path = os.path.splitext(file_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(file_path) enable_archive = self._settings.get("enable_metadata_archive_db", False) previous_source = model_data.get("metadata_source") or (model_data.get("civitai") or {}).get("source") @@ -485,7 +486,7 @@ class MetadataSyncService: + (f" with version: {model_version_id}" if model_version_id else "") ) - metadata_path = os.path.splitext(file_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(file_path) await self.update_model_metadata( metadata_path, metadata, @@ -505,7 +506,7 @@ class MetadataSyncService: ) -> Dict[str, Any]: """Apply metadata updates and persist to disk and cache.""" - metadata_path = os.path.splitext(file_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(file_path) metadata = await metadata_loader(metadata_path) for key, value in updates.items(): @@ -554,7 +555,7 @@ class MetadataSyncService: } expected_hash: Optional[str] = None - first_metadata_path = os.path.splitext(file_paths[0])[0] + ".metadata.json" + first_metadata_path = get_metadata_path(file_paths[0]) first_metadata = await metadata_loader(first_metadata_path) if first_metadata and "sha256" in first_metadata: expected_hash = first_metadata["sha256"].lower() @@ -565,7 +566,7 @@ class MetadataSyncService: try: actual_hash = await hash_calculator(path) - metadata_path = os.path.splitext(path)[0] + ".metadata.json" + metadata_path = get_metadata_path(path) metadata = await metadata_loader(metadata_path) stored_hash = metadata.get("sha256", "").lower() diff --git a/py/services/model_lifecycle_service.py b/py/services/model_lifecycle_service.py index f180baec..bb59f97f 100644 --- a/py/services/model_lifecycle_service.py +++ b/py/services/model_lifecycle_service.py @@ -11,6 +11,7 @@ from ..services.service_registry import ServiceRegistry from ..services.pending_delete_service import get_pending_delete_service from ..utils.constants import PREVIEW_EXTENSIONS from ..utils.metadata_manager import MetadataManager +from ..utils.sidecar_paths import get_metadata_path logger = logging.getLogger(__name__) @@ -45,7 +46,10 @@ async def delete_model_artifacts( main_extension = ".safetensors" if main_extension is None else main_extension main_file = f"{file_name}{main_extension}" if main_extension else file_name - patterns = [main_file, f"{file_name}.metadata.json"] + patterns = [ + main_file, + os.path.basename(get_metadata_path(os.path.join(target_dir, main_file))), + ] for ext in PREVIEW_EXTENSIONS: patterns.append(f"{file_name}{ext}") @@ -260,7 +264,7 @@ class ModelLifecycleService: _require_path_in_library_roots(file_path, self._scanner, label="File path") - metadata_path = os.path.splitext(file_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(file_path) metadata = await self._metadata_loader(metadata_path) metadata["exclude"] = True @@ -315,7 +319,7 @@ class ModelLifecycleService: if not os.path.exists(file_path): raise ValueError("Model file does not exist") - metadata_path = os.path.splitext(file_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(file_path) metadata_payload = await self._metadata_loader(metadata_path) metadata_payload["exclude"] = False @@ -384,10 +388,11 @@ class ModelLifecycleService: if os.path.exists(new_file_path): raise ValueError("A file with this name already exists") + metadata_filename = os.path.basename(get_metadata_path(file_path)) patterns = [ f"{old_file_name}{old_extension}", - f"{old_file_name}.metadata.json", - f"{old_file_name}.metadata.json.bak", + metadata_filename, + f"{metadata_filename}.bak", ] for ext in PREVIEW_EXTENSIONS: patterns.append(f"{old_file_name}{ext}") @@ -398,7 +403,7 @@ class ModelLifecycleService: if os.path.exists(path): existing_files.append((path, pattern)) - metadata_path = os.path.join(target_dir, f"{old_file_name}.metadata.json") + metadata_path = get_metadata_path(file_path) metadata: Optional[Dict[str, object]] = None hash_value: Optional[str] = None diff --git a/py/services/model_scanner.py b/py/services/model_scanner.py index 542671ae..a8525165 100644 --- a/py/services/model_scanner.py +++ b/py/services/model_scanner.py @@ -11,6 +11,7 @@ from ..utils.models import BaseModelMetadata, autov3_from_civitai_files from ..config import config from ..utils.file_utils import find_preview_file, get_preview_extension, calculate_sha256, calculate_autov3 from ..utils.metadata_manager import MetadataManager +from ..utils.sidecar_paths import get_metadata_path, get_preview_dir from ..utils.civitai_utils import resolve_license_info from .model_cache import ModelCache from .model_hash_index import ModelHashIndex @@ -1709,7 +1710,7 @@ class ModelScanner: file_path = item.get("file_path") if not file_path: continue - metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json" + metadata_path = get_metadata_path(file_path) if not os.path.exists(metadata_path): continue try: @@ -2339,7 +2340,7 @@ class ModelScanner: target_associated_path = os.path.join(target_path, new_associated_filename) # Store metadata file path for special handling - if file == f"{base_name}.metadata.json": + if file == os.path.basename(get_metadata_path(source_path)): source_metadata = source_file_path moved_metadata_path = target_associated_path else: @@ -2399,7 +2400,7 @@ class ModelScanner: metadata['file_name'] = os.path.splitext(os.path.basename(model_path))[0] if 'preview_url' in metadata and metadata['preview_url']: - preview_dir = os.path.dirname(model_path) + preview_dir = get_preview_dir(model_path) # Update preview filename to match the new base name new_base_name = os.path.splitext(os.path.basename(model_path))[0] preview_ext = get_preview_extension(metadata['preview_url']) @@ -2759,7 +2760,7 @@ class ModelScanner: # Sidecar write-back: JSON null encodes the checked-unavailable # state. Skip silently when the sidecar does not exist. - metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json" + metadata_path = get_metadata_path(file_path) if os.path.exists(metadata_path): with open(metadata_path, 'r', encoding='utf-8') as handle: payload = json.load(handle) diff --git a/py/services/pending_delete_service.py b/py/services/pending_delete_service.py index 39d36ea8..70bc0746 100644 --- a/py/services/pending_delete_service.py +++ b/py/services/pending_delete_service.py @@ -38,6 +38,7 @@ from typing import ( ) from ..utils.constants import PREVIEW_EXTENSIONS +from ..utils.sidecar_paths import get_metadata_path from ..utils import settings_paths logger = logging.getLogger(__name__) @@ -669,7 +670,10 @@ class PendingDeleteService: """Enumerate existing artifacts exactly like delete_model_artifacts.""" main_extension = ".safetensors" if main_extension is None else main_extension main_file = f"{file_name}{main_extension}" if main_extension else file_name - patterns = [main_file, f"{file_name}.metadata.json"] + patterns = [ + main_file, + os.path.basename(get_metadata_path(os.path.join(target_dir, main_file))), + ] for ext in PREVIEW_EXTENSIONS: patterns.append(f"{file_name}{ext}") diff --git a/py/services/preview_asset_service.py b/py/services/preview_asset_service.py index ac84ec5e..1e215347 100644 --- a/py/services/preview_asset_service.py +++ b/py/services/preview_asset_service.py @@ -10,6 +10,7 @@ from urllib.parse import urlparse from ..utils.constants import CARD_PREVIEW_WIDTH, PREVIEW_EXTENSIONS from ..utils.civitai_utils import rewrite_preview_url from ..utils.preview_selection import resolve_mature_threshold, select_preview_media +from ..utils.sidecar_paths import get_metadata_path, get_preview_dir from .settings_manager import get_settings_manager logger = logging.getLogger(__name__) @@ -159,7 +160,7 @@ class PreviewAssetService: """Replace an existing preview asset for a model.""" base_name = os.path.splitext(os.path.basename(model_path))[0] - folder = os.path.dirname(model_path) + folder = get_preview_dir(model_path) extension, optimized_data = await self._convert_preview( preview_data, content_type, original_filename @@ -179,7 +180,7 @@ class PreviewAssetService: with open(preview_path, "wb") as handle: handle.write(optimized_data) - metadata_path = os.path.splitext(model_path)[0] + ".metadata.json" + metadata_path = get_metadata_path(model_path) metadata = await metadata_loader(metadata_path) metadata["preview_url"] = preview_path metadata["preview_nsfw_level"] = nsfw_level diff --git a/py/services/tag_update_service.py b/py/services/tag_update_service.py index c1081384..d87ab221 100644 --- a/py/services/tag_update_service.py +++ b/py/services/tag_update_service.py @@ -2,10 +2,9 @@ from __future__ import annotations -import os - from typing import Awaitable, Callable, Dict, List, Sequence, Tuple +from ..utils.sidecar_paths import get_metadata_path from .auto_tag_service import extract_auto_tags @@ -24,8 +23,7 @@ class TagUpdateService: update_cache: Callable[[str, str, Dict[str, object]], Awaitable[bool]], ) -> Tuple[List[str], List[str]]: """Add tags to a metadata entry and return updated tags and auto_tags.""" - base, _ = os.path.splitext(file_path) - metadata_path = f"{base}.metadata.json" + metadata_path = get_metadata_path(file_path) metadata = await metadata_loader(metadata_path) raw_tags = metadata.get("tags", []) diff --git a/py/services/use_cases/filename_template_use_case.py b/py/services/use_cases/filename_template_use_case.py index 0a0813ef..e569d681 100644 --- a/py/services/use_cases/filename_template_use_case.py +++ b/py/services/use_cases/filename_template_use_case.py @@ -12,6 +12,7 @@ import os from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence from ...utils.constants import AUTO_ORGANIZE_BATCH_SIZE +from ...utils.sidecar_paths import get_metadata_path from ...utils.utils import calculate_filename_for_model from ..model_file_service import AutoOrganizeResult, ProgressCallback from ..model_lifecycle_service import ModelLifecycleService, load_local_metadata @@ -200,7 +201,7 @@ class FilenameTemplateUseCase: sidecar or no ``original_file_name`` entry exists (models never renamed, or renamed before the recording shipped). """ - metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json" + metadata_path = get_metadata_path(file_path) metadata = await self._metadata_loader(metadata_path) original = metadata.get("original_file_name") if not isinstance(original, str): diff --git a/py/utils/metadata_manager.py b/py/utils/metadata_manager.py index 43cfe976..1b03a813 100644 --- a/py/utils/metadata_manager.py +++ b/py/utils/metadata_manager.py @@ -8,6 +8,7 @@ from typing import Any, Dict, Optional, Type, Union, cast from .models import BaseModelMetadata, CheckpointMetadata, EmbeddingMetadata, LoraMetadata from .file_utils import normalize_path, find_preview_file, calculate_sha256, calculate_autov3 from .lora_metadata import extract_lora_metadata, extract_checkpoint_metadata +from .sidecar_paths import get_metadata_path, resolve_metadata_path logger = logging.getLogger(__name__) @@ -32,7 +33,7 @@ class MetadataManager: - metadata: BaseModelMetadata instance or None - should_skip: True if corrupted metadata file exists and model should be skipped """ - metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json" + metadata_path = get_metadata_path(file_path) # Check if metadata file exists if not os.path.exists(metadata_path): @@ -98,11 +99,7 @@ class MetadataManager: payload.update(unknown_fields) else: if not should_skip: - metadata_path = ( - file_path - if file_path.endswith(".metadata.json") - else f"{os.path.splitext(file_path)[0]}.metadata.json" - ) + metadata_path = resolve_metadata_path(file_path) if os.path.exists(metadata_path): try: with open(metadata_path, "r", encoding="utf-8") as handle: @@ -150,7 +147,7 @@ class MetadataManager: return model_data folder = model_data.get("folder") - metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json" + metadata_path = get_metadata_path(file_path) sidecar_exists = os.path.exists(metadata_path) cached = model_data.copy() payload = await MetadataManager.load_metadata_payload(file_path) @@ -188,12 +185,7 @@ class MetadataManager: bool: Success or failure """ # Determine if the input is a metadata path or a model file path - if path.endswith('.metadata.json'): - metadata_path = path - else: - # Use existing logic for model file paths - file_path = path - metadata_path = f"{os.path.splitext(file_path)[0]}.metadata.json" + metadata_path = resolve_metadata_path(path) temp_path = f"{metadata_path}.tmp" try: diff --git a/py/utils/sidecar_paths.py b/py/utils/sidecar_paths.py new file mode 100644 index 00000000..05d89e2e --- /dev/null +++ b/py/utils/sidecar_paths.py @@ -0,0 +1,215 @@ +"""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 ``/sidecars``), + mirroring the library-relative directory structure: + ``////.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 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 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 "" + + 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: /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 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 resolve_centralized_dir(model_path: str) -> Optional[str]: + """Return the centralized mirror directory for ``model_path``. + + The mirror layout is ``///`` + 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. + """ + + root = get_sidecar_root() + if not root: + return None + + target = _normalize_for_match(model_path) + model_dir = os.path.dirname(target) + + best_root: Optional[str] = None + for candidate in _iter_model_roots(): + if not candidate: + continue + normalized = _normalize_for_match(candidate) + if model_dir == normalized or model_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(model_dir, best_root) + parts = [root, sanitize_path_component(library), sanitize_path_component(os.path.basename(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)