refactor: extract shared utilities and deduplicate metadata path definitions
- Add first_non_empty() and to_int() to core/type_utils.py - Add get_filename_from_path() and METADATA_DIR to core/path_utils.py - Replace inline implementations across resolver, downloader, and sources with calls to the shared utilities - Move strip_known_model_extension from resolver to matcher
This commit is contained in:
+6
-13
@@ -86,6 +86,8 @@ class ModelResolverExtension:
|
||||
)
|
||||
from .core.path_templates import infer_download_path_templates
|
||||
from .core.scanner import get_model_files, invalidate_model_files_cache
|
||||
from .core.path_utils import get_filename_from_path
|
||||
from .core.type_utils import first_non_empty, to_int
|
||||
from .core.settings import (
|
||||
bool_setting as resolver_bool_setting,
|
||||
get_default_root_for_category,
|
||||
@@ -571,7 +573,7 @@ class ModelResolverExtension:
|
||||
for m in available_models:
|
||||
rel_path = m.get("relative_path", "")
|
||||
if rel_path:
|
||||
filename = rel_path.split("/")[-1].split("\\")[-1]
|
||||
filename = get_filename_from_path(rel_path)
|
||||
path_by_filename[filename] = m.get("path")
|
||||
# Also add without extension for matching (simple approach)
|
||||
if "." in filename:
|
||||
@@ -632,7 +634,7 @@ class ModelResolverExtension:
|
||||
category = ref.get("category", "unknown")
|
||||
|
||||
# Determine model name and strength
|
||||
model_name = original_path.split("/")[-1].split("\\")[-1]
|
||||
model_name = get_filename_from_path(original_path)
|
||||
strength = None
|
||||
|
||||
# For standard LoraLoader nodes, strength is in next widget_value
|
||||
@@ -2252,19 +2254,10 @@ class ModelResolverExtension:
|
||||
)
|
||||
|
||||
def _first_metadata_value(*values):
|
||||
for value in values:
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, str) and not value.strip():
|
||||
continue
|
||||
return value
|
||||
return ""
|
||||
return first_non_empty(*values)
|
||||
|
||||
def _metadata_int(value):
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return to_int(value)
|
||||
|
||||
inferred_source = ""
|
||||
if "civitai.com" in url:
|
||||
|
||||
+2
-10
@@ -24,7 +24,7 @@ from .log_system.log_funcs import (
|
||||
)
|
||||
from .resolver import normalize_sha256
|
||||
from .path_utils import is_path_within, get_path_identity, write_json_atomic
|
||||
from .type_utils import as_dict, as_list
|
||||
from .type_utils import as_dict, as_list, first_non_empty
|
||||
|
||||
try:
|
||||
import folder_paths
|
||||
@@ -135,15 +135,7 @@ def _as_list(value: Any) -> List[Any]:
|
||||
|
||||
|
||||
def _first_present(*values: Any) -> Any:
|
||||
for value in values:
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, str) and not value.strip():
|
||||
continue
|
||||
if isinstance(value, (list, dict)) and not value:
|
||||
continue
|
||||
return value
|
||||
return ""
|
||||
return first_non_empty(*values)
|
||||
|
||||
|
||||
def _coerce_int_or_value(value: Any) -> Any:
|
||||
|
||||
@@ -148,3 +148,17 @@ def write_json_atomic(
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
|
||||
|
||||
def get_filename_from_path(path: Any) -> str:
|
||||
"""Return the filename/basename of a path, handling both forward and backward slashes."""
|
||||
if not path:
|
||||
return ""
|
||||
return str(path).replace("\\", "/").split("/")[-1]
|
||||
|
||||
|
||||
# Path to metadata directory
|
||||
METADATA_DIR = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "metadata"
|
||||
)
|
||||
|
||||
|
||||
+2
-11
@@ -26,7 +26,7 @@ from .workflow_analyzer import (
|
||||
identify_missing_models,
|
||||
should_scan_as_model_reference,
|
||||
)
|
||||
from .matcher import find_matches
|
||||
from .matcher import find_matches, strip_known_model_extension
|
||||
from .type_utils import as_dict, as_list, MODEL_EXTENSIONS as _MODEL_EXTENSIONS
|
||||
from .workflow_updater import update_workflow_nodes
|
||||
from .sources.civitai import resolve_urn
|
||||
@@ -49,16 +49,7 @@ from .path_utils import (
|
||||
)
|
||||
|
||||
|
||||
def strip_known_model_extension(filename: str) -> str:
|
||||
"""Strip only known model extensions, preserving names like v4.0."""
|
||||
if not isinstance(filename, str):
|
||||
return ""
|
||||
|
||||
lowered = filename.lower()
|
||||
for ext in MODEL_EXTENSIONS:
|
||||
if lowered.endswith(ext):
|
||||
return filename[: -len(ext)]
|
||||
return filename
|
||||
# Imported from .matcher
|
||||
|
||||
|
||||
def get_workflow_url_info_for_filename(
|
||||
|
||||
@@ -22,6 +22,7 @@ from ..matcher import (
|
||||
base_model_matches as _base_model_matches,
|
||||
base_model_score as _base_model_score,
|
||||
)
|
||||
from ..type_utils import to_int
|
||||
from ..progress import report_progress
|
||||
from ..log_system.log_funcs import (
|
||||
log_debug,
|
||||
@@ -105,12 +106,7 @@ def is_civarchive_available() -> bool:
|
||||
|
||||
|
||||
def _coerce_int(value: Any) -> Optional[int]:
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return to_int(value)
|
||||
|
||||
|
||||
def _size_kb_to_bytes(value: Any) -> Optional[int]:
|
||||
|
||||
@@ -20,6 +20,7 @@ from ..matcher import (
|
||||
base_model_matches as _base_model_matches,
|
||||
base_model_score as _base_model_score,
|
||||
)
|
||||
from ..type_utils import first_non_empty
|
||||
from ..progress import report_progress
|
||||
from ..log_system.log_funcs import (
|
||||
log_debug,
|
||||
@@ -1750,15 +1751,7 @@ def _as_metadata_list(value: Any) -> List[Any]:
|
||||
|
||||
|
||||
def _first_metadata_value(*values: Any) -> Any:
|
||||
for value in values:
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, str) and not value.strip():
|
||||
continue
|
||||
if isinstance(value, (list, tuple, dict)) and not value:
|
||||
continue
|
||||
return value
|
||||
return None
|
||||
return first_non_empty(*values, default=None)
|
||||
|
||||
|
||||
def _metadata_size_to_bytes(value: Any) -> Optional[int]:
|
||||
|
||||
@@ -21,16 +21,14 @@ from ..log_system.log_funcs import (
|
||||
log_error,
|
||||
log_exception,
|
||||
)
|
||||
from ..path_utils import write_json_atomic
|
||||
from ..path_utils import write_json_atomic, METADATA_DIR
|
||||
|
||||
HF_API_URL = "https://huggingface.co/api"
|
||||
HF_AUTHOR_FALLBACKS = ["Comfy-Org"]
|
||||
BRAVE_SEARCH_API_URL = "https://api.search.brave.com/res/v1/web/search"
|
||||
HF_AUTHOR_INDEX_CACHE_TTL_SECONDS = 24 * 60 * 60
|
||||
HF_AUTHOR_INDEX_CACHE_VERSION = 1
|
||||
METADATA_DIR = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "metadata"
|
||||
)
|
||||
|
||||
HF_AUTHOR_INDEX_CACHE_PATH = os.path.join(
|
||||
METADATA_DIR, "huggingface-author-index.json"
|
||||
)
|
||||
|
||||
@@ -23,10 +23,7 @@ from ..log_system.log_funcs import (
|
||||
log_exception,
|
||||
)
|
||||
|
||||
# Path to metadata directory
|
||||
METADATA_DIR = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "metadata"
|
||||
)
|
||||
from ..path_utils import METADATA_DIR
|
||||
MODEL_LIST_FILE = os.path.join(METADATA_DIR, "model-list.json")
|
||||
MODEL_LIST_META_FILE = os.path.join(METADATA_DIR, "model-list.meta.json")
|
||||
MODEL_LIST_SOURCE_URL = (
|
||||
|
||||
@@ -20,10 +20,7 @@ from ..log_system.log_funcs import (
|
||||
log_exception,
|
||||
)
|
||||
|
||||
# Path to metadata directory
|
||||
METADATA_DIR = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "metadata"
|
||||
)
|
||||
from ..path_utils import METADATA_DIR
|
||||
POPULAR_MODELS_FILE = os.path.join(METADATA_DIR, "popular-models.json")
|
||||
MODEL_ALIASES_FILE = os.path.join(METADATA_DIR, "model-aliases.json")
|
||||
BASE_MODELS_FILE = os.path.join(METADATA_DIR, "base-models.json")
|
||||
|
||||
@@ -56,3 +56,32 @@ MODEL_EXTENSIONS = {
|
||||
".onnx",
|
||||
".gguf",
|
||||
}
|
||||
|
||||
|
||||
def first_non_empty(*values: Any, default: Any = "") -> Any:
|
||||
"""
|
||||
Return the first value that is not None, not an empty/whitespace-only string,
|
||||
and not an empty collection.
|
||||
"""
|
||||
for value in values:
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, str) and not value.strip():
|
||||
continue
|
||||
if isinstance(value, (list, tuple, dict, set)) and not value:
|
||||
continue
|
||||
return value
|
||||
return default
|
||||
|
||||
|
||||
def to_int(value: Any, default: Any = None) -> Any:
|
||||
"""
|
||||
Safely cast a value to an integer, returning a default value on failure.
|
||||
"""
|
||||
if value is None or value == "":
|
||||
return default
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
|
||||
Reference in New Issue
Block a user