michaelschiemer/deployment/ansible/callback_plugins/default_with_clean_msg.py

# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
# (c) 2017 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# Modified to add clean multiline msg formatting
#
# This plugin extends the default callback with enhanced multiline message formatting.
# It suppresses warnings by implementing its own versions of methods that would
# otherwise call get_option() before options are initialized.

from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

DOCUMENTATION = '''
    name: default_with_clean_msg
    type: stdout
    short_description: default Ansible screen output with clean multiline messages
    version_added: historical
    description:
        This is the default output callback for ansible-playbook with enhanced
        multiline message formatting. Multiline messages in msg fields are
        displayed as clean, readable blocks instead of escaped newline characters.
'''

from ansible import constants as C
from ansible.plugins.callback.default import CallbackModule as DefaultCallbackModule


class CallbackModule(DefaultCallbackModule):
    '''
    This extends the default callback with enhanced multiline message formatting.
    Multiline messages in 'msg' fields are displayed as clean, readable blocks.
    '''

    CALLBACK_NAME = 'default_with_clean_msg'

    def _print_clean_msg(self, result, color=C.COLOR_VERBOSE):
        '''
        Print multiline messages in a clean, readable format with borders.
        This makes it easy to read and copy multiline content from debug outputs.
        '''
        msg_body = result._result.get('msg')
        if isinstance(msg_body, str) and msg_body.strip():
            lines = msg_body.strip().splitlines()
            if len(lines) > 1:  # Only format if multiline
                max_len = max(len(line) for line in lines if line.strip())
                if max_len > 0:
                    border = "=" * min(max_len, 80)  # Limit border width
                    self._display.display("\n" + border, color=color)
                    for line in lines:
                        self._display.display(line, color=color)
                    self._display.display(border + "\n", color=color)

    def v2_playbook_on_task_start(self, task, is_conditional):
        # Suppress warnings by implementing our own version that doesn't call get_option early
        # Initialize state if needed
        if not hasattr(self, '_play'):
            self._play = None
        if not hasattr(self, '_last_task_banner'):
            self._last_task_banner = None
        if not hasattr(self, '_task_type_cache'):
            self._task_type_cache = {}
        
        # Cache task prefix
        self._task_type_cache[task._uuid] = 'TASK'
        
        # Store task name
        if self._play and hasattr(self._play, 'strategy'):
            from ansible.utils.fqcn import add_internal_fqcns
            if self._play.strategy in add_internal_fqcns(('free', 'host_pinned')):
                self._last_task_name = None
            else:
                self._last_task_name = task.get_name().strip()
        else:
            self._last_task_name = task.get_name().strip()
        
        # Print task banner (only if we should display it)
        # We skip the parent's check for display_skipped_hosts/display_ok_hosts to avoid warnings
        if self._play and hasattr(self._play, 'strategy'):
            from ansible.utils.fqcn import add_internal_fqcns
            if self._play.strategy not in add_internal_fqcns(('free', 'host_pinned')):
                self._last_task_banner = task._uuid
                self._display.banner('TASK [%s]' % task.get_name().strip())

    def v2_runner_on_start(self, host, task):
        # Suppress warnings by not calling parent if options aren't ready
        # This method is optional and only shows per-host start messages
        # We can safely skip it to avoid warnings
        pass

    def v2_runner_on_ok(self, result):
        # Suppress warnings by implementing our own version that doesn't call get_option
        host_label = self.host_label(result)
        
        # Handle TaskInclude separately
        from ansible.playbook.task_include import TaskInclude
        if isinstance(result._task, TaskInclude):
            if self._last_task_banner != result._task._uuid:
                self.v2_playbook_on_task_start(result._task, False)
            return
        
        # Clean results and handle warnings
        self._clean_results(result._result, result._task.action)
        self._handle_warnings(result._result)
        
        # Handle loop results
        if result._task.loop and 'results' in result._result:
            self._process_items(result)
            return
        
        # Determine status and color
        if result._result.get('changed', False):
            if self._last_task_banner != result._task._uuid:
                self.v2_playbook_on_task_start(result._task, False)
            self._display.display("changed: [%s]" % host_label, color=C.COLOR_CHANGED)
            color = C.COLOR_CHANGED
        else:
            # Always display ok hosts (skip get_option check to avoid warnings)
            if self._last_task_banner != result._task._uuid:
                self.v2_playbook_on_task_start(result._task, False)
            self._display.display("ok: [%s]" % host_label, color=C.COLOR_OK)
            color = C.COLOR_OK
        
        # Add our clean message formatting
        self._print_clean_msg(result, color=color)

    def v2_runner_on_failed(self, result, ignore_errors=False):
        # Suppress warnings by implementing our own version
        host_label = self.host_label(result)
        self._clean_results(result._result, result._task.action)
        
        if self._last_task_banner != result._task._uuid:
            self.v2_playbook_on_task_start(result._task, False)
        
        self._handle_exception(result._result, use_stderr=False)
        self._handle_warnings(result._result)
        
        if result._task.loop and 'results' in result._result:
            self._process_items(result)
        else:
            msg = "fatal: [%s]: FAILED! => %s" % (host_label, self._dump_results(result._result))
            self._display.display(msg, color=C.COLOR_ERROR)
        
        if ignore_errors:
            self._display.display("...ignoring", color=C.COLOR_SKIP)
        
        # Add our clean message formatting
        self._print_clean_msg(result, color=C.COLOR_ERROR)

    def v2_runner_on_skipped(self, result):
        # Suppress warnings by implementing our own version
        # Always display skipped hosts (skip get_option check to avoid warnings)
        self._clean_results(result._result, result._task.action)
        
        if self._last_task_banner != result._task._uuid:
            self.v2_playbook_on_task_start(result._task, False)
        
        if result._task.loop is not None and 'results' in result._result:
            self._process_items(result)
        else:
            msg = "skipping: [%s]" % result._host.get_name()
            if self._run_is_verbose(result):
                msg += " => %s" % self._dump_results(result._result)
            self._display.display(msg, color=C.COLOR_SKIP)
        
        # Add our clean message formatting
        self._print_clean_msg(result, color=C.COLOR_SKIP)

    def v2_runner_on_unreachable(self, result):
        # Suppress warnings by implementing our own version
        if self._last_task_banner != result._task._uuid:
            self.v2_playbook_on_task_start(result._task, False)
        
        host_label = self.host_label(result)
        msg = "fatal: [%s]: UNREACHABLE! => %s" % (host_label, self._dump_results(result._result))
        self._display.display(msg, color=C.COLOR_UNREACHABLE)
        
        if result._task.ignore_unreachable:
            self._display.display("...ignoring", color=C.COLOR_SKIP)
        
        # Add our clean message formatting
        self._print_clean_msg(result, color=C.COLOR_UNREACHABLE)