nodes/Azornes-Comfyui-Model-Resolver
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: centralize type coercion and file I/O helpers to eliminate duplication

Browse Source
This commit is contained in:
Dariusz L
2026-06-25 17:33:59 +02:00
parent d979bbcf21
commit 6850ef1267
11 changed files with 115 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+11 -61
View File
@@ -87,7 +87,7 @@ 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.type_utils import first_non_empty, to_int, to_bool
from .core.settings import (
bool_setting as resolver_bool_setting,
get_default_root_for_category,
@@ -479,12 +479,7 @@ class ModelResolverExtension:
data = await request.json()
filename = data.get("filename", "")
category = data.get("category", "")
force_rescan = data.get("force_rescan", False)
force_rescan = (
force_rescan
if isinstance(force_rescan, bool)
else str(force_rescan).lower() == "true"
)
force_rescan = to_bool(data.get("force_rescan"), False)
if not filename:
return web.json_response(
@@ -1121,27 +1116,11 @@ class ModelResolverExtension:
base_model_context = data.get("base_model_context", "")
progress_id = str(data.get("progress_id") or "").strip()
progress_source = str(data.get("progress_source") or "").strip()
civitai_candidate_limit_raw = data.get(
"civitai_candidate_limit", 5
)
try:
civitai_candidate_limit = int(
civitai_candidate_limit_raw
)
except (TypeError, ValueError):
civitai_candidate_limit = 5
civitai_candidate_limit = to_int(data.get("civitai_candidate_limit"), 5)
civitai_candidate_limit = max(
1, min(civitai_candidate_limit, 20)
)
civarchive_candidate_limit_raw = data.get(
"civarchive_candidate_limit", 10
)
try:
civarchive_candidate_limit = int(
civarchive_candidate_limit_raw
)
except (TypeError, ValueError):
civarchive_candidate_limit = 10
civarchive_candidate_limit = to_int(data.get("civarchive_candidate_limit"), 10)
civarchive_candidate_limit = max(
1, min(civarchive_candidate_limit, 30)
)
@@ -1163,36 +1142,12 @@ class ModelResolverExtension:
hf_use_brave_fallback = data.get(
"hf_use_brave_fallback", True
)
is_urn = (
is_urn_raw
if isinstance(is_urn_raw, bool)
else (str(is_urn_raw).lower() == "true")
)
hf_use_api_search = (
hf_use_api_search
if isinstance(hf_use_api_search, bool)
else str(hf_use_api_search).lower() == "true"
)
civitai_use_trpc_search = (
civitai_use_trpc_search
if isinstance(civitai_use_trpc_search, bool)
else str(civitai_use_trpc_search).lower() == "true"
)
civitai_use_html_fallback = (
civitai_use_html_fallback
if isinstance(civitai_use_html_fallback, bool)
else str(civitai_use_html_fallback).lower() == "true"
)
hf_use_comfy_org_fallback = (
hf_use_comfy_org_fallback
if isinstance(hf_use_comfy_org_fallback, bool)
else str(hf_use_comfy_org_fallback).lower() == "true"
)
hf_use_brave_fallback = (
hf_use_brave_fallback
if isinstance(hf_use_brave_fallback, bool)
else str(hf_use_brave_fallback).lower() == "true"
)
is_urn = to_bool(is_urn_raw, False)
hf_use_api_search = to_bool(hf_use_api_search, True)
civitai_use_trpc_search = to_bool(civitai_use_trpc_search, True)
civitai_use_html_fallback = to_bool(civitai_use_html_fallback, True)
hf_use_comfy_org_fallback = to_bool(hf_use_comfy_org_fallback, True)
hf_use_brave_fallback = to_bool(hf_use_brave_fallback, True)
# For URN-only requests, model_id and version_id are required instead of filename
model_id = data.get("model_id")
@@ -1241,12 +1196,7 @@ class ModelResolverExtension:
if len(normalized_sources) == 1
else "all"
)
force_search = data.get("force_search", False)
force_search = (
force_search
if isinstance(force_search, bool)
else str(force_search).lower() == "true"
)
force_search = to_bool(data.get("force_search"), False)
update_search_progress(
progress_id,
core/downloader.py
+4 -15
View File
@@ -23,7 +23,7 @@ from .log_system.log_funcs import (
log_exception,
)
from .resolver import normalize_sha256
from .path_utils import is_path_within, get_path_identity, write_json_atomic
from .path_utils import is_path_within, get_path_identity, write_json_atomic, read_json_safe, calculate_file_sha256 as _calculate_file_sha256
from .type_utils import as_dict, as_list, first_non_empty
try:
@@ -271,14 +271,8 @@ def read_completed_metadata_sha256(file_path: str) -> str:
if not os.path.exists(metadata_path):
return ""
try:
with open(metadata_path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
except Exception as e:
log_warn(f"Could not read metadata sidecar for SHA256 check: {metadata_path} ({e})")
return ""
if not isinstance(payload, dict):
payload = read_json_safe(metadata_path, {})
if not isinstance(payload, dict) or not payload:
return ""
hash_status = str(payload.get("hash_status") or "completed").strip().lower()
@@ -293,12 +287,7 @@ def read_completed_metadata_sha256(file_path: str) -> str:
def calculate_file_sha256(file_path: str) -> str:
"""Calculate SHA256 for an existing local file."""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as handle:
for chunk in iter(lambda: handle.read(CHUNK_SIZE), b""):
if chunk:
sha256_hash.update(chunk)
return sha256_hash.hexdigest()
return _calculate_file_sha256(file_path) or ""
def build_lora_manager_metadata(
core/matcher.py
+12
View File
@@ -388,3 +388,15 @@ def calculate_archived_model_confidence(
return round(best * 100, 1)
def calculate_candidate_rank(
confidence: float,
base_model: Optional[str],
base_model_context: Optional[str],
) -> Tuple[bool, float]:
"""Calculate whether base model matches context and compute candidate's final rank score."""
matches = base_model_matches(base_model, base_model_context)
score = base_model_score(base_model, base_model_context)
return matches, confidence + score
core/path_utils.py
+34
View File
@@ -6,9 +6,12 @@ and directory matching to be shared across all modules.
"""
import os
import json
import hashlib
from typing import Any, Tuple, Optional
def get_path_identity(path: str) -> str:
"""Return a stable path identity across symlinks/junctions."""
if not path:
@@ -162,3 +165,34 @@ METADATA_DIR = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "metadata"
)
def calculate_file_sha256(file_path: str, chunk_size: int = 131072) -> Optional[str]:
"""Calculate SHA256 hex digest for an existing local file safely."""
if not file_path or not os.path.exists(file_path):
return None
sha256_hash = hashlib.sha256()
try:
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(chunk_size), b""):
if byte_block:
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
except Exception as e:
from .log_system.log_funcs import log_error
log_error(f"Error computing hash for {file_path}: {e}")
return None
def read_json_safe(file_path: str, default: Any = None) -> Any:
"""Safely read and parse a JSON file, returning a default value on error."""
if not file_path or not os.path.exists(file_path):
return default
try:
with open(file_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
from .log_system.log_funcs import log_warn
log_warn(f"Error reading JSON from {os.path.basename(file_path)}: {e}")
return default
core/settings.py
+8 -21
View File
@@ -6,7 +6,7 @@ import json
import re
from pathlib import Path
from typing import Any, Dict, Iterable, Mapping, Optional
from .path_utils import write_json_atomic
from .path_utils import write_json_atomic, read_json_safe
SETTINGS_FILE = Path(__file__).resolve().parents[1] / "model_resolver_settings.json"
@@ -120,18 +120,12 @@ _INVALID_SEGMENT_CHARS_RE = re.compile(r'[<>:"|?*\x00-\x1f]+')
_UNKNOWN_PLACEHOLDER_RE = re.compile(r"\{[^{}]+\}")
from .type_utils import to_bool
def bool_setting(value: Any, default: bool = True) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
return bool(value)
return to_bool(value, default)
def normalize_download_category(category: str) -> str:
@@ -188,15 +182,8 @@ def normalize_download_path_template(template: Any) -> str:
def _read_settings_file() -> Dict[str, Any]:
try:
if SETTINGS_FILE.is_file():
with SETTINGS_FILE.open("r", encoding="utf-8") as handle:
data = json.load(handle)
if isinstance(data, dict):
return data
except Exception:
return {}
return {}
data = read_json_safe(str(SETTINGS_FILE), {})
return data if isinstance(data, dict) else {}
def _coerce_dict(value: Any) -> Dict[str, Any]:
core/sources/civarchive.py
+3 -5
View File
@@ -21,6 +21,7 @@ from ..matcher import (
normalize_base_model as _normalize_base_model,
base_model_matches as _base_model_matches,
base_model_score as _base_model_score,
calculate_candidate_rank,
)
from ..type_utils import to_int
from ..progress import report_progress
@@ -1555,11 +1556,8 @@ def search_civarchive_for_file(
resolved["confidence"] = confidence
resolved["match_type"] = "exact" if confidence == 100.0 else "similar"
base_model_matches = _base_model_matches(
resolved.get("base_model"), base_model_context
)
rank = confidence + _base_model_score(
resolved.get("base_model"), base_model_context
base_model_matches, rank = calculate_candidate_rank(
confidence, resolved.get("base_model"), base_model_context
)
if rank > best_rank:
core/sources/civitai.py
+2 -14
View File
@@ -22,6 +22,7 @@ from ..matcher import (
)
from ..type_utils import first_non_empty
from ..progress import report_progress
from ..path_utils import calculate_file_sha256
from ..log_system.log_funcs import (
log_debug,
log_info,
@@ -1412,20 +1413,7 @@ def _get_sha256_hash(file_path: str) -> Optional[str]:
Returns:
SHA256 hash as hex string, or None if file doesn't exist
"""
if not file_path or not os.path.exists(file_path):
return None
BUF_SIZE = 1024 * 128 # 128KB chunks
sha256_hash = hashlib.sha256()
try:
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(BUF_SIZE), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
except Exception as e:
log_error(f"Error computing hash for {file_path}: {e}")
return None
return calculate_file_sha256(file_path)
def get_model_info_by_hash(
core/sources/lora_manager_archive.py
+5 -7
View File
@@ -16,6 +16,7 @@ from ..matcher import (
normalize_base_model as _normalize_base_model,
base_model_matches as _base_model_matches,
base_model_score as _base_model_score,
calculate_candidate_rank,
)
from ..progress import report_progress
from ..log_system.log_funcs import (
@@ -616,8 +617,8 @@ def search_lora_manager_archive(
)
if confidence < 40:
continue
rank = confidence + _base_model_score(
row["base_model"], base_model_context
_, rank = calculate_candidate_rank(
confidence, row["base_model"], base_model_context
)
scored_candidates.append((rank, confidence, row))
@@ -765,14 +766,11 @@ def search_lora_manager_archive_for_file(
confidence = calculate_archived_model_confidence(filename, candidate.get("name", ""), candidate.get("version_name", ""), candidate_filename)
if exact_only and confidence < 100.0:
continue
base_model_matches = _base_model_matches(
candidate.get("base_model"), base_model_context
base_model_matches, rank = calculate_candidate_rank(
confidence, candidate.get("base_model"), base_model_context
)
if base_model_context and not base_model_matches:
continue
rank = confidence + _base_model_score(
candidate.get("base_model"), base_model_context
)
if rank > best_rank:
best_rank = rank
best_confidence = confidence
core/sources/model_list.py
+8 -19
View File
@@ -55,16 +55,10 @@ def _fetch_json_url(url: str) -> Dict[str, Any]:
def _read_json_file(path: str, default: Any) -> Any:
try:
if os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
log_warn(f"Error reading {os.path.basename(path)}: {e}")
return default
return read_json_safe(path, default)
from ..path_utils import write_json_atomic
from ..path_utils import write_json_atomic, read_json_safe
def _write_json_file_atomic(path: str, data: Any):
@@ -114,17 +108,12 @@ def _load_model_list() -> List[Dict]:
if _model_list_cache is not None:
return _model_list_cache
try:
if os.path.exists(MODEL_LIST_FILE):
with open(MODEL_LIST_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
_model_list_cache = data.get("models", [])
log_info(
f"Loaded {len(_model_list_cache)} models from model-list.json"
)
return _model_list_cache
except Exception as e:
log_error(f"Error loading model list: {e}")
data = read_json_safe(MODEL_LIST_FILE, {})
_model_list_cache = data.get("models", [])
log_info(
f"Loaded {len(_model_list_cache)} models from model-list.json"
)
return _model_list_cache
_model_list_cache = []
return _model_list_cache
core/sources/popular.py
+10 -40
View File
@@ -20,7 +20,7 @@ from ..log_system.log_funcs import (
log_exception,
)
from ..path_utils import METADATA_DIR
from ..path_utils import METADATA_DIR, read_json_safe
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")
@@ -38,16 +38,8 @@ def _load_popular_models() -> Dict[str, Any]:
if _popular_models_cache is not None:
return _popular_models_cache
try:
if os.path.exists(POPULAR_MODELS_FILE):
with open(POPULAR_MODELS_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
_popular_models_cache = data.get("models", {})
return _popular_models_cache
except Exception as e:
log_error(f"Error loading popular models: {e}")
_popular_models_cache = {}
data = read_json_safe(POPULAR_MODELS_FILE, {})
_popular_models_cache = data.get("models", {})
return _popular_models_cache
@@ -58,16 +50,8 @@ def _load_model_aliases() -> Dict[str, List[str]]:
if _model_aliases_cache is not None:
return _model_aliases_cache
try:
if os.path.exists(MODEL_ALIASES_FILE):
with open(MODEL_ALIASES_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
_model_aliases_cache = data.get("aliases", {})
return _model_aliases_cache
except Exception as e:
log_error(f"Error loading model aliases: {e}")
_model_aliases_cache = {}
data = read_json_safe(MODEL_ALIASES_FILE, {})
_model_aliases_cache = data.get("aliases", {})
return _model_aliases_cache
@@ -83,11 +67,9 @@ def load_base_model_aliases() -> Dict[str, List[str]]:
aliases_dict = {}
try:
if os.path.exists(BASE_MODELS_FILE):
with open(BASE_MODELS_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
base_models = data.get("base_models", [])
for model in base_models:
data = read_json_safe(BASE_MODELS_FILE, {})
base_models = data.get("base_models", [])
for model in base_models:
name = model.get("name", "")
normalized_name = _normalize_base_model(name)
if not normalized_name:
@@ -220,23 +202,11 @@ def get_base_models_config() -> Dict[str, Any]:
def _read_base_models_file() -> Dict[str, Any]:
try:
if os.path.exists(BASE_MODELS_FILE):
with open(BASE_MODELS_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
log_warn(f"Error reading base-models.json: {e}")
return {"base_models": []}
return read_json_safe(BASE_MODELS_FILE, {"base_models": []})
def _read_base_models_meta() -> Dict[str, Any]:
try:
if os.path.exists(BASE_MODELS_META_FILE):
with open(BASE_MODELS_META_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
log_warn(f"Error reading base-models.meta.json: {e}")
return {}
return read_json_safe(BASE_MODELS_META_FILE, {})
def generate_aliases(name: str) -> List[str]:
core/type_utils.py
+18
View File
@@ -85,3 +85,21 @@ def to_int(value: Any, default: Any = None) -> Any:
except (TypeError, ValueError):
return default
def to_bool(value: Any, default: bool = False) -> bool:
"""
Safely cast a value to a boolean, converting string "true", "yes", "1" etc.
"""
if value is None:
return default
if isinstance(value, bool):
return value
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
return bool(value)