From e8b49dd184b36d3ee66237935799842f2f43daf4 Mon Sep 17 00:00:00 2001 From: Dariusz L Date: Tue, 21 Apr 2026 23:51:46 +0200 Subject: [PATCH] Add AzLogs logging system and integrate Introduce a new logging subsystem under core/log_system (logger.py, config.py, log_funcs.py) implementing AzLogs (LogLevel enum, colored console output, optional file rotation) and a lightweight wrapper (log_debug/info/warn/error/exception) with safe fallback when AzLogs is unavailable. Integrate these log functions across many core modules (downloader, linker, matcher, scanner, workflow_analyzer, workflow_updater, sources and related files) and export logger and helper log functions from core.__init__. Replace several direct logging calls in linker with the new log_* helpers and update core.__init__ exports. Also include small code style and robustness tweaks (path/filename normalization, JSON/file handling, minor refactors and formatting fixes) to improve matching and model-list/popular lookups. --- __init__.py | 2 + core/__init__.py | 17 +- core/downloader.py | 2 + core/linker.py | 16 +- core/log_system/config.py | 6 + core/log_system/log_funcs.py | 68 +++++++ core/log_system/logger.py | 338 +++++++++++++++++++++++++++++++++++ core/matcher.py | 129 +++++++------ core/scanner.py | 2 + core/sources/__init__.py | 28 +-- core/sources/civitai.py | 2 + core/sources/huggingface.py | 2 + core/sources/model_list.py | 162 +++++++++-------- core/sources/popular.py | 56 +++--- core/workflow_analyzer.py | 2 + core/workflow_updater.py | 2 + 16 files changed, 649 insertions(+), 185 deletions(-) create mode 100644 core/log_system/config.py create mode 100644 core/log_system/log_funcs.py create mode 100644 core/log_system/logger.py diff --git a/__init__.py b/__init__.py index 739332d..551aef4 100644 --- a/__init__.py +++ b/__init__.py @@ -6,7 +6,9 @@ @description: Extension for relinking missing models and downloading from HuggingFace/CivitAI """ + import logging +from .core.log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception # Web directory for JavaScript interface WEB_DIRECTORY = "./web" diff --git a/core/__init__.py b/core/__init__.py index cf38d89..a27d8b6 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -11,13 +11,22 @@ Modules: - sources: Search integrations (HuggingFace, CivitAI, popular models) """ +from .log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception +from .log_system.logger import logger from .linker import analyze_and_find_matches, apply_resolution from .scanner import get_model_files from .matcher import find_matches + __all__ = [ - 'analyze_and_find_matches', - 'apply_resolution', - 'get_model_files', - 'find_matches' + "analyze_and_find_matches", + "apply_resolution", + "get_model_files", + "find_matches", + "logger", + "log_debug", + "log_info", + "log_warn", + "log_error", + "log_exception", ] diff --git a/core/downloader.py b/core/downloader.py index 6521000..43e9685 100644 --- a/core/downloader.py +++ b/core/downloader.py @@ -13,6 +13,8 @@ from typing import Dict, Any, Optional, Callable, List from pathlib import Path from collections import deque +from .log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + try: import folder_paths except ImportError: diff --git a/core/linker.py b/core/linker.py index 0ebc9a3..b57aff0 100644 --- a/core/linker.py +++ b/core/linker.py @@ -7,18 +7,16 @@ Integrates all components to provide high-level API for model linking. import os import re import json -import logging from typing import Dict, Any, List, Optional, Tuple from urllib.parse import unquote +from .log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception from .scanner import get_model_files from .workflow_analyzer import analyze_workflow_models, identify_missing_models from .matcher import find_matches from .workflow_updater import update_workflow_nodes from .sources.civitai import resolve_urn -logger = logging.getLogger(__name__) - # Regex patterns for URL extraction (matches HuggingFace and CivitAI URLs) URL_PATTERN = re.compile(r'(https?://(?:huggingface\.co|civitai\.com)[^\s"\'<>\)\\]+)') @@ -207,7 +205,7 @@ def analyze_and_find_matches( """ # Extract URLs from workflow (node.properties.models + regex) workflow_urls = extract_workflow_urls(workflow_json) - logger.debug(f"Extracted {len(workflow_urls)} URLs from workflow") + log_debug(f"Extracted {len(workflow_urls)} URLs from workflow") # Analyze workflow to find all model references all_model_refs = analyze_workflow_models(workflow_json) @@ -239,14 +237,12 @@ def analyze_and_find_matches( "model_name": model_info.get("model_name"), "version_name": model_info.get("version_name"), } - if model_info: missing["civitai_info"] = model_info missing["expected_filename"] = model_info["expected_filename"] - logger.debug( - f"URN {missing['original_path']} → {missing['expected_filename']}" + log_debug( + f"URN {missing['original_path']} -> {missing['expected_filename']}" ) - # Add URN type to missing model for search functionality if "urn" in missing: missing["urn_type"] = missing["urn"].get("type", "") @@ -257,10 +253,10 @@ def analyze_and_find_matches( is_lora_v2 = missing.get("is_lora_v2") exists = missing.get("exists") name = missing.get("name") or missing.get("original_path", "") - logger.debug(f"Checking {name}: is_lora_v2={is_lora_v2}, exists={exists}") + log_debug(f"Checking {name}: is_lora_v2={is_lora_v2}, exists={exists}") if is_lora_v2 and exists: - logger.info(f"Skipping LoraManager lora {name} - already exists locally") + log_info(f"Skipping LoraManager lora {name} - already exists locally") continue target_for_matching = missing.get("original_path", "") diff --git a/core/log_system/config.py b/core/log_system/config.py new file mode 100644 index 0000000..7fa2bf5 --- /dev/null +++ b/core/log_system/config.py @@ -0,0 +1,6 @@ +# Log level for development. +# Possible values: 'DEBUG', 'INFO', 'WARN', 'ERROR', 'NONE' +LOG_LEVEL = "DEBUG" + +# Logger module name. Set to None to auto-detect from folder name. +LOG_MODULE_NAME = None diff --git a/core/log_system/log_funcs.py b/core/log_system/log_funcs.py new file mode 100644 index 0000000..7bd022c --- /dev/null +++ b/core/log_system/log_funcs.py @@ -0,0 +1,68 @@ +""" +@author: Azornes +@title: AzLogs +@version: 1.4.0 +@description: Logging Initializator +""" + +import os +import traceback + +try: + from .logger import logger, LogLevel, debug, info, warn, error, exception + from .config import LOG_LEVEL, LOG_MODULE_NAME + + _module_name = ( + LOG_MODULE_NAME + if LOG_MODULE_NAME is not None + else os.path.basename(os.path.dirname(os.path.abspath(__file__))) + ) + + logger.set_module_level(_module_name, LogLevel[LOG_LEVEL]) + + logger.configure( + { + "log_to_file": True, + "log_dir": os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs"), + } + ) + + _initialized = True +except ImportError: + _initialized = False + + +def log_debug(*args, **kwargs): + if _initialized: + debug(_module_name, *args, **kwargs) + else: + print("[DEBUG]", *args) + + +def log_info(*args, **kwargs): + if _initialized: + info(_module_name, *args, **kwargs) + else: + print("[INFO]", *args) + + +def log_warn(*args, **kwargs): + if _initialized: + warn(_module_name, *args, **kwargs) + else: + print("[WARN]", *args) + + +def log_error(*args, **kwargs): + if _initialized: + error(_module_name, *args, **kwargs) + else: + print("[ERROR]", *args) + + +def log_exception(*args): + if _initialized: + exception(_module_name, *args) + else: + print("[ERROR]", *args) + traceback.print_exc() diff --git a/core/log_system/logger.py b/core/log_system/logger.py new file mode 100644 index 0000000..69f9d83 --- /dev/null +++ b/core/log_system/logger.py @@ -0,0 +1,338 @@ +""" +@author: Azornes +@title: AzLogs +@version: 1.4.0 +@description: Logging Setup - Central logging system + +Features: +- Different log levels (DEBUG, INFO, WARN, ERROR) +- Ability to enable/disable logs globally or per module +- Colored logs in console +- Log file rotation +- Configuration via environment variables +""" + +import os +import sys +import json +import logging +import datetime +from enum import IntEnum +from logging.handlers import RotatingFileHandler +import traceback + + +# Log levels +class LogLevel(IntEnum): + DEBUG = 10 + INFO = 20 + WARN = 30 + ERROR = 40 + NONE = 100 + + +# Level mapping +LEVEL_MAP = { + LogLevel.DEBUG: logging.DEBUG, + LogLevel.INFO: logging.INFO, + LogLevel.WARN: logging.WARNING, + LogLevel.ERROR: logging.ERROR, + LogLevel.NONE: logging.CRITICAL + 1, +} + +# ANSI colors for different log levels +COLORS = { + LogLevel.DEBUG: "\033[90m", # Gray + LogLevel.INFO: "\033[94m", # Blue + LogLevel.WARN: "\033[93m", # Yellow + LogLevel.ERROR: "\033[91m", # Red + "RESET": "\033[0m", # Reset +} + +# Default configuration +DEFAULT_CONFIG = { + "global_level": LogLevel.INFO, + "module_settings": {}, + "use_colors": True, + "log_to_file": False, + "log_dir": "logs", + "max_file_size_mb": 10, + "backup_count": 5, + "timestamp_format": "%H:%M:%S", +} + + +class ColoredFormatter(logging.Formatter): + """Formatter that adds colors to console logs""" + + def __init__(self, fmt=None, datefmt=None, use_colors=True): + super().__init__(fmt, datefmt) + self.use_colors = use_colors + + def format(self, record): + # Get the formatted message from the record + message = record.getMessage() + if record.exc_info: + message += "\n" + self.formatException(record.exc_info) + + levelname = record.levelname + + # Build the log prefix + prefix = "[{}] [{}] [{}]".format( + self.formatTime(record, self.datefmt), record.name, record.levelname + ) + + # Apply color and bold styling to the prefix + if self.use_colors and hasattr(LogLevel, levelname): + level_enum = getattr(LogLevel, levelname) + if level_enum in COLORS: + # Apply bold (\033[1m) and color, then reset + prefix = f"\033[1m{COLORS[level_enum]}{prefix}{COLORS['RESET']}" + + return f"{prefix} {message}" + + +class AzLogsLogger: + """Main logger class for AzLogs""" + + _instance = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super(AzLogsLogger, cls).__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + if self._initialized: + return + + self.config = DEFAULT_CONFIG.copy() + self.enabled = True + self.loggers = {} + + # Load configuration from environment variables + self._load_config_from_env() + + self._initialized = True + + def _load_config_from_env(self): + """Load configuration from environment variables""" + + # Global level + if "AZLOGS_LOG_LEVEL" in os.environ: + level_name = os.environ["AZLOGS_LOG_LEVEL"].upper() + if hasattr(LogLevel, level_name): + self.config["global_level"] = getattr(LogLevel, level_name) + + # Module settings + if "AZLOGS_MODULE_LEVELS" in os.environ: + try: + module_settings = json.loads(os.environ["AZLOGS_MODULE_LEVELS"]) + for module, level_name in module_settings.items(): + if hasattr(LogLevel, level_name.upper()): + self.config["module_settings"][module] = getattr( + LogLevel, level_name.upper() + ) + except json.JSONDecodeError: + pass + + # Other settings + if "AZLOGS_USE_COLORS" in os.environ: + self.config["use_colors"] = ( + os.environ["AZLOGS_USE_COLORS"].lower() == "true" + ) + + if "AZLOGS_LOG_TO_FILE" in os.environ: + self.config["log_to_file"] = ( + os.environ["AZLOGS_LOG_TO_FILE"].lower() == "true" + ) + + if "AZLOGS_LOG_DIR" in os.environ: + self.config["log_dir"] = os.environ["AZLOGS_LOG_DIR"] + + if "AZLOGS_MAX_FILE_SIZE_MB" in os.environ: + try: + self.config["max_file_size_mb"] = int( + os.environ["AZLOGS_MAX_FILE_SIZE_MB"] + ) + except ValueError: + pass + + if "AZLOGS_BACKUP_COUNT" in os.environ: + try: + self.config["backup_count"] = int(os.environ["AZLOGS_BACKUP_COUNT"]) + except ValueError: + pass + + def configure(self, config): + """Configure the logger""" + self.config.update(config) + + # If file logging is enabled, ensure the directory exists + if self.config.get("log_to_file") and self.config.get("log_dir"): + try: + os.makedirs(self.config["log_dir"], exist_ok=True) + except OSError as e: + # This is a critical situation, so use print + print( + f"[CRITICAL] Could not create log directory: {self.config['log_dir']}. Error: {e}" + ) + traceback.print_exc() + # Disable file logging to avoid further errors + self.config["log_to_file"] = False + + return self + + def set_enabled(self, enabled): + """Enable/disable the logger globally""" + self.enabled = enabled + return self + + def set_global_level(self, level): + """Set global logging level""" + self.config["global_level"] = level + return self + + def set_module_level(self, module, level): + """Set logging level for a specific module""" + self.config["module_settings"][module] = level + return self + + def is_level_enabled(self, module, level): + """Check if a given logging level is active for the module""" + if not self.enabled: + return False + + # Determine effective logging level, considering module and global settings + effective_level = self.config["module_settings"].get( + module, self.config["global_level"] + ) + + # If effective level is NONE, logging is completely disabled + if effective_level == LogLevel.NONE: + return False + + # Otherwise check if log level is high enough + return level >= effective_level + + def _get_logger(self, module): + """Get or create a logger for the module""" + if module in self.loggers: + return self.loggers[module] + + # Create new logger + logger = logging.getLogger(f"azlogs.{module}") + logger.setLevel(logging.DEBUG) # Set lowest level, filtering will be done later + logger.propagate = False + + # Add console handler + console_handler = logging.StreamHandler(sys.stdout) + console_formatter = ColoredFormatter( + fmt="[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s", + datefmt=self.config["timestamp_format"], + use_colors=self.config["use_colors"], + ) + console_handler.setFormatter(console_formatter) + logger.addHandler(console_handler) + + # Add file handler if file logging is enabled + if self.config["log_to_file"]: + log_file = os.path.join(self.config["log_dir"], f"azlogs_{module}.log") + file_handler = RotatingFileHandler( + log_file, + maxBytes=self.config["max_file_size_mb"] * 1024 * 1024, + backupCount=self.config["backup_count"], + encoding="utf-8", + ) + file_formatter = logging.Formatter( + fmt="[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + file_handler.setFormatter(file_formatter) + logger.addHandler(file_handler) + + self.loggers[module] = logger + return logger + + def log(self, module, level, *args, **kwargs): + """Write log""" + if not self.is_level_enabled(module, level): + return + + logger = self._get_logger(module) + + # Convert arguments to string + message = " ".join(str(arg) for arg in args) + + # Add exception info if provided + exc_info = kwargs.get("exc_info", None) + + # Map LogLevel to logging level + log_level = LEVEL_MAP.get(level, logging.INFO) + + # Write log + logger.log(log_level, message, exc_info=exc_info) + + def debug(self, module, *args, **kwargs): + """Log at DEBUG level""" + + def info(self, module, *args, **kwargs): + """Log at INFO level""" + + def warn(self, module, *args, **kwargs): + """Log at WARN level""" + + def error(self, module, *args, **kwargs): + """Log at ERROR level""" + + def exception(self, module, *args): + """Log exception at ERROR level""" + + +# Singleton +logger = AzLogsLogger() + + +# Helper functions +def debug(module, *args, **kwargs): + """Log at DEBUG level""" + + +def info(module, *args, **kwargs): + """Log at INFO level""" + + +def warn(module, *args, **kwargs): + """Log at WARN level""" + + +def error(module, *args, **kwargs): + """Log at ERROR level""" + + +def exception(module, *args): + """Log exception at ERROR level""" + + +# Function to quickly enable/disable debugging +def set_debug(enabled=True): + """Enable/disable debugging globally""" + if enabled: + logger.set_global_level(LogLevel.DEBUG) + else: + logger.set_global_level(LogLevel.INFO) + return logger + + +# Function to enable/disable file logging +def set_file_logging(enabled=True, log_dir=None): + """Enable/disable logging to file""" + logger.config["log_to_file"] = enabled + if log_dir: + logger.config["log_dir"] = log_dir + os.makedirs(log_dir, exist_ok=True) + + # Reset loggers to apply new settings + logger.loggers = {} + return logger diff --git a/core/matcher.py b/core/matcher.py index 8bf4f09..108c365 100644 --- a/core/matcher.py +++ b/core/matcher.py @@ -9,45 +9,47 @@ import re from typing import List, Dict, Tuple from difflib import SequenceMatcher +from .log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + def normalize_filename(filename: str) -> str: """ Normalize a filename for comparison. - + Removes file extension, converts to lowercase, and normalizes separators (underscores, hyphens, spaces). - + Args: filename: Filename to normalize - + Returns: Normalized string for comparison """ # Remove file extension base = os.path.splitext(filename)[0] - + # Convert to lowercase base = base.lower() - + # Normalize separators: replace underscores, hyphens, and spaces with a single space - base = re.sub(r'[_\-\s]+', ' ', base) - + base = re.sub(r"[_\-\s]+", " ", base) + # Strip whitespace base = base.strip() - + return base def calculate_similarity(str1: str, str2: str) -> float: """ Calculate similarity score between two strings (0.0 to 1.0). - + Uses SequenceMatcher to compute a ratio. - + Args: str1: First string str2: Second string - + Returns: Similarity score from 0.0 (completely different) to 1.0 (identical) """ @@ -57,13 +59,13 @@ def calculate_similarity(str1: str, str2: str) -> float: def calculate_similarity_with_normalization(str1: str, str2: str) -> float: """ Calculate similarity score with filename normalization. - + Normalizes both strings before comparing. - + Args: str1: First string (typically model filename) str2: Second string (typically candidate model filename) - + Returns: Similarity score from 0.0 to 1.0 """ @@ -76,17 +78,17 @@ def find_matches( target_model: str, candidate_models: List[Dict[str, str]], threshold: float = 0.0, - max_results: int = 10 + max_results: int = 10, ) -> List[Dict[str, any]]: """ Find similar models using fuzzy matching. - + Args: target_model: The target model filename/path to match candidate_models: List of candidate model dictionaries with 'filename' or 'path' key threshold: Minimum similarity score (0.0 to 1.0) to include in results max_results: Maximum number of results to return - + Returns: List of match dictionaries sorted by similarity (highest first): { @@ -97,37 +99,41 @@ def find_matches( } """ matches = [] - + # Normalize path separators in target_model based on current OS # This ensures paths with \ vs / separators are treated as identical - target_model_normalized = os.path.normpath(target_model) if target_model else '' - + target_model_normalized = os.path.normpath(target_model) if target_model else "" + # Extract just the filename from target_model (remove any subfolder paths) # target_model might be just a filename or might include subfolder paths target_filename = os.path.basename(target_model_normalized) - + # Normalize target filename once for exact match comparisons target_norm = normalize_filename(target_filename) - + 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', '') - + 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 = os.path.basename(candidate_path) - + if not candidate_filename: continue - + # Normalize candidate path separators based on current OS # This ensures paths with \ vs / separators are treated as identical - candidate_path_normalized = os.path.normpath(candidate_path) if candidate_path else '' - candidate_relative_path = candidate.get('relative_path', '') - candidate_relative_path_normalized = os.path.normpath(candidate_relative_path) if candidate_relative_path else '' - + candidate_path_normalized = ( + os.path.normpath(candidate_path) if candidate_path else "" + ) + candidate_relative_path = candidate.get("relative_path", "") + candidate_relative_path_normalized = ( + os.path.normpath(candidate_relative_path) if candidate_relative_path else "" + ) + # Check if normalized paths are identical (100% match) # This handles cases where paths differ only by separator (e.g., path/to/model vs path\to\model) # Compare both absolute paths and relative paths @@ -139,62 +145,69 @@ def find_matches( # Also check if relative path matches the target (which might be relative) if candidate_relative_path_normalized == target_model_normalized: path_match = True - + if path_match: # Exact path match after normalization = 100% confidence similarity = 1.0 - matches.append({ - 'model': candidate, - 'filename': candidate_filename, - 'similarity': similarity, - 'confidence': round(similarity * 100, 1) - }) + matches.append( + { + "model": candidate, + "filename": candidate_filename, + "similarity": similarity, + "confidence": round(similarity * 100, 1), + } + ) continue - + # Calculate similarity comparing just filenames (not paths) # This ensures we're comparing apples to apples - + # First check for exact match (after normalization) - should be 100% # Only exact matches should get 100% confidence candidate_norm = normalize_filename(candidate_filename) - + if target_norm == candidate_norm: # Exact match after normalization = 100% confidence similarity = 1.0 else: # Calculate similarity using SequenceMatcher # This gives a ratio between 0.0 and 1.0 based on longest common subsequence - similarity = calculate_similarity_with_normalization(target_filename, candidate_filename) - + similarity = calculate_similarity_with_normalization( + target_filename, candidate_filename + ) + # Also try comparing without extensions for better matching target_base = os.path.splitext(target_filename)[0] candidate_base = os.path.splitext(candidate_filename)[0] - similarity_no_ext = calculate_similarity_with_normalization(target_base, candidate_base) - + 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 # SequenceMatcher 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 - + # Only include if above threshold if similarity >= threshold: - matches.append({ - 'model': candidate, - 'filename': candidate_filename, - 'similarity': similarity, - 'confidence': round(similarity * 100, 1) # Convert to percentage - }) - + matches.append( + { + "model": candidate, + "filename": candidate_filename, + "similarity": similarity, + "confidence": round(similarity * 100, 1), # Convert to percentage + } + ) + # Sort by similarity (highest first) - matches.sort(key=lambda x: x['similarity'], reverse=True) - + matches.sort(key=lambda x: x["similarity"], reverse=True) + # Limit to max_results matches = matches[:max_results] - - return matches + return matches diff --git a/core/scanner.py b/core/scanner.py index 1347e1c..db90d65 100644 --- a/core/scanner.py +++ b/core/scanner.py @@ -8,6 +8,8 @@ import os import logging from typing import List, Dict, Tuple +from .log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + # Import folder_paths lazily - it may not be available until ComfyUI is initialized try: import folder_paths diff --git a/core/sources/__init__.py b/core/sources/__init__.py index f772fed..e444a4b 100644 --- a/core/sources/__init__.py +++ b/core/sources/__init__.py @@ -4,20 +4,26 @@ Model Sources Module Provides search functionality for finding models from various sources. """ +from ..log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + from .popular import search_popular_models, get_popular_model_url from .model_list import search_model_list, search_model_list_multiple -from .huggingface import search_huggingface, search_huggingface_for_file, get_huggingface_download_url +from .huggingface import ( + search_huggingface, + search_huggingface_for_file, + get_huggingface_download_url, +) from .civitai import search_civitai, search_civitai_for_file, get_civitai_download_url __all__ = [ - 'search_popular_models', - 'get_popular_model_url', - 'search_model_list', - 'search_model_list_multiple', - 'search_huggingface', - 'search_huggingface_for_file', - 'get_huggingface_download_url', - 'search_civitai', - 'search_civitai_for_file', - 'get_civitai_download_url' + "search_popular_models", + "get_popular_model_url", + "search_model_list", + "search_model_list_multiple", + "search_huggingface", + "search_huggingface_for_file", + "get_huggingface_download_url", + "search_civitai", + "search_civitai_for_file", + "get_civitai_download_url", ] diff --git a/core/sources/civitai.py b/core/sources/civitai.py index 9b9faea..593b1af 100644 --- a/core/sources/civitai.py +++ b/core/sources/civitai.py @@ -13,6 +13,8 @@ import requests from typing import Dict, Any, Optional, List from urllib.parse import urlparse, parse_qs, quote +from ..log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + logger = logging.getLogger(__name__) CIVITAI_API_URL = "https://civitai.com/api/v1" diff --git a/core/sources/huggingface.py b/core/sources/huggingface.py index dbee763..85fbb2e 100644 --- a/core/sources/huggingface.py +++ b/core/sources/huggingface.py @@ -11,6 +11,8 @@ import requests from typing import Dict, Any, Optional, List from urllib.parse import urlparse, quote +from ..log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + logger = logging.getLogger(__name__) HF_API_URL = "https://huggingface.co/api" diff --git a/core/sources/model_list.py b/core/sources/model_list.py index f8eac4b..2b06621 100644 --- a/core/sources/model_list.py +++ b/core/sources/model_list.py @@ -10,11 +10,15 @@ import logging from typing import Dict, Any, Optional, List from difflib import SequenceMatcher +from ..log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + logger = logging.getLogger(__name__) # Path to metadata directory -METADATA_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'metadata') -MODEL_LIST_FILE = os.path.join(METADATA_DIR, 'model-list.json') +METADATA_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "metadata" +) +MODEL_LIST_FILE = os.path.join(METADATA_DIR, "model-list.json") # Cache for loaded data _model_list_cache: Optional[List[Dict]] = None @@ -23,20 +27,22 @@ _model_list_cache: Optional[List[Dict]] = None def _load_model_list() -> List[Dict]: """Load model list database.""" global _model_list_cache - + if _model_list_cache is not None: return _model_list_cache - + try: if os.path.exists(MODEL_LIST_FILE): - with open(MODEL_LIST_FILE, 'r', encoding='utf-8') as f: + with open(MODEL_LIST_FILE, "r", encoding="utf-8") as f: data = json.load(f) - _model_list_cache = data.get('models', []) - logger.info(f"Loaded {len(_model_list_cache)} models from model-list.json") + _model_list_cache = data.get("models", []) + logger.info( + f"Loaded {len(_model_list_cache)} models from model-list.json" + ) return _model_list_cache except Exception as e: logger.error(f"Error loading model list: {e}") - + _model_list_cache = [] return _model_list_cache @@ -46,7 +52,7 @@ def _normalize_filename(filename: str) -> str: # Remove extension base = os.path.splitext(filename)[0].lower() # Replace separators with spaces - base = base.replace('-', ' ').replace('_', ' ').replace('.', ' ') + base = base.replace("-", " ").replace("_", " ").replace(".", " ") return base @@ -55,149 +61,153 @@ def _similarity(a: str, b: str) -> float: return SequenceMatcher(None, a, b).ratio() -def search_model_list(filename: str, exact_only: bool = False) -> Optional[Dict[str, Any]]: +def search_model_list( + filename: str, exact_only: bool = False +) -> Optional[Dict[str, Any]]: """ Search model-list.json for a model by filename. Uses exact match first, then fuzzy matching (unless exact_only=True). - + Args: filename: Model filename to search for exact_only: If True, only return exact matches (for downloads). If False, also try fuzzy matching (for local file resolution). - + Returns: Dict with url, filename, type, etc. if found, None otherwise """ models = _load_model_list() if not models: return None - + filename_lower = filename.lower() filename_base = os.path.splitext(filename_lower)[0] filename_norm = _normalize_filename(filename) - + # 1. Exact match first (always try this) for model in models: - model_filename = model.get('filename', '') + model_filename = model.get("filename", "") if model_filename.lower() == filename_lower: - url = model.get('url', '') + url = model.get("url", "") if url: return { - 'source': 'model_list', - 'filename': model_filename, - 'url': url, - 'name': model.get('name', ''), - 'type': model.get('type', ''), - 'directory': model.get('save_path', 'checkpoints'), - 'size': model.get('size', ''), - 'match_type': 'exact' + "source": "model_list", + "filename": model_filename, + "url": url, + "name": model.get("name", ""), + "type": model.get("type", ""), + "directory": model.get("save_path", "checkpoints"), + "size": model.get("size", ""), + "match_type": "exact", } - + # If exact_only is True, don't try fuzzy matching - prevents confusing # users with wrong model suggestions for downloads if exact_only: return None - + # 2. Fuzzy substring match - check if filename contains or is contained by model name # WMD returns immediately on first substring match - do the same for reliability for model in models: - model_filename = model.get('filename', '') + model_filename = model.get("filename", "") if not model_filename: continue - + model_base = os.path.splitext(model_filename.lower())[0] - + # Check substring matches (exactly like WMD does) if filename_base in model_base or model_base in filename_base: - url = model.get('url', '') + url = model.get("url", "") if url: score = _similarity(filename_norm, _normalize_filename(model_filename)) return { - 'source': 'model_list', - 'filename': model_filename, - 'url': url, - 'name': model.get('name', ''), - 'type': model.get('type', ''), - 'directory': model.get('save_path', 'checkpoints'), - 'size': model.get('size', ''), - 'match_type': 'fuzzy', - 'confidence': round(score * 100, 1) + "source": "model_list", + "filename": model_filename, + "url": url, + "name": model.get("name", ""), + "type": model.get("type", ""), + "directory": model.get("save_path", "checkpoints"), + "size": model.get("size", ""), + "match_type": "fuzzy", + "confidence": round(score * 100, 1), } - + # 3. Try normalized similarity matching on all models (fallback) best_match = None best_score = 0.0 - + for model in models: - model_filename = model.get('filename', '') + model_filename = model.get("filename", "") if not model_filename: continue - + model_norm = _normalize_filename(model_filename) score = _similarity(filename_norm, model_norm) - + if score > best_score and score > 0.5: # Require 50% similarity - url = model.get('url', '') + url = model.get("url", "") if url: best_score = score best_match = { - 'source': 'model_list', - 'filename': model_filename, - 'url': url, - 'name': model.get('name', ''), - 'type': model.get('type', ''), - 'directory': model.get('save_path', 'checkpoints'), - 'size': model.get('size', ''), - 'match_type': 'similar', - 'confidence': round(score * 100, 1) + "source": "model_list", + "filename": model_filename, + "url": url, + "name": model.get("name", ""), + "type": model.get("type", ""), + "directory": model.get("save_path", "checkpoints"), + "size": model.get("size", ""), + "match_type": "similar", + "confidence": round(score * 100, 1), } - + return best_match def search_model_list_multiple(filename: str, limit: int = 5) -> List[Dict[str, Any]]: """ Search model-list.json and return multiple fuzzy matches. - + Args: filename: Model filename to search for limit: Maximum results to return - + Returns: List of matching models sorted by relevance """ models = _load_model_list() if not models: return [] - + filename_norm = _normalize_filename(filename) results = [] - + for model in models: - model_filename = model.get('filename', '') + model_filename = model.get("filename", "") if not model_filename: continue - + model_norm = _normalize_filename(model_filename) score = _similarity(filename_norm, model_norm) - + if score > 0.4: # Minimum 40% similarity - url = model.get('url', '') + url = model.get("url", "") if url: - results.append({ - 'source': 'model_list', - 'filename': model_filename, - 'url': url, - 'name': model.get('name', ''), - 'type': model.get('type', ''), - 'directory': model.get('save_path', 'checkpoints'), - 'size': model.get('size', ''), - 'confidence': round(score * 100, 1) - }) - + results.append( + { + "source": "model_list", + "filename": model_filename, + "url": url, + "name": model.get("name", ""), + "type": model.get("type", ""), + "directory": model.get("save_path", "checkpoints"), + "size": model.get("size", ""), + "confidence": round(score * 100, 1), + } + ) + # Sort by confidence descending - results.sort(key=lambda x: x['confidence'], reverse=True) - + results.sort(key=lambda x: x["confidence"], reverse=True) + return results[:limit] diff --git a/core/sources/popular.py b/core/sources/popular.py index 5f72d07..cb2fe86 100644 --- a/core/sources/popular.py +++ b/core/sources/popular.py @@ -9,12 +9,16 @@ import json import logging from typing import Dict, Any, Optional, List +from ..log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + logger = logging.getLogger(__name__) # Path to metadata directory -METADATA_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'metadata') -POPULAR_MODELS_FILE = os.path.join(METADATA_DIR, 'popular-models.json') -MODEL_ALIASES_FILE = os.path.join(METADATA_DIR, 'model-aliases.json') +METADATA_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "metadata" +) +POPULAR_MODELS_FILE = os.path.join(METADATA_DIR, "popular-models.json") +MODEL_ALIASES_FILE = os.path.join(METADATA_DIR, "model-aliases.json") # Cache for loaded data _popular_models_cache: Optional[Dict] = None @@ -24,19 +28,19 @@ _model_aliases_cache: Optional[Dict] = None def _load_popular_models() -> Dict[str, Any]: """Load popular models database.""" global _popular_models_cache - + if _popular_models_cache is not None: return _popular_models_cache - + try: if os.path.exists(POPULAR_MODELS_FILE): - with open(POPULAR_MODELS_FILE, 'r', encoding='utf-8') as f: + with open(POPULAR_MODELS_FILE, "r", encoding="utf-8") as f: data = json.load(f) - _popular_models_cache = data.get('models', {}) + _popular_models_cache = data.get("models", {}) return _popular_models_cache except Exception as e: logger.error(f"Error loading popular models: {e}") - + _popular_models_cache = {} return _popular_models_cache @@ -44,19 +48,19 @@ def _load_popular_models() -> Dict[str, Any]: def _load_model_aliases() -> Dict[str, List[str]]: """Load model aliases database.""" global _model_aliases_cache - + if _model_aliases_cache is not None: return _model_aliases_cache - + try: if os.path.exists(MODEL_ALIASES_FILE): - with open(MODEL_ALIASES_FILE, 'r', encoding='utf-8') as f: + with open(MODEL_ALIASES_FILE, "r", encoding="utf-8") as f: data = json.load(f) - _model_aliases_cache = data.get('aliases', {}) + _model_aliases_cache = data.get("aliases", {}) return _model_aliases_cache except Exception as e: logger.error(f"Error loading model aliases: {e}") - + _model_aliases_cache = {} return _model_aliases_cache @@ -64,61 +68,61 @@ def _load_model_aliases() -> Dict[str, List[str]]: def get_popular_model_url(filename: str) -> Optional[Dict[str, Any]]: """ Look up a model filename in the popular models database. - + Args: filename: Model filename to look up - + Returns: Dictionary with url, type, directory if found, None otherwise """ models = _load_popular_models() - + # Direct lookup if filename in models: return models[filename].copy() - + # Try lowercase filename_lower = filename.lower() for name, info in models.items(): if name.lower() == filename_lower: return info.copy() - + # Try aliases aliases = _load_model_aliases() for canonical, alias_list in aliases.items(): if filename in alias_list or filename_lower in [a.lower() for a in alias_list]: if canonical in models: result = models[canonical].copy() - result['canonical_name'] = canonical + result["canonical_name"] = canonical return result - + return None def search_popular_models(query: str, limit: int = 10) -> List[Dict[str, Any]]: """ Search popular models database by filename pattern. - + Args: query: Search query (partial filename) limit: Maximum results to return - + Returns: List of matching models with url info """ models = _load_popular_models() query_lower = query.lower() - + results = [] for name, info in models.items(): if query_lower in name.lower(): result = info.copy() - result['filename'] = name + result["filename"] = name results.append(result) - + if len(results) >= limit: break - + return results diff --git a/core/workflow_analyzer.py b/core/workflow_analyzer.py index 054fd0f..4a46c21 100644 --- a/core/workflow_analyzer.py +++ b/core/workflow_analyzer.py @@ -8,6 +8,8 @@ import os import logging from typing import List, Dict, Any, Optional +from .log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + # Import folder_paths lazily - it may not be available until ComfyUI is initialized try: import folder_paths diff --git a/core/workflow_updater.py b/core/workflow_updater.py index d8c662a..474956c 100644 --- a/core/workflow_updater.py +++ b/core/workflow_updater.py @@ -8,6 +8,8 @@ import os import logging from typing import Dict, Any, List, Optional +from .log_system.log_funcs import log_debug, log_info, log_warn, log_error, log_exception + def convert_to_relative_path( absolute_path: str, category: str, base_directory: str = None