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

refactor: consolidate shared utilities into type_utils and path_utils

Browse Source 
Replace duplicated SHA256 normalization, metadata sidecar discovery, trained-words
extraction, type coercion, and cache helpers into shared modules. Clarify that the
_civitai.py/_civarchive.py/_huggingface.py/_workflow_analyzer.py refactorings are
behavior-preserving so future contributors keep behavior aligned in one place.
This commit is contained in:
Dariusz L
2026-07-02 22:41:56 +02:00
parent 3865914d0f
commit 8ce70c30e6
8 changed files with 245 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+10 -15
View File
@@ -171,7 +171,6 @@ class ModelResolverExtension:
analyze_and_find_matches,
apply_resolution,
get_local_model_hash_metadata,
normalize_sha256,
search_local_matches_by_hash,
search_local_matches,
)
@@ -182,7 +181,15 @@ class ModelResolverExtension:
read_json_safe,
write_json_atomic,
)
from .core.type_utils import first_non_empty, to_int, to_bool, MODEL_EXTENSIONS, format_size_bytes
from .core.type_utils import (
first_non_empty,
to_int,
to_bool,
MODEL_EXTENSIONS,
format_size_bytes,
normalize_sha256,
extract_sha256_from_metadata,
)
from .core.settings import (
TEMPLATE_KEY_ALIASES,
bool_setting as resolver_bool_setting,
@@ -1321,19 +1328,7 @@ class ModelResolverExtension:
"civitai_checked": bool(civitai_checked),
}
def extract_result_sha256(result):
if not isinstance(result, dict):
return ""
for key in ("sha256", "hash"):
normalized = normalize_sha256(result.get(key))
if normalized:
return normalized
hashes = result.get("hashes")
if isinstance(hashes, dict):
for key in ("SHA256", "sha256", "Sha256"):
normalized = normalize_sha256(hashes.get(key))
if normalized:
return normalized
return ""
return extract_sha256_from_metadata(result)
def result_filename_matches(result):
if not isinstance(result, dict):
core/path_utils.py
+31
View File
@@ -224,4 +224,35 @@ def get_comfy_root_path(folder_paths_module: Optional[Any] = None) -> str:
return ""
def find_metadata_sidecar_path(model_path: str) -> str:
"""
Znajduje istniejący plik metadanych (np. .metadata.json, .civitai.info, .json)
powiązany z plikiem modelu i zwraca jego pełną ścieżkę. Zwraca pusty ciąg, jeśli brak.
"""
if not model_path:
return ""
directory = os.path.dirname(model_path)
filename = get_filename_from_path(model_path)
# Base name without extension
base_name = filename.rsplit(".", 1)[0] if "." in filename else filename
# Wzorce nazw na podstawie civitai.py i resolver.py
possible_names = [
base_name + ".metadata.json",
filename + ".metadata.json",
base_name + ".civitai.info",
filename + ".civitai.info",
base_name + ".json",
filename + ".json",
filename.replace("_", " ").split()[0] + ".metadata.json" if "_" in base_name else None,
]
for name in possible_names:
if name:
path = os.path.join(directory, name)
if os.path.exists(path):
return path
return ""
core/resolver.py
+20 -96
View File
@@ -23,7 +23,7 @@ from .workflow_analyzer import (
should_scan_as_model_reference,
)
from .matcher import find_matches, strip_known_model_extension
from .type_utils import as_dict, as_list, MODEL_EXTENSIONS as _MODEL_EXTENSIONS
from .type_utils import as_dict, as_list, MODEL_EXTENSIONS as _MODEL_EXTENSIONS, unique_ordered_strings, extract_sha256_from_metadata, normalize_sha256
from .workflow_updater import update_workflow_nodes
from .sources.civitai import resolve_urn
from .sources.huggingface import parse_huggingface_url as parse_hf_url
@@ -44,6 +44,7 @@ from .path_utils import (
dedupe_local_base_directories,
get_filename_from_path,
read_json_safe,
find_metadata_sidecar_path,
)
@@ -275,63 +276,9 @@ def search_local_matches(
return deduplicated_matches
SHA256_PATTERN = re.compile(r"^[a-fA-F0-9]{64}$")
def normalize_sha256(value: Any) -> str:
"""Return a normalized SHA256 hex string or an empty string."""
if value is None:
return ""
text = str(value).strip()
for prefix in ("sha256:", "sha256="):
if text.lower().startswith(prefix):
text = text[len(prefix):].strip()
return text.lower() if SHA256_PATTERN.match(text) else ""
def _metadata_sidecar_candidates(model_path: str) -> List[str]:
if not model_path:
return []
directory = os.path.dirname(model_path)
filename = get_filename_from_path(model_path)
stem, _ = os.path.splitext(filename)
candidates = [
os.path.join(directory, f"{stem}.metadata.json"),
f"{model_path}.metadata.json",
os.path.join(directory, f"{filename}.metadata.json"),
]
seen = set()
unique = []
for candidate in candidates:
key = os.path.normcase(os.path.abspath(candidate))
if key in seen:
continue
seen.add(key)
unique.append(candidate)
return unique
def _collect_hashes_from_container(value: Any) -> List[str]:
container = as_dict(value)
hashes = as_dict(container.get("hashes"))
return [
normalize_sha256(candidate)
for candidate in (
container.get("sha256"),
container.get("hash"),
container.get("SHA256"),
hashes.get("SHA256"),
hashes.get("sha256"),
hashes.get("hash"),
)
if normalize_sha256(candidate)
]
h = extract_sha256_from_metadata(value)
return [h] if h else []
def _metadata_file_matches_model(file_info: Dict[str, Any], model: Dict[str, Any]) -> bool:
@@ -384,20 +331,9 @@ def _extract_model_sha256_from_metadata(
if isinstance(file_info, dict) and _metadata_file_matches_model(file_info, model):
values.extend(_collect_hashes_from_container(file_info))
return [value for value in _unique_ordered_strings(values) if normalize_sha256(value)]
return [value for value in unique_ordered_strings(values) if normalize_sha256(value)]
def _unique_ordered_strings(values: List[Any]) -> List[str]:
seen = set()
unique = []
for value in values:
text = str(value or "").strip()
if not text or text in seen:
continue
seen.add(text)
unique.append(text)
return unique
def search_local_matches_by_hash(
sha256: str,
@@ -432,10 +368,7 @@ def search_local_matches_by_hash(
if not model_path:
continue
metadata_path = next(
(candidate for candidate in _metadata_sidecar_candidates(model_path) if os.path.exists(candidate)),
"",
)
metadata_path = find_metadata_sidecar_path(model_path)
if not metadata_path:
continue
@@ -495,34 +428,25 @@ def get_local_model_hash_metadata(
model_info.setdefault("filename", get_filename_from_path(normalized_path))
model_info.setdefault("relative_path", model_info.get("filename", ""))
first_metadata_path = ""
metadata_path = find_metadata_sidecar_path(normalized_path)
last_hash_status = ""
for metadata_path in _metadata_sidecar_candidates(normalized_path):
if not os.path.exists(metadata_path):
continue
if not first_metadata_path:
first_metadata_path = metadata_path
if metadata_path:
metadata = read_json_safe(metadata_path, None)
if not isinstance(metadata, dict) or not metadata:
log.debug(f"Could not read metadata sidecar for local hash lookup: {metadata_path}")
continue
last_hash_status = str(metadata.get("hash_status") or "").strip()
hashes = _extract_model_sha256_from_metadata(metadata, model_info)
if hashes:
return {
"exists": exists,
"metadata_path": metadata_path,
"hash_status": last_hash_status,
"hashes": hashes,
"sha256": hashes[0],
}
if isinstance(metadata, dict) and metadata:
last_hash_status = str(metadata.get("hash_status") or "").strip()
hashes = _extract_model_sha256_from_metadata(metadata, model_info)
if hashes:
return {
"exists": exists,
"metadata_path": metadata_path,
"hash_status": last_hash_status,
"hashes": hashes,
"sha256": hashes[0],
}
return {
"exists": exists,
"metadata_path": first_metadata_path,
"metadata_path": metadata_path,
"hash_status": last_hash_status,
"hashes": [],
"sha256": "",
core/sources/civarchive.py
+7 -19
View File
@@ -32,9 +32,11 @@ from ..type_utils import (
get_version_sort_key,
DEFAULT_BROWSER_USER_AGENT,
parse_civitai_model_path,
fetch_remote_file_size,
looks_like_model_file,
normalize_model_image,
fetch_remote_file_size_cached,
clear_remote_size_cache,
extract_trained_words,
)
from ..progress import report_progress, get_progress_reporter
from ..log_system import create_module_logger
@@ -59,7 +61,6 @@ MODEL_TITLE_MATCH_THRESHOLD = 82.0
_search_cache: Dict[str, Any] = {}
_download_size_cache: Dict[str, Optional[int]] = {}
REQUEST_HEADERS = {
"accept": "application/json,text/html;q=0.9,*/*;q=0.8",
@@ -76,7 +77,7 @@ class CivArchiveSearchError(Exception):
def clear_search_cache():
"""Clear cached CivArchive search results."""
_search_cache.clear()
_download_size_cache.clear()
clear_remote_size_cache()
_report_progress = get_progress_reporter("CivArchive progress callback")
@@ -618,12 +619,7 @@ def _fetch_remote_file_size_bytes(url: Any, timeout: int = 15) -> Optional[int]:
probe_url = _prepare_size_probe_url(url)
if not probe_url:
return None
if probe_url in _download_size_cache:
return _download_size_cache[probe_url]
size = fetch_remote_file_size(probe_url, headers=REQUEST_HEADERS, timeout=timeout)
_download_size_cache[probe_url] = size
return size
return fetch_remote_file_size_cached(probe_url, headers=REQUEST_HEADERS, timeout=timeout)
@@ -978,11 +974,7 @@ def _normalize_archive_version(
for image in raw_images
if isinstance(image, dict) and (image.get("url") or image.get("imageUrl") or image.get("src"))
]
trained_words = version.get("trainedWords") or version.get("trigger") or []
if isinstance(trained_words, str):
trained_words = [trained_words]
elif not isinstance(trained_words, list):
trained_words = []
trained_words = extract_trained_words(version.get("trainedWords"), version.get("trigger"))
return {
"id": version_id,
"name": version.get("name") or "",
@@ -1699,11 +1691,7 @@ def _build_result_from_payload(
if not isinstance(tags, list):
tags = [tags] if tags else []
trained_words = version.get("trainedWords") or version.get("trigger") or []
if isinstance(trained_words, str):
trained_words = [trained_words]
elif not isinstance(trained_words, list):
trained_words = []
trained_words = extract_trained_words(version.get("trainedWords"), version.get("trigger"))
civarchive_url = (
f"{CIVARCHIVE_BASE_URL}/models/{model_id}?modelVersionId={version_id}"
core/sources/civitai.py
+27 -132
View File
@@ -30,9 +30,12 @@ from ..type_utils import (
DEFAULT_BROWSER_USER_AGENT,
parse_civitai_model_path,
normalize_model_image,
as_dict,
as_list,
extract_trained_words,
)
from ..progress import report_progress, get_progress_reporter
from ..path_utils import calculate_file_sha256, get_filename_from_path, read_json_safe
from ..path_utils import calculate_file_sha256, get_filename_from_path, read_json_safe, find_metadata_sidecar_path
from ..log_system import create_module_logger
log = create_module_logger(__name__)
@@ -1405,31 +1408,6 @@ def get_model_info_by_hash(
return None
def _extract_trained_words(version_info: Dict[str, Any]) -> List[str]:
"""
Extract trained words/phrases from model version info.
"""
trained_words = []
# Try to get from metadata
metadata = version_info.get("trainedWords", [])
if isinstance(metadata, list):
trained_words.extend(metadata)
elif isinstance(metadata, str) and metadata:
trained_words.append(metadata)
# Also check metadata field
model = version_info.get("model", {})
if isinstance(model, dict):
model_tags = model.get("tags", [])
if isinstance(model_tags, list):
for tag in model_tags:
if tag not in trained_words:
trained_words.append(tag)
return trained_words
def _extract_model_images(version_info: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Extract images with metadata from model version info.
@@ -1456,43 +1434,6 @@ def _extract_model_images(version_info: Dict[str, Any]) -> List[Dict[str, Any]]:
return images
def _get_metadata_file_path(model_path: str) -> str:
"""
Get the path to the metadata file for a model.
For example, for 'model.safetensors', it returns 'model.metadata.json'
Also checks for variations without extension or with different extensions.
"""
if not model_path:
return ""
# Get directory and filename
directory = os.path.dirname(model_path)
filename = get_filename_from_path(model_path)
# Try different variations of the metadata file name
base_name = filename.rsplit(".", 1)[0] if "." in filename else filename
possible_names = [
base_name + ".metadata.json",
filename + ".metadata.json",
base_name + ".civitai.info",
filename + ".civitai.info",
base_name + ".json",
filename.replace("_", " ").split()[0] + ".metadata.json"
if "_" in base_name
else None,
]
for name in possible_names:
if name:
path = os.path.join(directory, name)
if os.path.exists(path):
log.info(f"Found metadata file: {path}")
return path
return ""
def _read_model_metadata(metadata_path: str) -> Optional[Dict[str, Any]]:
"""
Read model metadata from a JSON file.
@@ -1503,9 +1444,9 @@ def _read_model_metadata(metadata_path: str) -> Optional[Dict[str, Any]]:
return None
try:
files = _as_metadata_list(data.get("files"))
files = as_list(data.get("files"))
file_hashes = [
_as_metadata_dict(file_info).get("hashes")
as_dict(file_info).get("hashes")
for file_info in files
if isinstance(file_info, dict)
]
@@ -1540,28 +1481,6 @@ def _read_model_metadata(metadata_path: str) -> Optional[Dict[str, Any]]:
return None
def _as_metadata_dict(value: Any) -> Dict[str, Any]:
return value if isinstance(value, dict) else {}
def _as_metadata_list(value: Any) -> List[Any]:
if value is None:
return []
if isinstance(value, list):
return value
if isinstance(value, tuple):
return list(value)
return [value] if value else []
def _first_metadata_value(*values: Any) -> Any:
return first_non_empty(*values, default=None)
def _metadata_size_to_bytes(value: Any) -> Optional[int]:
return parse_size_to_bytes(value)
def _format_model_location(file_path: str) -> str:
if not file_path:
return ""
@@ -1572,36 +1491,12 @@ def _format_model_location(file_path: str) -> str:
return location
def _normalize_metadata_trained_words(*values: Any) -> List[str]:
words: List[str] = []
seen = set()
for value in values:
for item in _as_metadata_list(value):
if isinstance(item, dict):
item = (
item.get("word")
or item.get("name")
or item.get("text")
or item.get("value")
)
text = str(item or "").strip()
if not text:
continue
key = text.lower()
if key in seen:
continue
seen.add(key)
words.append(text)
return words
def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
"""
Convert metadata file format to model info format used by our extension.
"""
embedded_civitai_data = _as_metadata_dict(metadata.get("civitai"))
embedded_civitai_data = as_dict(metadata.get("civitai"))
looks_like_civitai_version = bool(
metadata.get("modelId")
or metadata.get("files")
@@ -1611,10 +1506,10 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
civitai_data = embedded_civitai_data or (
metadata if looks_like_civitai_version else {}
)
selected_version = _as_metadata_dict(metadata.get("selected_version"))
selected_version = as_dict(metadata.get("selected_version"))
if not selected_version and looks_like_civitai_version:
selected_version = civitai_data
path_metadata = _as_metadata_dict(metadata.get("path_metadata"))
path_metadata = as_dict(metadata.get("path_metadata"))
# Extract images with metadata
images = []
@@ -1630,7 +1525,7 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
images.append(img_info)
# Get trained words from common sidecar shapes, including LoRA Manager metadata.
trained_words = _normalize_metadata_trained_words(
trained_words = extract_trained_words(
metadata.get("trained_words"),
metadata.get("trainedWords"),
civitai_data.get("trainedWords"),
@@ -1641,11 +1536,11 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
)
# Get model info
model_info = _as_metadata_dict(civitai_data.get("model"))
file_infos = _as_metadata_list(civitai_data.get("files"))
file_info = _as_metadata_dict(file_infos[0]) if file_infos else {}
metadata_hashes = _as_metadata_dict(metadata.get("hashes"))
file_hashes = _as_metadata_dict(file_info.get("hashes"))
model_info = as_dict(civitai_data.get("model"))
file_infos = as_list(civitai_data.get("files"))
file_info = as_dict(file_infos[0]) if file_infos else {}
metadata_hashes = as_dict(metadata.get("hashes"))
file_hashes = as_dict(file_info.get("hashes"))
# Build model_id from CivitAI data
model_id = (
@@ -1664,7 +1559,7 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
or metadata.get("metadata_source")
or "metadata"
)
stored_page_url = _first_metadata_value(
stored_page_url = first_non_empty(
metadata.get("version_url"),
metadata.get("model_url"),
metadata.get("page_url"),
@@ -1689,7 +1584,7 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
"source": details_source,
"details_source": details_source,
"model_id": model_id,
"model_name": _first_metadata_value(
"model_name": first_non_empty(
metadata.get("model_name"),
metadata.get("modelName"),
model_info.get("name"),
@@ -1699,14 +1594,14 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
or "",
"model_type": model_info.get("type", "") or civitai_data.get("type", ""),
"version_id": version_id,
"version_name": _first_metadata_value(
"version_name": first_non_empty(
metadata.get("version_name"),
metadata.get("versionName"),
selected_version.get("name"),
civitai_data.get("name"),
)
or "",
"sha256": _first_metadata_value(
"sha256": first_non_empty(
metadata.get("sha256"),
metadata.get("hash"),
metadata_hashes.get("SHA256"),
@@ -1715,8 +1610,8 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
file_hashes.get("sha256"),
)
or "",
"size": _metadata_size_to_bytes(
_first_metadata_value(
"size": parse_size_to_bytes(
first_non_empty(
metadata.get("size"),
metadata.get("file_size"),
metadata.get("fileSize"),
@@ -1735,12 +1630,12 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
"model_url": stored_page_url or civitai_model_url,
"source_url": stored_page_url,
"platform_url": metadata.get("platform_url") or path_metadata.get("platform_url"),
"download_url": _first_metadata_value(
"download_url": first_non_empty(
metadata.get("download_url"),
civitai_data.get("downloadUrl"),
file_info.get("downloadUrl"),
),
"base_model": _first_metadata_value(
"base_model": first_non_empty(
metadata.get("base_model"),
metadata.get("baseModel"),
selected_version.get("base_model"),
@@ -1748,11 +1643,11 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
civitai_data.get("baseModel"),
)
or "",
"tags": _as_metadata_list(metadata.get("tags") or model_info.get("tags")),
"tags": as_list(metadata.get("tags") or model_info.get("tags")),
"trained_words": trained_words,
"images": images,
"clip_skip": civitai_data.get("clipSkip"),
"description": _first_metadata_value(
"description": first_non_empty(
metadata.get("modelDescription"),
metadata.get("model_description"),
metadata.get("description"),
@@ -1760,7 +1655,7 @@ def _metadata_to_model_info(metadata: Dict[str, Any]) -> Dict[str, Any]:
civitai_data.get("description"),
)
or "",
"model_description": _first_metadata_value(
"model_description": first_non_empty(
metadata.get("modelDescription"),
metadata.get("model_description"),
model_info.get("description"),
@@ -1789,7 +1684,7 @@ def get_model_info_for_file(
log.info(f"get_model_info_for_file called with: {file_path}")
# First check for metadata file
metadata_path = _get_metadata_file_path(file_path)
metadata_path = find_metadata_sidecar_path(file_path)
log.info(f"Looking for metadata file, checked: {metadata_path}")
if metadata_path:
core/sources/huggingface.py
+3 -13
View File
@@ -18,7 +18,7 @@ from ..log_system import create_module_logger
log = create_module_logger(__name__)
from ..path_utils import write_json_atomic, METADATA_DIR, read_json_safe
from ..type_utils import check_credential_http, parse_size_header as _parse_size_header, fetch_remote_file_size
from ..type_utils import check_credential_http, parse_size_header as _parse_size_header, fetch_remote_file_size, fetch_remote_file_size_cached, clear_remote_size_cache
HF_API_URL = "https://huggingface.co/api"
HF_AUTHOR_FALLBACKS = ["Comfy-Org"]
@@ -33,7 +33,6 @@ HF_AUTHOR_INDEX_CACHE_PATH = os.path.join(
# Cache for search results
_search_cache: Dict[str, Any] = {}
_author_index_cache: Dict[str, Dict[str, Any]] = {}
_download_size_cache: Dict[str, Optional[int]] = {}
_author_index_lock = threading.RLock()
@@ -41,7 +40,7 @@ def clear_search_cache():
"""Clear cached HuggingFace search results and in-memory indexes."""
global _search_cache, _author_index_cache
_search_cache.clear()
_download_size_cache.clear()
clear_remote_size_cache()
with _author_index_lock:
_author_index_cache.clear()
@@ -369,16 +368,7 @@ def _fetch_remote_file_size_bytes(
probe_url = _normalize_huggingface_size_probe_url(url)
if not probe_url:
return None
request_headers = {**(headers or {}), "Accept-Encoding": "identity"}
cache_scope = "token" if request_headers.get("Authorization") else "public"
cache_key = f"{cache_scope}::{probe_url}"
if cache_key in _download_size_cache:
return _download_size_cache[cache_key]
size = fetch_remote_file_size(probe_url, headers=request_headers, timeout=timeout)
_download_size_cache[cache_key] = size
return size
return fetch_remote_file_size_cached(probe_url, headers=headers, timeout=timeout)
core/type_utils.py
+143
View File
@@ -667,3 +667,146 @@ def normalize_model_image(image_data: Dict[str, Any], default_civitai_url: str =
}
import re
SHA256_PATTERN = re.compile(r"^[a-fA-F0-9]{64}$")
def normalize_sha256(value: Any) -> str:
"""Return a normalized SHA256 hex string or an empty string."""
if value is None:
return ""
text = str(value).strip()
for prefix in ("sha256:", "sha256="):
if text.lower().startswith(prefix):
text = text[len(prefix):].strip()
return text.lower() if SHA256_PATTERN.match(text) else ""
def unique_ordered_strings(values: List[Any]) -> List[str]:
"""Zwraca unikalne, niepuste ciągi znaków z zachowaniem oryginalnej kolejności."""
seen = set()
unique = []
for value in values:
text = str(value or "").strip()
if not text or text in seen:
continue
seen.add(text)
unique.append(text)
return unique
def extract_sha256_from_metadata(metadata: Any) -> str:
"""
Wyciąga i zwraca pierwszy poprawny znormalizowany hash SHA256 z metadanych.
"""
if not isinstance(metadata, dict):
return ""
for key in ("sha256", "hash", "SHA256", "Sha256"):
val = metadata.get(key)
if val:
normalized = normalize_sha256(val)
if normalized:
return normalized
hashes = metadata.get("hashes")
if isinstance(hashes, dict):
for key in ("SHA256", "sha256", "Sha256", "hash"):
val = hashes.get(key)
if val:
normalized = normalize_sha256(val)
if normalized:
return normalized
# Obsługa zagnieżdżonej listy plików w metadanych Civitai
files = metadata.get("files")
if isinstance(files, list):
for file_info in files:
if isinstance(file_info, dict):
f_hashes = file_info.get("hashes")
if isinstance(f_hashes, dict):
for key in ("SHA256", "sha256", "Sha256", "hash"):
val = f_hashes.get(key)
if val:
normalized = normalize_sha256(val)
if normalized:
return normalized
return ""
def extract_trained_words(*values: Any) -> List[str]:
"""
Parsuje i zwraca znormalizowaną listę unikalnych słów kluczowych
i tagów z przekazanych wartości (słowników wersji, list lub ciągów znaków).
"""
words: List[str] = []
seen = set()
def process_item(item: Any):
if isinstance(item, dict):
if "trainedWords" in item or "trigger" in item or "model" in item:
for sub_val in (item.get("trainedWords"), item.get("trigger")):
if sub_val:
process_item(sub_val)
model = item.get("model")
if isinstance(model, dict) and "tags" in model:
process_item(model.get("tags"))
return
else:
val = (
item.get("word")
or item.get("name")
or item.get("text")
or item.get("value")
)
process_item(val)
return
if isinstance(item, (list, tuple, set)):
for sub in item:
process_item(sub)
return
text = str(item or "").strip()
if not text:
return
key = text.lower()
if key not in seen:
seen.add(key)
words.append(text)
for value in values:
process_item(value)
return words
_remote_size_cache: Dict[tuple[str, Optional[str]], Optional[int]] = {}
def fetch_remote_file_size_cached(
url: str,
headers: Optional[Dict[str, str]] = None,
timeout: int = 15,
) -> Optional[int]:
"""
Sprawdza rozmiar zdalnego pliku, korzystając ze wspólnej pamięci podręcznej.
"""
auth = headers.get("Authorization") if headers else None
cache_key = (url, auth)
if cache_key in _remote_size_cache:
return _remote_size_cache[cache_key]
size = fetch_remote_file_size(url, headers=headers, timeout=timeout)
_remote_size_cache[cache_key] = size
return size
def clear_remote_size_cache() -> None:
"""Czyści cache rozmiarów plików zdalnych."""
_remote_size_cache.clear()
core/workflow_analyzer.py
+4 -16
View File
@@ -21,7 +21,7 @@ except ImportError:
log.warning("Model Resolver: folder_paths not available yet - will retry later")
from .type_utils import URN_TYPE_MAP, MODEL_EXTENSIONS
from .type_utils import URN_TYPE_MAP, MODEL_EXTENSIONS, unique_ordered_strings
URN_REGEX = re.compile(r"^urn:air:([^:]+):([^:]+):([^:]+):(\d+)@(\d+)$")
@@ -295,18 +295,6 @@ def normalize_widget_name(value: Any) -> str:
return re.sub(r"[_\s-]+", "_", str(value or "").strip().lower()).strip("_")
def _unique_strings(values: List[Any]) -> List[str]:
seen = set()
unique = []
for value in values:
text = str(value or "").strip()
if not text or text in seen:
continue
seen.add(text)
unique.append(text)
return unique
def _widget_item_name_candidates(item: Any) -> List[str]:
if not isinstance(item, dict):
return []
@@ -324,7 +312,7 @@ def _widget_item_name_candidates(item: Any) -> List[str]:
elif widget:
candidates.append(widget)
return _unique_strings(candidates)
return unique_ordered_strings(candidates)
def _has_widget_input(item: Any) -> bool:
@@ -386,7 +374,7 @@ def get_widget_name_candidates(node: Dict[str, Any], widget_index: int) -> List[
if isinstance(inputs, list) and widget_index < len(inputs):
candidates.extend(_widget_item_name_candidates(inputs[widget_index]))
return _unique_strings(candidates)
return unique_ordered_strings(candidates)
def get_widget_name_hint(node: Dict[str, Any], widget_index: int) -> str:
@@ -402,7 +390,7 @@ def is_workflow_model_widget_candidate(node: Dict[str, Any], widget_index: int)
def _ordered_unique_categories(values: List[Any]) -> List[str]:
return _unique_strings([value for value in values if value])
return unique_ordered_strings([value for value in values if value])
def _merge_category_hints(