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.
This commit is contained in:
Will Miao
2026-09-26 09:02:37 +08:00
parent ede15032ce
commit 297d8787bd
14 changed files with 302 additions and 57 deletions
+2 -1
View File
@@ -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()
+21 -8
View File
@@ -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
)
+2 -1
View File
@@ -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
)
+3 -1
View File
@@ -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)
+20 -10
View File
@@ -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:
+6 -5
View File
@@ -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()
+11 -6
View File
@@ -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
+5 -4
View File
@@ -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)
+5 -1
View File
@@ -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}")
+3 -2
View File
@@ -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
+2 -4
View File
@@ -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", [])
@@ -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):
+5 -13
View File
@@ -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:
+215
View File
@@ -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 ``<settings_dir>/sidecars``),
mirroring the library-relative directory structure:
``<root>/<library>/<root_basename>/<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 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: <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 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 ``<sidecar_root>/<library>/<root_basename>/<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.
"""
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)