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

feat: add FATAL log level, refactor logging colors and stacklevel

Browse Source 
- Add FATAL (50) to LogLevel enum, LEVEL_MAP, and LEVEL_THEME (bright red #E74C3C)
- Drop legacy COLORS dict in favor of LEVEL_THEME RGB values + ANSI_RESET
- Replace hard-coded stacklevel=4 with _INTERNAL_FRAMES constant so caller
  location is always correct regardless of call-chain depth
- Simplify _normalize_module_name by removing redundant path normalization
- Add fatal() to ModuleLogger, AzLogsLogger, and module-level API
- Lazily init _log in path_utils via _get_log() to avoid circular imports
This commit is contained in:
Dariusz L
2026-06-30 22:07:15 +02:00
parent b59be44935
commit c6441aa92e
4 changed files with 82 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/log_system/config.py
+1 -1
View File
@@ -5,7 +5,7 @@
@description: Logging Configuration
"""
# Log level for development.
# Possible values: 'DEBUG', 'INFO', 'WARN', 'ERROR', 'NONE'
# Possible values: 'DEBUG', 'INFO', 'WARN', 'ERROR', 'FATAL', 'NONE'
LOG_LEVEL = "DEBUG"
# Logger module name. Set to None to auto-detect from folder name.
core/log_system/log_funcs.py
+30 -36
View File
@@ -15,7 +15,7 @@ _project_root = None
_default_module_name = __name__
try:
from .logger import logger, LogLevel, debug, info, warn, error, exception
from .logger import logger, LogLevel, debug, info, warn, error, exception, fatal
from .config import LOG_LEVEL, LOG_MODULE_NAME, USE_COLORS, PROJECT_ALIASES
def _find_project_root(start_path):
@@ -62,37 +62,14 @@ except ImportError as e:
def _normalize_module_name(module_name):
"""Normalize a dotted module name for display in logs.
Strips common prefixes (custom_nodes, project aliases) and __init__
Strips common prefixes (custom_nodes, project aliases) and ``__init__``
suffixes to produce a clean, readable module path.
"""
module_name = str(module_name or "").strip()
if not module_name:
if not module_name or module_name == "__main__":
return _default_module_name
if module_name == "__main__":
return _default_module_name
# If it is a path (contains slashes or is absolute), normalize it to a relative dotted path first
if os.sep in module_name or "/" in module_name or os.path.isabs(module_name):
normalized_path = module_name.replace("/", os.sep).replace("\\", os.sep)
path_no_ext = os.path.splitext(normalized_path)[0]
if _project_root:
try:
relative_path = os.path.relpath(path_no_ext, _project_root)
except ValueError:
relative_path = os.path.basename(path_no_ext)
else:
relative_path = os.path.basename(path_no_ext)
if relative_path == ".":
module_name = ""
else:
module_name = relative_path.replace(os.sep, ".")
if not module_name:
return _default_module_name
parts = [part for part in module_name.split(".") if part]
parts = [p for p in module_name.split(".") if p]
if not parts:
return _default_module_name
@@ -106,7 +83,8 @@ def _normalize_module_name(module_name):
# Strip custom_nodes.<package_name> prefix
if "custom_nodes" in parts:
parts = parts[parts.index("custom_nodes") + 2:]
idx = parts.index("custom_nodes")
parts = parts[idx + 2:] # skip custom_nodes + package folder
# Strip project aliases from config
for alias in PROJECT_ALIASES:
@@ -123,6 +101,15 @@ def _normalize_module_name(module_name):
return cleaned or _default_module_name
# Number of internal call frames between ModuleLogger methods and
# Python's logging.Logger.log(). Used to compute the correct
# ``stacklevel`` so that log records report the *caller's* file and
# line number instead of an internal wrapper.
#
# Call chain: ModuleLogger.method → module-level func → AzLogsLogger.log → logging.Logger.log
_INTERNAL_FRAMES = 3
class ModuleLogger:
"""Per-module logger with a fixed module name.
@@ -130,6 +117,7 @@ class ModuleLogger:
log = create_module_logger(__name__)
log.debug("message")
log.info("message")
log.info("downloaded", extra={"model": "xyz"})
"""
def __init__(self, module_name):
@@ -137,21 +125,21 @@ class ModuleLogger:
def debug(self, *args, **kwargs):
if _initialized:
kwargs.setdefault("stacklevel", 4)
kwargs["stacklevel"] = kwargs.pop("stacklevel", 1) + _INTERNAL_FRAMES
debug(self.module_name, *args, **kwargs)
else:
print(f"[DEBUG] [{self.module_name}]", *args)
def info(self, *args, **kwargs):
if _initialized:
kwargs.setdefault("stacklevel", 4)
kwargs["stacklevel"] = kwargs.pop("stacklevel", 1) + _INTERNAL_FRAMES
info(self.module_name, *args, **kwargs)
else:
print(f"[INFO] [{self.module_name}]", *args)
def warning(self, *args, **kwargs):
if _initialized:
kwargs.setdefault("stacklevel", 4)
kwargs["stacklevel"] = kwargs.pop("stacklevel", 1) + _INTERNAL_FRAMES
warn(self.module_name, *args, **kwargs)
else:
print(f"[WARN] [{self.module_name}]", *args)
@@ -161,18 +149,26 @@ class ModuleLogger:
def error(self, *args, **kwargs):
if _initialized:
kwargs.setdefault("stacklevel", 4)
kwargs["stacklevel"] = kwargs.pop("stacklevel", 1) + _INTERNAL_FRAMES
error(self.module_name, *args, **kwargs)
else:
print(f"[ERROR] [{self.module_name}]", *args)
def exception(self, *args):
def exception(self, *args, **kwargs):
if _initialized:
exception(self.module_name, *args, stacklevel=4)
kwargs["stacklevel"] = kwargs.pop("stacklevel", 1) + _INTERNAL_FRAMES
exception(self.module_name, *args, **kwargs)
else:
print(f"[ERROR] [{self.module_name}]", *args)
traceback.print_exc()
def fatal(self, *args, **kwargs):
if _initialized:
kwargs["stacklevel"] = kwargs.pop("stacklevel", 1) + _INTERNAL_FRAMES
fatal(self.module_name, *args, **kwargs)
else:
print(f"[FATAL] [{self.module_name}]", *args)
def create_module_logger(module_name=None):
"""Create a ModuleLogger with a normalized module name.
@@ -186,5 +182,3 @@ def create_module_logger(module_name=None):
"""
resolved_name = _normalize_module_name(module_name or _default_module_name)
return ModuleLogger(resolved_name)
core/log_system/logger.py
+37 -25
View File
@@ -30,6 +30,7 @@ class LogLevel(IntEnum):
INFO = 20
WARN = 30
ERROR = 40
FATAL = 50
NONE = 100
@@ -39,25 +40,20 @@ LEVEL_MAP = {
LogLevel.INFO: logging.INFO,
LogLevel.WARN: logging.WARNING,
LogLevel.ERROR: logging.ERROR,
LogLevel.FATAL: logging.CRITICAL,
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
"TIME_BG": "\033[48;2;38;63;76;97m", # #263f4c background, white text
"MESSAGE": "\033[96m", # Cyan
"RESET": "\033[0m", # Reset
}
# ANSI reset sequence (all coloring uses LEVEL_THEME RGB values)
ANSI_RESET = "\033[0m"
# RGB color theme per log level
LEVEL_THEME = {
"DEBUG": (155, 89, 182), # #9B59B6
"INFO": (46, 204, 113), # #2ECC71
"WARN": (243, 156, 18), # #F39C12
"ERROR": (192, 57, 43), # #C0392B
"DEBUG": (155, 89, 182), # #9B59B6
"INFO": (46, 204, 113), # #2ECC71
"WARN": (243, 156, 18), # #F39C12
"ERROR": (192, 57, 43), # #C0392B
"FATAL": (231, 76, 60), # #E74C3C — brighter red than ERROR
}
# Default configuration
@@ -246,7 +242,12 @@ class ColoredFormatter(logging.Formatter):
return parts[0], parts[1]
def _display_level_name(self, levelname):
return "WARN" if levelname == "WARNING" else str(levelname or "")
name = str(levelname or "")
if name == "WARNING":
return "WARN"
if name == "CRITICAL":
return "FATAL"
return name
def _format_level(self, levelname):
if self.include_brackets:
@@ -267,27 +268,28 @@ class ColoredFormatter(logging.Formatter):
if not rgb:
return text
r, g, b = rgb
return f"\033[1;97;48;2;{r};{g};{b}m {text} {COLORS['RESET']}"
return f"\033[1;97;48;2;{r};{g};{b}m {text} {ANSI_RESET}"
def _color_timestamp(self, text):
return f"\033[97;48;2;38;63;76m {text} {COLORS['RESET']}"
return f"\033[97;48;2;38;63;76m {text} {ANSI_RESET}"
def _color_root_label(self, text):
return f"\033[97;48;2;38;63;76m {text} {COLORS['RESET']}"
return f"\033[97;48;2;38;63;76m {text} {ANSI_RESET}"
def _color_separator(self, levelname, text):
rgb = LEVEL_THEME.get(levelname)
if not rgb:
return text
r, g, b = rgb
return f"\033[48;2;{r};{g};{b}m{text}{COLORS['RESET']}"
return f"\033[48;2;{r};{g};{b}m{text}{ANSI_RESET}"
def _color_message(self, levelname, text):
rgb = LEVEL_THEME.get(levelname)
if not rgb:
return f"{COLORS['MESSAGE']}{text}{COLORS['RESET']}"
# Fallback: cyan for unknown levels
return f"\033[96m{text}{ANSI_RESET}"
r, g, b = rgb
return f"\033[38;2;{r};{g};{b}m{text}{COLORS['RESET']}"
return f"\033[38;2;{r};{g};{b}m{text}{ANSI_RESET}"
class AzLogsLogger:
@@ -377,9 +379,9 @@ class AzLogsLogger:
try:
os.makedirs(self.config["log_dir"], exist_ok=True)
except OSError as e:
# This is a critical situation, so use print
# This is a fatal situation, so use print
print(
f"[CRITICAL] Could not create log directory: {self.config['log_dir']}. Error: {e}"
f"[FATAL] Could not create log directory: {self.config['log_dir']}. Error: {e}"
)
traceback.print_exc()
# Disable file logging to avoid further errors
@@ -542,15 +544,16 @@ class AzLogsLogger:
# Convert arguments to string
message = " ".join(str(arg) for arg in args)
# Add exception info if provided
# Extract known kwargs
exc_info = kwargs.get("exc_info", None)
stacklevel = kwargs.get("stacklevel", 1)
extra = kwargs.get("extra", 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, stacklevel=stacklevel)
logger.log(log_level, message, exc_info=exc_info, stacklevel=stacklevel, extra=extra)
def debug(self, module, *args, **kwargs):
"""Log at DEBUG level"""
@@ -573,6 +576,10 @@ class AzLogsLogger:
kwargs["exc_info"] = True
self.log(module, LogLevel.ERROR, *args, **kwargs)
def fatal(self, module, *args, **kwargs):
"""Log at FATAL level"""
self.log(module, LogLevel.FATAL, *args, **kwargs)
# Singleton
logger = AzLogsLogger()
@@ -605,6 +612,11 @@ def exception(module, *args, **kwargs):
logger.log(module, LogLevel.ERROR, *args, **kwargs)
def fatal(module, *args, **kwargs):
"""Log at FATAL level"""
logger.log(module, LogLevel.FATAL, *args, **kwargs)
# Function to quickly enable/disable debugging
def set_debug(enabled=True):
"""Enable/disable debugging globally"""
core/path_utils.py
+14 -6
View File
@@ -11,6 +11,18 @@ import hashlib
from typing import Any, Tuple, Optional
_log = None
def _get_log():
"""Return a lazily-created module logger (avoids circular imports)."""
global _log
if _log is None:
from .log_system.log_funcs import create_module_logger
_log = create_module_logger(__name__)
return _log
def get_path_identity(path: str) -> str:
"""Return a stable path identity across symlinks/junctions."""
@@ -178,9 +190,7 @@ def calculate_file_sha256(file_path: str, chunk_size: int = 131072) -> Optional[
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
except Exception as e:
from .log_system.log_funcs import create_module_logger
_log = create_module_logger(__name__)
_log.error(f"Error computing hash for {file_path}: {e}")
_get_log().error(f"Error computing hash for {file_path}: {e}")
return None
@@ -192,9 +202,7 @@ def read_json_safe(file_path: str, default: Any = None) -> Any:
with open(file_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
from .log_system.log_funcs import create_module_logger
_log = create_module_logger(__name__)
_log.warning(f"Error reading JSON from {os.path.basename(file_path)}: {e}")
_get_log().warning(f"Error reading JSON from {os.path.basename(file_path)}: {e}")
return default