# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Shared runtime support for the MAFw Click executable.
This module contains root-group behavior, logging and exception display, banner
rendering, and warning-capture setup used by the command modules.
:author: Bulgheroni Antonio
"""
import logging
import sys
import warnings
from enum import IntEnum
from typing import Any
import click
from click.exceptions import ClickException
from rich import traceback
from rich.logging import RichHandler
from rich_pyfiglet import RichFiglet
from mafw.tools.click_extensions import (
AbbreviateGroup,
)
from mafw.tools.parallel import is_free_threading
from mafw.tools.shell_tools import CONSOLE
suppress = [click]
"""Modules omitted from rich traceback locals."""
traceback.install(show_locals=True, suppress=suppress)
[docs]
class MAFwGroup(AbbreviateGroup):
"""Click group with abbreviation and MAFw-specific exit handling."""
group_class: type[click.Group] = AbbreviateGroup
[docs]
def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
"""Parse arguments and record the selected root command before invoke."""
rv = super().parse_args(ctx, args)
if (
ctx.parent is None
): # this is assuring that we store only the first command corresponding to the root context
ctx.meta['_mafw_invoked_subcommand'] = self._detect_command_name(args)
return rv
[docs]
def _detect_command_name(self, args: list[str]) -> str | None:
"""Return the first subcommand token while skipping root option values."""
# This is probably an overkill. After the default parsing the first element in args should be the first
# (abbreviated or fully typed) command. All global options will be removed from args.
# so potentially we could return self.get_command(ctx, args[0]),
# I will leave it like this because it might be useful in the future.
option_params = {opt for param in self.params for opt in getattr(param, 'opts', [])}
expects_value = {
opt for param in self.params if not getattr(param, 'is_flag', False) for opt in getattr(param, 'opts', [])
}
skip_next = False
for token in args:
if skip_next:
skip_next = False
continue
if token in option_params:
if token in expects_value:
skip_next = True
continue
if token.startswith('-'):
continue
selected = self.get_command(click.Context(self), token)
return selected.name if selected is not None else token
return None
[docs]
def invoke(self, ctx: click.Context) -> Any:
"""Invoke the command and normalize Click exceptions to MAFw exits."""
print_banner(ctx)
try:
return super().invoke(ctx)
except ClickException as exc:
exc.show()
sys.exit(exc.exit_code)
except (SystemExit, click.exceptions.Exit):
raise
except Exception:
sys.exit(ReturnValue.Error)
[docs]
def main(self, *args: Any, **kwargs: Any) -> None: # type: ignore[override]
"""Run the CLI and convert returned statuses to process exit codes."""
try:
rv = super().main(*args, standalone_mode=False, **kwargs) # type: ignore[call-overload]
if isinstance(rv, (ReturnValue, int)):
sys.exit(rv)
sys.exit(ReturnValue.OK)
except ClickException as exc:
exc.show()
sys.exit(exc.exit_code)
except SystemExit:
raise
except Exception:
sys.exit(ReturnValue.Error)
LEVELS = {'debug': 10, 'info': 20, 'warning': 30, 'error': 40, 'critical': 50}
"""Numeric logging levels accepted by the root CLI option."""
CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
"""Click settings shared by the MAFw root group."""
_warnings_captured = False
"""Whether logging has installed the warning capture hook."""
welcome_message = RichFiglet(
'MAFw',
colors=['#ff0000', 'magenta1', 'blue3'],
horizontal=True,
# font='banner4',
remove_blank_lines=True,
border='ROUNDED',
border_color='#ff0000',
)
"""Styled welcome banner rendered for rich command invocations."""
[docs]
def print_banner(ctx: click.Context) -> None:
"""
Print the welcome banner only once, only for rich UI,
and only if not disabled.
.. note::
ctx.obj is not yet populated when Group.invoke() is called,
but ctx.params contains parsed options for the current group. We
therefore inspect the root context params.
"""
# avoid printing while click is still doing resilient parsing (e.g., --help)
if getattr(ctx, 'resilient_parsing', False):
return
root = ctx.find_root()
# Parse-time command resolution records the selected subcommand before
# invoke() emits any output, so completion can remain quiet.
# Possible improvements: allow selection of commands for which there must not be any banner printing.
if root.meta.get('_mafw_invoked_subcommand') == 'completion':
return
params = getattr(root, 'params', {}) or {}
ui = params.get('ui')
no_banner = params.get('no_banner')
# Normalize and check
if no_banner:
return
if not ui: # pragma: no cover
# if nothing available, be conservative and don't print
return
if str(ui).lower() != 'rich':
return
# Print only once per process / invocation
if root.meta.get('_banner_printed', False):
return
console = CONSOLE
console.print(welcome_message)
root.meta['_banner_printed'] = True
[docs]
def is_bugged_warning_capture_version() -> bool:
"""Check if the Python version has the warning capture bug in free-threading mode.
Python 3.14.0 through 3.14.3 (free-threading builds only) have a bug where
``logging.captureWarnings(True)`` can crash or misbehave. This function returns
True when the current interpreter is affected.
.. note::
See https://github.com/python/cpython/pull/146374 for details.
.. todo::
Remove this workaround when Python 3.14 is no longer a supported version
(expected late 2030).
:return: True if the interpreter is affected by the warning capture bug, False otherwise.
:rtype: bool
"""
version = sys.version_info
return version.major == 3 and version.minor == 14 and version.micro <= 3 and is_free_threading()
warnings.formatwarning = custom_formatwarning
# Disable warning capture on bugged free-threading builds (3.14.0–3.14.3).
if is_bugged_warning_capture_version():
logging.captureWarnings(False)
else:
logging.captureWarnings(True)
# get the root logger
log = logging.getLogger()
"""Root logger configured by the CLI options."""
[docs]
class ReturnValue(IntEnum):
"""Enumerator to handle the script return value."""
OK = 0
"""No error"""
Error = 1
"""Generic error"""
[docs]
def logger_setup(level: str, ui: str, tracebacks: bool) -> None:
"""Set up the logger.
This function is actually configuring the root logger level from the command line options and it attaches either
a RichHandler or a StreamHandler depending on the user interface type.
The `tracebacks` flag is used only by the RichHandler. Printing the tracebacks is rather useful when debugging
the code, but it could be detrimental for final users. In normal circumstances, tracebacks is set to False,
and is turned on when the debug flag is activated.
:param level: Logging level as a string.
:type level: str
:param ui: User interface as a string ('rich' or 'console').
:type ui: str
:param tracebacks: Enable/disable the logging of exception tracebacks.
"""
level = level.lower()
ui = ui.lower()
log.setLevel(LEVELS[level])
handler: logging.Handler
if ui == 'rich':
fs = '%(message)s'
handler = RichHandler(
rich_tracebacks=tracebacks, markup=True, show_path=False, log_time_format='%Y%m%d-%H:%M:%S'
)
else:
fs = '%(asctime)s - %(levelname)s - %(message)s'
handler = logging.StreamHandler()
formatter = logging.Formatter(fs)
handler.setFormatter(formatter)
log.addHandler(handler)
# _ensure_warning_capture()
[docs]
def display_exception(exception: Exception, show_traceback: bool = False) -> None:
"""
Display exception information with optional debug details.
This function logs exception information at the critical level. When show_traceback is enabled,
it logs the full exception including traceback information. Otherwise, it logs a simplified
message directing users to enable debug mode for more details.
:param exception: The exception to be displayed and logged.
:type exception: Exception
:param show_traceback: Flag indicating whether to show detailed traceback information. Defaults to False
:type show_traceback: bool
"""
if show_traceback:
log.critical('A critical error occurred')
log.exception(exception)
else:
log.critical('A critical error occurred. Set option -D to get traceback output')
log.exception(f'{exception.__class__.__name__}: {exception}', exc_info=False, stack_info=False, stacklevel=1)