From 6850ef1267a677ada64a4486b11bb5c4d8282c41 Mon Sep 17 00:00:00 2001 From: Dariusz L Date: Thu, 25 Jun 2026 17:33:59 +0200 Subject: [PATCH] refactor: centralize type coercion and file I/O helpers to eliminate duplication --- __init__.py | 72 +++++----------------------- core/downloader.py | 19 ++------ core/matcher.py | 12 +++++ core/path_utils.py | 34 +++++++++++++ core/settings.py | 29 ++++------- core/sources/civarchive.py | 8 ++-- core/sources/civitai.py | 16 +------ core/sources/lora_manager_archive.py | 12 ++--- core/sources/model_list.py | 27 ++++------- core/sources/popular.py | 50 ++++--------------- core/type_utils.py | 18 +++++++ 11 files changed, 115 insertions(+), 182 deletions(-) diff --git a/__init__.py b/__init__.py index 9b7f3f1..27e2bfc 100644 --- a/__init__.py +++ b/__init__.py @@ -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, diff --git a/core/downloader.py b/core/downloader.py index 5f39d84..61081af 100644 --- a/core/downloader.py +++ b/core/downloader.py @@ -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( diff --git a/core/matcher.py b/core/matcher.py index 1717b0d..0d3a657 100644 --- a/core/matcher.py +++ b/core/matcher.py @@ -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 + + diff --git a/core/path_utils.py b/core/path_utils.py index e57ab54..cea0680 100644 --- a/core/path_utils.py +++ b/core/path_utils.py @@ -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 + + diff --git a/core/settings.py b/core/settings.py index e327b7e..ae009be 100644 --- a/core/settings.py +++ b/core/settings.py @@ -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]: diff --git a/core/sources/civarchive.py b/core/sources/civarchive.py index 377f70e..16a229b 100644 --- a/core/sources/civarchive.py +++ b/core/sources/civarchive.py @@ -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: diff --git a/core/sources/civitai.py b/core/sources/civitai.py index 1fbb262..d5f9b90 100644 --- a/core/sources/civitai.py +++ b/core/sources/civitai.py @@ -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( diff --git a/core/sources/lora_manager_archive.py b/core/sources/lora_manager_archive.py index 6cfd3dd..184a66d 100644 --- a/core/sources/lora_manager_archive.py +++ b/core/sources/lora_manager_archive.py @@ -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 diff --git a/core/sources/model_list.py b/core/sources/model_list.py index 85fb70c..6eff6dc 100644 --- a/core/sources/model_list.py +++ b/core/sources/model_list.py @@ -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 diff --git a/core/sources/popular.py b/core/sources/popular.py index 2e7e617..fbf36bc 100644 --- a/core/sources/popular.py +++ b/core/sources/popular.py @@ -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]: diff --git a/core/type_utils.py b/core/type_utils.py index a0bce11..d53d572 100644 --- a/core/type_utils.py +++ b/core/type_utils.py @@ -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) + +