diff --git a/__init__.py b/__init__.py index 89ddde1..b344491 100644 --- a/__init__.py +++ b/__init__.py @@ -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): diff --git a/core/path_utils.py b/core/path_utils.py index 660aa7d..56dc05b 100644 --- a/core/path_utils.py +++ b/core/path_utils.py @@ -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 "" diff --git a/core/resolver.py b/core/resolver.py index daa9439..19b56c2 100644 --- a/core/resolver.py +++ b/core/resolver.py @@ -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": "", diff --git a/core/sources/civarchive.py b/core/sources/civarchive.py index 82a4cf9..7f9ab56 100644 --- a/core/sources/civarchive.py +++ b/core/sources/civarchive.py @@ -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}" diff --git a/core/sources/civitai.py b/core/sources/civitai.py index d4dfb62..30b2116 100644 --- a/core/sources/civitai.py +++ b/core/sources/civitai.py @@ -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: diff --git a/core/sources/huggingface.py b/core/sources/huggingface.py index c283db3..b3315a5 100644 --- a/core/sources/huggingface.py +++ b/core/sources/huggingface.py @@ -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) diff --git a/core/type_utils.py b/core/type_utils.py index 0fdb9b1..a292001 100644 --- a/core/type_utils.py +++ b/core/type_utils.py @@ -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() + + + diff --git a/core/workflow_analyzer.py b/core/workflow_analyzer.py index c6c4000..46ffbf9 100644 --- a/core/workflow_analyzer.py +++ b/core/workflow_analyzer.py @@ -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(