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

refactor: centralize API error handling and confidence calculation logic

Browse Source
This commit is contained in:
Dariusz L
2026-06-25 02:24:02 +02:00
parent e4c3b0fb6f
commit 92e2601631
5 changed files with 1132 additions and 1320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+989 -1098
View File
File diff suppressed because it is too large Load Diff
core/matcher.py
+122 -23
View File
@@ -195,30 +195,11 @@ def find_matches(
candidate["_match_filename_norm"] = candidate_norm
if target_norm == candidate_norm:
# Exact match after normalization = 100% confidence
similarity = 1.0
confidence = 100.0
else:
# Calculate similarity using the configured fuzzy matcher
# This gives a ratio between 0.0 and 1.0 based on longest common subsequence
similarity = calculate_similarity_with_normalization(
target_filename, candidate_filename
)
# Also try comparing without extensions for better matching
candidate_base = os.path.splitext(candidate_filename)[0]
similarity_no_ext = calculate_similarity_with_normalization(
target_base, candidate_base
)
# Use the higher of the two similarity scores
# But ensure we never get 1.0 unless it's an exact normalized match
similarity = max(similarity, similarity_no_ext)
# Cap similarity at 0.999 for non-exact matches to prevent false 100% scores
# Fuzzy matchers can sometimes give 1.0 for very similar but not identical strings
# due to normalization artifacts
if similarity >= 0.999 and target_norm != candidate_norm:
similarity = 0.999
confidence = calculate_filename_confidence(target_filename, candidate_filename)
similarity = confidence / 100.0
# Only include if above threshold
if similarity >= threshold:
@@ -226,7 +207,7 @@ def find_matches(
"model": candidate,
"filename": candidate_filename,
"similarity": similarity,
"confidence": round(similarity * 100, 1), # Convert to percentage
"confidence": confidence,
}
if max_results <= 0:
return []
@@ -299,3 +280,121 @@ def base_model_score(candidate: str, preferred: Optional[str]) -> float:
return 0.0
return 1000.0 if base_model_matches(candidate, preferred) else -1000.0
# ==================== CENTRALIZED MATCHING & CONFIDENCE HELPERS ====================
MODEL_FILE_EXTENSIONS = {
".ckpt",
".pt",
".pt2",
".bin",
".pth",
".safetensors",
".pkl",
".sft",
".onnx",
".gguf",
}
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_FILE_EXTENSIONS:
if lowered.endswith(ext):
return filename[: -len(ext)]
return filename
def has_known_model_extension(filename: str) -> bool:
"""Check if a filename ends with a known model extension."""
return strip_known_model_extension(filename) != filename
def normalize_model_title(value: str) -> str:
"""Normalize model title for comparison."""
value = strip_known_model_extension(str(value or "")).lower()
value = re.sub(r"[^a-z0-9]+", " ", value)
return re.sub(r"\s+", " ", value).strip()
def calculate_model_title_confidence(query: str, model_name: str) -> float:
"""Calculate confidence score (0.0 to 100.0) for a model title matching a query."""
query_norm = normalize_model_title(query)
model_norm = normalize_model_title(model_name)
if not query_norm or not model_norm:
return 0.0
if query_norm == model_norm:
return 100.0
return round(
max(
calculate_similarity_with_normalization(query_norm, model_norm),
calculate_similarity_with_normalization(
query_norm.replace(" ", ""), model_norm.replace(" ", "")
),
)
* 100,
1,
)
def calculate_filename_confidence(target_filename: str, candidate_filename: str) -> float:
"""Calculate similarity confidence percentage (0.0 to 100.0) between two filenames."""
target_norm = normalize_filename(target_filename)
candidate_norm = normalize_filename(candidate_filename)
if target_norm == candidate_norm:
return 100.0
similarity = calculate_similarity_with_normalization(
target_filename, candidate_filename
)
similarity_no_ext = calculate_similarity_with_normalization(
os.path.splitext(target_filename)[0], os.path.splitext(candidate_filename)[0]
)
best_similarity = max(similarity, similarity_no_ext)
# Cap similarity at 0.999 for non-exact matches to prevent false 100% scores
if best_similarity >= 0.999:
best_similarity = 0.999
return round(best_similarity * 100, 1)
def calculate_archived_model_confidence(
query: str,
model_name: str = "",
version_name: str = "",
filename: str = "",
) -> float:
"""Calculate confidence score (0.0 to 100.0) for a model in an archive matching a query."""
candidates = [value for value in [filename, model_name, version_name] if value]
if not candidates:
return 0.0
query_norm = normalize_filename(query)
best = 0.0
for candidate in candidates:
candidate_norm = normalize_filename(candidate)
if query_norm == candidate_norm:
return 100.0
similarity = calculate_similarity_with_normalization(query, candidate)
similarity_no_ext = calculate_similarity_with_normalization(
os.path.splitext(query)[0],
os.path.splitext(candidate)[0],
)
score = max(similarity, similarity_no_ext)
if query_norm and candidate_norm and (
query_norm in candidate_norm or candidate_norm in query_norm
):
score = max(score, 0.85)
best = max(best, score)
return round(best * 100, 1)
core/sources/civarchive.py
+9 -92
View File
@@ -15,8 +15,8 @@ from urllib.parse import parse_qs, urljoin, urlparse
import requests
from ..matcher import (
calculate_similarity_with_normalization,
normalize_filename,
calculate_archived_model_confidence,
calculate_model_title_confidence,
normalize_base_model as _normalize_base_model,
base_model_matches as _base_model_matches,
base_model_score as _base_model_score,
@@ -40,18 +40,7 @@ DEFAULT_CIVARCHIVE_CANDIDATE_LIMIT = 10
MAX_CIVARCHIVE_CANDIDATE_LIMIT = 30
MODEL_TITLE_MATCH_THRESHOLD = 82.0
MODEL_FILE_EXTENSIONS = {
".ckpt",
".pt",
".pt2",
".bin",
".pth",
".safetensors",
".pkl",
".sft",
".onnx",
".gguf",
}
_search_cache: Dict[str, Any] = {}
@@ -925,79 +914,7 @@ def get_civarchive_model_details(
}
def _calculate_confidence(
query: str,
model_name: str = "",
version_name: str = "",
filename: str = "",
) -> float:
candidates = [value for value in [filename, model_name, version_name] if value]
if not candidates:
return 0.0
query_norm = normalize_filename(query)
best = 0.0
for candidate in candidates:
candidate_norm = normalize_filename(candidate)
if query_norm == candidate_norm:
return 100.0
similarity = calculate_similarity_with_normalization(query, candidate)
similarity_no_ext = calculate_similarity_with_normalization(
os.path.splitext(query)[0],
os.path.splitext(candidate)[0],
)
score = max(similarity, similarity_no_ext)
if query_norm and candidate_norm and (
query_norm in candidate_norm or candidate_norm in query_norm
):
score = max(score, 0.85)
best = max(best, score)
return round(best * 100, 1)
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_FILE_EXTENSIONS:
if lowered.endswith(ext):
return filename[: -len(ext)]
return filename
def _has_known_model_extension(filename: str) -> bool:
return _strip_known_model_extension(filename) != filename
def _normalize_model_title(value: str) -> str:
value = _strip_known_model_extension(str(value or "")).lower()
value = re.sub(r"[^a-z0-9]+", " ", value)
return re.sub(r"\s+", " ", value).strip()
def _calculate_model_title_confidence(query: str, model_name: str) -> float:
query_norm = _normalize_model_title(query)
model_norm = _normalize_model_title(model_name)
if not query_norm or not model_norm:
return 0.0
if query_norm == model_norm:
return 100.0
return round(
max(
calculate_similarity_with_normalization(query_norm, model_norm),
calculate_similarity_with_normalization(
query_norm.replace(" ", ""), model_norm.replace(" ", "")
),
)
* 100,
1,
)
def _version_sort_key(version: Dict[str, Any]) -> tuple:
@@ -1147,7 +1064,7 @@ def _find_model_title_match_in_model_details(
) -> Optional[Dict[str, Any]]:
"""For extensionless workflow values, resolve by CivArchive model page title."""
model_name = model_details.get("name", "")
title_confidence = _calculate_model_title_confidence(title_query, model_name)
title_confidence = calculate_model_title_confidence(title_query, model_name)
if title_confidence < MODEL_TITLE_MATCH_THRESHOLD:
log_debug(
f"CivArchive title candidate rejected: model_id={model_id}, query={title_query}, model_name={model_name}, confidence={title_confidence}"
@@ -1245,7 +1162,7 @@ def _build_result_from_payload(
for file_info in files:
file_name = file_info.get("name") or ""
confidence = _calculate_confidence(
confidence = calculate_archived_model_confidence(
query_value,
context.get("name", ""),
version.get("name", ""),
@@ -1257,7 +1174,7 @@ def _build_result_from_payload(
if selected_file is None:
selected_file = _select_primary_file(files)
best_confidence = _calculate_confidence(
best_confidence = calculate_archived_model_confidence(
query_value,
context.get("name", ""),
version.get("name", ""),
@@ -1434,7 +1351,7 @@ def _resolve_search_candidate(
result = None
if result and not result.get("confidence"):
result["confidence"] = _calculate_confidence(
result["confidence"] = calculate_archived_model_confidence(
query,
candidate.get("name", ""),
"",
@@ -1624,13 +1541,13 @@ def search_civarchive_for_file(
confidence = float(
resolved.get("title_confidence")
or resolved.get("confidence")
or _calculate_model_title_confidence(
or calculate_model_title_confidence(
normalized_filename,
resolved.get("name", ""),
)
)
else:
confidence = _calculate_confidence(
confidence = calculate_archived_model_confidence(
normalized_filename,
resolved.get("name", ""),
resolved.get("version_name", ""),
core/sources/civitai.py
+8 -76
View File
@@ -13,8 +13,8 @@ from typing import Dict, Any, Optional, List, Callable
from urllib.parse import urlparse, parse_qs, quote
from ..matcher import (
calculate_similarity_with_normalization,
normalize_filename,
calculate_filename_confidence,
calculate_model_title_confidence,
normalize_base_model as _normalize_base_model,
base_model_matches as _base_model_matches,
base_model_score as _base_model_score,
@@ -38,18 +38,7 @@ DEFAULT_CIVITAI_CANDIDATE_LIMIT = 5
MAX_CIVITAI_CANDIDATE_LIMIT = 20
MODEL_TITLE_MATCH_THRESHOLD = 82.0
MODEL_FILE_EXTENSIONS = {
".ckpt",
".pt",
".pt2",
".bin",
".pth",
".safetensors",
".pkl",
".sft",
".onnx",
".gguf",
}
CIVITAI_TYPE_MAP = {
"checkpoint": "Checkpoint",
@@ -286,64 +275,7 @@ def _build_civitai_result_from_version(
}
def _calculate_filename_confidence(target_filename: str, candidate_filename: str) -> float:
"""Calculate filename confidence using the same normalized approach as local matching."""
target_norm = normalize_filename(target_filename)
candidate_norm = normalize_filename(candidate_filename)
if target_norm == candidate_norm:
return 100.0
similarity = calculate_similarity_with_normalization(
target_filename, candidate_filename
)
similarity_no_ext = calculate_similarity_with_normalization(
os.path.splitext(target_filename)[0], os.path.splitext(candidate_filename)[0]
)
return round(max(similarity, similarity_no_ext) * 100, 1)
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_FILE_EXTENSIONS:
if lowered.endswith(ext):
return filename[: -len(ext)]
return filename
def _has_known_model_extension(filename: str) -> bool:
return _strip_known_model_extension(filename) != filename
def _normalize_model_title(value: str) -> str:
value = _strip_known_model_extension(str(value or "")).lower()
value = re.sub(r"[^a-z0-9]+", " ", value)
return re.sub(r"\s+", " ", value).strip()
def _calculate_model_title_confidence(query: str, model_name: str) -> float:
query_norm = _normalize_model_title(query)
model_norm = _normalize_model_title(model_name)
if not query_norm or not model_norm:
return 0.0
if query_norm == model_norm:
return 100.0
return round(
max(
calculate_similarity_with_normalization(query_norm, model_norm),
calculate_similarity_with_normalization(
query_norm.replace(" ", ""), model_norm.replace(" ", "")
),
)
* 100,
1,
)
def _version_sort_key(version: Dict[str, Any]) -> tuple:
@@ -389,7 +321,7 @@ def _find_model_title_match_in_model(
) -> Optional[Dict[str, Any]]:
"""For extensionless workflow values, resolve by CivitAI model page title."""
model_name = model_data.get("name", "")
title_confidence = _calculate_model_title_confidence(title_query, model_name)
title_confidence = calculate_model_title_confidence(title_query, model_name)
if title_confidence < MODEL_TITLE_MATCH_THRESHOLD:
log_debug(
f"CivitAI title candidate rejected: model_id={model_id}, query={title_query}, model_name={model_name}, confidence={title_confidence}"
@@ -483,7 +415,7 @@ def _find_matching_file_in_versions(
}
if not exact_only:
confidence = _calculate_filename_confidence(filename, file_name)
confidence = calculate_filename_confidence(filename, file_name)
if confidence > best_confidence:
best_confidence = confidence
best_match = {
@@ -722,7 +654,7 @@ def _find_civitai_file_in_model(
"base_model": resolved.get("base_model"),
"tags": resolved.get("tags", []),
"match_type": "exact",
"confidence": _calculate_filename_confidence(
"confidence": calculate_filename_confidence(
filename, expected_filename
),
"sha256": sha256,
@@ -802,7 +734,7 @@ def _find_civitai_file_in_model(
if not exact_only:
expected_filename = resolved.get("expected_filename", "")
confidence = _calculate_filename_confidence(filename, expected_filename)
confidence = calculate_filename_confidence(filename, expected_filename)
ranking_score = confidence + _base_model_score(
resolved.get("base_model"), base_model_context
)
@@ -844,7 +776,7 @@ def _find_civitai_file_in_model(
)
result["confidence"] = match.get(
"confidence",
_calculate_filename_confidence(filename, result.get("filename", "")),
calculate_filename_confidence(filename, result.get("filename", "")),
)
if match["match_type"] == "similar":
log_info(
core/sources/lora_manager_archive.py
+4 -31
View File
@@ -12,8 +12,7 @@ import time
from typing import Any, Callable, Dict, List, Optional
from ..matcher import (
calculate_similarity_with_normalization,
normalize_filename,
calculate_archived_model_confidence,
normalize_base_model as _normalize_base_model,
base_model_matches as _base_model_matches,
base_model_score as _base_model_score,
@@ -148,33 +147,7 @@ def _normalize_model_type(model_type: Optional[str]) -> str:
return type_map.get(value, value)
def _calculate_confidence(
query: str, model_name: str, version_name: str = "", filename: str = ""
) -> float:
candidates = [value for value in [model_name, version_name, filename] if value]
if not candidates:
return 0.0
query_norm = normalize_filename(query)
best = 0.0
for candidate in candidates:
candidate_norm = normalize_filename(candidate)
if query_norm == candidate_norm:
return 100.0
similarity = calculate_similarity_with_normalization(query, candidate)
similarity_no_ext = calculate_similarity_with_normalization(
os.path.splitext(query)[0], os.path.splitext(candidate)[0]
)
candidate_score = max(similarity, similarity_no_ext)
if query_norm in candidate_norm or candidate_norm in query_norm:
candidate_score = max(candidate_score, 0.85)
best = max(best, candidate_score)
return round(best * 100, 1)
@@ -501,7 +474,7 @@ def _build_result_from_row(
elif not isinstance(trained_words, list):
trained_words = []
confidence = _calculate_confidence(
confidence = calculate_archived_model_confidence(
query,
row["model_name"] or "",
row["version_name"] or "",
@@ -635,7 +608,7 @@ def search_lora_manager_archive(
candidate_index=row_index,
candidate_count=row_count,
)
confidence = _calculate_confidence(
confidence = calculate_archived_model_confidence(
normalized_query,
row["model_name"] or "",
row["version_name"] or "",
@@ -789,7 +762,7 @@ def search_lora_manager_archive_for_file(
candidate_count=total_candidates,
)
candidate_filename = candidate.get("filename", "")
confidence = _calculate_confidence(filename, candidate.get("name", ""), candidate.get("version_name", ""), candidate_filename)
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(