refactor: remove workflow analyzer module

This commit is contained in:
Dariusz L
2026-08-02 19:41:30 +02:00
parent 9a7b16cde2
commit dc11d426ae
2 changed files with 1 additions and 64 deletions
+1 -1
View File
@@ -5,7 +5,7 @@ Modules:
- linker: Main API for analyzing and resolving missing models
- scanner: Directory scanning for available models
- matcher: Fuzzy matching for finding similar models
- workflow_analyzer: Workflow JSON parsing
- workflow: Workflow parsing, matching, analysis, and inventory
- workflow_updater: Workflow modification
- downloader: Model downloading with progress tracking
- sources: Search integrations (HuggingFace, CivitAI, popular models)
-63
View File
@@ -1,63 +0,0 @@
"""Workflow missing-model grouping compatibility module."""
from typing import Any, Dict, List
def identify_missing_models(
workflow_models: List[Dict[str, Any]], available_models: List[Dict[str, str]] = None
) -> List[Dict[str, Any]]:
"""
Identify which models from the workflow are missing.
Deduplicates by category and filename - same model file only appears once even if
referenced by multiple nodes.
Args:
workflow_models: List of model references from analyze_workflow_models
available_models: Optional list of available models (if None, checks via folder_paths)
Returns:
List of missing model references (deduplicated by category and filename).
Each entry has 'all_node_refs' containing all node references for that model.
"""
# Group missing models by category and filename to deduplicate workflow
# references without merging unrelated model folders that happen to share a name.
missing_by_model: Dict[tuple[str, str], Dict[str, Any]] = {}
for model_ref in workflow_models:
# If exists is False, it's missing
if not model_ref.get("exists", False):
filename = model_ref.get("original_path", "")
category = model_ref.get("category", "")
group_key = (str(category or ""), str(filename or ""))
if group_key not in missing_by_model:
# First occurrence - use this as the primary entry
missing_by_model[group_key] = {
**model_ref,
"reference_count": 1,
"all_node_refs": [
model_ref.copy()
], # Track all nodes needing this model
}
else:
# Duplicate - just add to the node refs list
existing = missing_by_model[group_key]
existing["all_node_refs"].append(model_ref.copy())
existing["reference_count"] = len(existing["all_node_refs"])
if model_ref.get("auto_download_capable"):
existing["auto_download_capable"] = True
if model_ref.get("auto_download_candidate"):
existing["auto_download_candidate"] = True
if model_ref.get("input_choice_matches_value"):
existing["input_choice_matches_value"] = True
existing_source = str(existing.get("input_choice_source") or "").lower()
model_source = str(model_ref.get("input_choice_source") or "").lower()
if existing_source != "hybrid" and model_source in {
"static",
"hybrid",
"workflow_schema",
}:
existing["input_choice_source"] = model_source
# Return deduplicated list
return list(missing_by_model.values())