59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""Shared helpers for collections of local model metadata entries."""
|
|
|
|
import os
|
|
from collections.abc import Iterable, Mapping
|
|
from typing import Any, List
|
|
|
|
from .contracts import ResolvedModel
|
|
from .path_utils import get_filename_from_path, get_model_path_identity
|
|
from .type_utils import MODEL_EXTENSIONS
|
|
|
|
|
|
def is_model_file_path(path: str) -> bool:
|
|
"""Return whether ``path`` points to an existing supported model file."""
|
|
if not path or not os.path.isfile(path):
|
|
return False
|
|
|
|
filename = get_filename_from_path(path).lower()
|
|
if filename.endswith((".metadata.json", ".civitai.info")):
|
|
return False
|
|
|
|
return os.path.splitext(filename)[1].lower() in MODEL_EXTENSIONS
|
|
|
|
|
|
ModelRecord = ResolvedModel | Mapping[str, Any]
|
|
|
|
|
|
def normalize_models(models: Iterable[ModelRecord] | None) -> List[ResolvedModel]:
|
|
"""Normalize local model inputs once at the metadata boundary."""
|
|
result: List[ResolvedModel] = []
|
|
if models is None:
|
|
return result
|
|
if isinstance(models, (str, bytes, Mapping)) or not isinstance(models, Iterable):
|
|
raise TypeError("models must be an iterable of model records")
|
|
|
|
for model in models:
|
|
if isinstance(model, ResolvedModel):
|
|
result.append(model)
|
|
continue
|
|
if not isinstance(model, Mapping):
|
|
continue
|
|
try:
|
|
result.append(ResolvedModel.from_mapping(model))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return result
|
|
|
|
|
|
def dedupe_models(models: Iterable[ModelRecord] | None) -> List[ResolvedModel]:
|
|
"""Keep the first normalized local model for every filesystem identity."""
|
|
result: List[ResolvedModel] = []
|
|
seen = set()
|
|
for model in normalize_models(models):
|
|
identity = get_model_path_identity(model.path)
|
|
if not identity or identity in seen:
|
|
continue
|
|
seen.add(identity)
|
|
result.append(model)
|
|
return result
|