refactor: improve code quality and consistency across backend modules

Author: Azornes

__init__.py

@@ -637,7 +637,8 @@ async def analyze_workflow(request):
                             total=0,
                         )
 
-                    update_analysis_progress = lambda payload: self._update_analysis_progress(analysis_id, payload)
+                    def update_analysis_progress(payload):
+                        self._update_analysis_progress(analysis_id, payload)
 
                     # Analyze and find matches
                     result = await asyncio.to_thread(
@@ -1483,7 +1484,10 @@ async def metadata_build_start_route(request):
                     requested_worker_count=worker_count,
                 )
 
-                update_metadata_build_progress = lambda progress_payload: self._update_metadata_build_progress(progress_id, progress_payload)
+                def update_metadata_build_progress(progress_payload):
+                    self._update_metadata_build_progress(
+                        progress_id, progress_payload
+                    )
 
                 def is_metadata_build_cancelled():
                     return self.metadata_builder_progress.is_cancelled(progress_id)
@@ -1602,7 +1606,8 @@ async def get_loaded_models(request):
                         {"error": "Workflow JSON must be an object"}, status=400
                     )
 
-                update_loaded_progress = lambda *args, **kwargs: self._update_loaded_progress(loaded_id, *args, **kwargs)
+                def update_loaded_progress(*args, **kwargs):
+                    self._update_loaded_progress(loaded_id, *args, **kwargs)
 
                 def get_workflow_node_count():
                     node_count = 0
@@ -1740,7 +1745,13 @@ def build_loaded_models_response():
                         total=workflow_node_count,
                     )
 
-                    update_workflow_analysis_progress = lambda payload: self._update_workflow_analysis_progress(loaded_id, workflow_node_count, interpolate_percent, payload)
+                    def update_workflow_analysis_progress(payload):
+                        self._update_workflow_analysis_progress(
+                            loaded_id,
+                            workflow_node_count,
+                            interpolate_percent,
+                            payload,
+                        )
 
                     # Analyze workflow to get all model references
                     all_model_refs = analyze_workflow_models(
@@ -2170,9 +2181,7 @@ def result_hash_matches(result, *, require_filename=False):
                     result_hash = extract_result_sha256(result)
                     if result_hash != provided_hash:
                         return False
-                    if require_filename and not result_filename_matches(result):
-                        return False
-                    return True
+                    return not require_filename or result_filename_matches(result)
 
                 def huggingface_page_url(result):
                     try:

core/downloader.py

@@ -2213,7 +2213,7 @@ def download_file(
         try:
             if os.path.exists(dest_path):
                 os.remove(dest_path)
-        except:
+        except Exception:
             pass
 
     except Exception as e:

core/log_system/logger.py

@@ -298,7 +298,7 @@ class AzLogsLogger:
 
     def __new__(cls):
         if cls._instance is None:
-            cls._instance = super(AzLogsLogger, cls).__new__(cls)
+            cls._instance = super().__new__(cls)
             cls._instance._initialized = False
         return cls._instance
 

core/matcher.py

@@ -128,17 +128,15 @@ def find_matches(
 
     # Normalize target filename once for exact match comparisons
     target_norm = normalize_filename(target_filename)
-    target_base = os.path.splitext(target_filename)[0]
 
     for candidate in candidate_models:
         # Get filename from candidate (prefer 'filename' key, fallback to extracting from 'path' or 'relative_path')
         candidate_filename = candidate.get("filename")
         candidate_path = candidate.get("path", "") or candidate.get("relative_path", "")
 
         # If no filename key, try to extract from path or relative_path
-        if not candidate_filename:
-            if candidate_path:
-                candidate_filename = get_filename_from_path(candidate_path)
+        if not candidate_filename and candidate_path:
+            candidate_filename = get_filename_from_path(candidate_path)
 
         if not candidate_filename:
             continue
@@ -167,10 +165,13 @@ def find_matches(
         if candidate_path_normalized and target_model_normalized:
             if candidate_path_normalized == target_model_normalized:
                 path_match = True
-        elif candidate_relative_path_normalized and target_model_normalized:
+        elif (
+            candidate_relative_path_normalized
+            and target_model_normalized
+            and candidate_relative_path_normalized == target_model_normalized
+        ):
             # Also check if relative path matches the target (which might be relative)
-            if candidate_relative_path_normalized == target_model_normalized:
-                path_match = True
+            path_match = True
 
         if path_match:
             # Exact path match after normalization = 100% confidence

core/metadata_builder.py

@@ -191,7 +191,7 @@ def extract_local_header_snapshot(model_path: str) -> Dict[str, Any]:
     if not isinstance(metadata, dict) or not metadata:
         return {}
 
-    keys = sorted(str(key) for key in metadata.keys())
+    keys = sorted(str(key) for key in metadata)
     limited_metadata = {}
     for key in keys[:LOCAL_HEADER_MAX_KEYS]:
         limited_metadata[key] = _limited_local_value(metadata.get(key))

core/path_utils.py

@@ -503,7 +503,7 @@ def infer_safetensors_base_model(
 
     tensor_keys = [
         str(key)
-        for key in header_json.keys()
+        for key in header_json
         if key != "__metadata__"
     ][:max_tensor_keys]
     keys_text = " ".join(tensor_keys)
@@ -768,7 +768,7 @@ def extract_safetensors_header_metadata(file_path: str) -> Dict[str, Any]:
         metadata = {}
 
     result: Dict[str, Any] = {}
-    metadata_keys = sorted(str(key) for key in metadata.keys())
+    metadata_keys = sorted(str(key) for key in metadata)
     if metadata_keys:
         result["header_metadata_keys"] = metadata_keys[:SAFETENSORS_METADATA_KEY_LIMIT]
 

core/scanner.py

@@ -136,22 +136,25 @@ def scan_directory(
 
             dirs[:] = filtered_dirs
 
-            if "folder" in (extensions or set()) and root != base_directory:
-                if category != "diffusers" or "model_index.json" in files:
-                    try:
-                        relative_path = os.path.relpath(root, base_directory)
-                    except ValueError:
-                        relative_path = get_filename_from_path(root)
-
-                    models.append(
-                        {
-                            "filename": get_filename_from_path(root),
-                            "path": root,
-                            "relative_path": relative_path,
-                            "category": category,
-                            "base_directory": base_directory,
-                        }
-                    )
+            if (
+                "folder" in (extensions or set())
+                and root != base_directory
+                and (category != "diffusers" or "model_index.json" in files)
+            ):
+                try:
+                    relative_path = os.path.relpath(root, base_directory)
+                except ValueError:
+                    relative_path = get_filename_from_path(root)
+
+                models.append(
+                    {
+                        "filename": get_filename_from_path(root),
+                        "path": root,
+                        "relative_path": relative_path,
+                        "category": category,
+                        "base_directory": base_directory,
+                    }
+                )
 
             for filename in files:
                 # Check if file has a model extension

core/sources/civarchive.py

@@ -2152,7 +2152,9 @@ def search_civarchive_for_file(
             scored_candidates.sort(key=lambda x: x[0], reverse=True)
 
             candidate_count = len(scored_candidates)
-            for candidate_index, (prelim_confidence, candidate) in enumerate(scored_candidates, start=1):
+            for candidate_index, (_prelim_confidence, candidate) in enumerate(
+                scored_candidates, start=1
+            ):
                 if len(seen) >= detail_limit:
                     break
                 identity = _candidate_identity(candidate)

core/sources/civitai.py

@@ -10,7 +10,7 @@
 from typing import Any, Callable, Dict, List, Optional
 from urllib.parse import quote, urlencode
 
-import requests  # noqa: F401 – used indirectly via unittest.mock patch targets
+import requests  # noqa: F401 - used indirectly via unittest.mock patch targets
 
 from ..log_system import create_module_logger
 from ..matcher import (
@@ -667,17 +667,18 @@ def resolved_version_matches(resolved: Dict[str, Any]) -> bool:
         )
         if expected_filename == filename_lower:
             return True
-        if not exact_only and _filename_base_partial_match(filename_base, expected_base):
-            return True
-        return False
+        return not exact_only and _filename_base_partial_match(
+            filename_base, expected_base
+        )
 
     if preferred_version_id is not None:
         resolved = resolve_urn(model_id, preferred_version_id, api_key)
-        if resolved:
-            if resolved_version_matches(resolved) and _base_model_matches(
-                resolved.get("base_model"), base_model_context
-            ):
-                return build_result_from_resolved_version(resolved, preferred_version_id)
+        if (
+            resolved
+            and resolved_version_matches(resolved)
+            and _base_model_matches(resolved.get("base_model"), base_model_context)
+        ):
+            return build_result_from_resolved_version(resolved, preferred_version_id)
 
     best_resolved_result = None
     best_resolved_confidence = 0.0
@@ -1605,8 +1606,7 @@ def _apply_safetensors_header_metadata(
     current_name = str(result.get("model_name") or "").strip()
     if header_name and (
         not current_name
-        or current_name == stem
-        or current_name == filename
+        or current_name in (stem, filename)
         or result.get("source") == "local"
     ):
         result["model_name"] = header_name

core/sources/huggingface.py

@@ -446,17 +446,19 @@ def _find_matching_file_in_repo(
         file_name_lower = file_name.lower()
         file_base = os.path.splitext(file_name_lower)[0]
 
-        if not exact_only:
-            if filename_base in file_base or file_base in filename_base:
-                if file_path.endswith(".safetensors") or file_path.endswith(".ckpt"):
-                    partial_match = _build_huggingface_result(
-                        repo_id,
-                        file_path,
-                        file_info,
-                        "partial",
-                        headers=headers,
-                    )
-                    break
+        if (
+            not exact_only
+            and (filename_base in file_base or file_base in filename_base)
+            and (file_path.endswith(".safetensors") or file_path.endswith(".ckpt"))
+        ):
+            partial_match = _build_huggingface_result(
+                repo_id,
+                file_path,
+                file_info,
+                "partial",
+                headers=headers,
+            )
+            break
 
     return partial_match
 
@@ -505,18 +507,17 @@ def _find_matching_file_in_author_index(
         file_base = os.path.splitext(file_name_lower)[0]
         file_path_lower = file_path.lower()
 
-        if filename_base in file_base or file_base in filename_base:
-            if file_path_lower.endswith(".safetensors") or file_path_lower.endswith(
-                ".ckpt"
-            ):
-                partial_match = _build_huggingface_result(
-                    repo_id,
-                    file_path,
-                    file_info,
-                    "partial",
-                    headers=headers,
-                )
-                break
+        if (
+            filename_base in file_base or file_base in filename_base
+        ) and file_path_lower.endswith((".safetensors", ".ckpt")):
+            partial_match = _build_huggingface_result(
+                repo_id,
+                file_path,
+                file_info,
+                "partial",
+                headers=headers,
+            )
+            break
 
     return partial_match