ansible-doctor/ansibledoctor/config.py

#!/usr/bin/env python3
"""Global settings definition."""

import logging
import os
import re
from io import StringIO

import colorama
import structlog
from appdirs import AppDirs
from dynaconf import Dynaconf, ValidationError, Validator

import ansibledoctor.exception
from ansibledoctor.utils import Singleton


class Config:
    """Create configuration object."""

    ANNOTATIONS = {
        "meta": {
            "name": "meta",
            "automatic": True,
            "subtypes": ["value"],
            "allow_multiple": False,
        },
        "todo": {
            "name": "todo",
            "automatic": True,
            "subtypes": ["value"],
            "allow_multiple": True,
        },
        "var": {
            "name": "var",
            "automatic": True,
            "subtypes": ["value", "example", "description", "type", "deprecated"],
            "allow_multiple": False,
        },
        "example": {
            "name": "example",
            "automatic": True,
            "subtypes": [],
            "allow_multiple": False,
        },
        "tag": {
            "name": "tag",
            "automatic": True,
            "subtypes": ["value", "description"],
            "allow_multiple": False,
        },
    }

    def __init__(self):
        self.config_files = [
            os.path.join(AppDirs("ansible-doctor").user_config_dir, "config.yml"),
            ".ansibledoctor",
            ".ansibledoctor.yml",
            ".ansibledoctor.yaml",
        ]
        self.config_merge = True
        self.args = {}
        self.load()

    def load(self, root_path=None, args=None):
        tmpl_src = os.path.join(os.path.dirname(os.path.realpath(__file__)), "templates")
        tmpl_provider = ["local", "git"]

        if args:
            if args.get("config_file"):
                self.config_merge = False
                self.config_files = [os.path.abspath(args.get("config_file"))]
                args.pop("config_file")

            self.args = args

        self.config = Dynaconf(
            envvar_prefix="ANSIBLE_DOCTOR",
            merge_enabled=self.config_merge,
            core_loaders=["YAML"],
            root_path=root_path,
            settings_files=self.config_files,
            fresh_vars=["base_dir", "output_dir"],
            validators=[
                Validator(
                    "base_dir",
                    default=os.getcwd(),
                    apply_default_on_none=True,
                    is_type_of=str,
                ),
                Validator(
                    "dry_run",
                    default=False,
                    is_type_of=bool,
                ),
                Validator(
                    "recursive",
                    default=False,
                    is_type_of=bool,
                ),
                Validator(
                    "exclude_files",
                    default=[],
                    is_type_of=list,
                ),
                Validator(
                    "exclude_tags",
                    default=[],
                    is_type_of=list,
                ),
                Validator(
                    "role.name",
                    is_type_of=str,
                ),
                Validator(
                    "role.autodetect",
                    default=True,
                    is_type_of=bool,
                ),
                Validator(
                    "logging.level",
                    default="WARNING",
                    is_in=[
                        "DEBUG",
                        "INFO",
                        "WARNING",
                        "ERROR",
                        "CRITICAL",
                        "debug",
                        "info",
                        "warning",
                        "error",
                        "critical",
                    ],
                ),
                Validator(
                    "logging.json",
                    default=False,
                    is_type_of=bool,
                ),
                Validator(
                    "recursive",
                    default=False,
                    is_type_of=bool,
                ),
                Validator(
                    "template.src",
                    default=f"local>{tmpl_src}",
                    is_type_of=str,
                    condition=lambda x: re.match(r"^(local|git)\s*>\s*", x),
                    messages={
                        "condition": f"Template provider must be one of {tmpl_provider}.",
                    },
                ),
                Validator(
                    "template.name",
                    default="readme",
                    is_type_of=str,
                ),
                Validator(
                    "template.options.tabulate_variables",
                    default=False,
                    is_type_of=bool,
                ),
                Validator(
                    "renderer.autotrim",
                    default=True,
                    is_type_of=bool,
                ),
                Validator(
                    "renderer.include_header",
                    default="",
                    is_type_of=str,
                ),
                Validator(
                    "renderer.dest",
                    default=os.path.relpath(os.getcwd()),
                    is_type_of=str,
                ),
                Validator(
                    "renderer.force_overwrite",
                    default=False,
                    is_type_of=bool,
                ),
            ],
        )

        self.validate()

        # Override correct log level from argparse
        levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
        log_level = levels.index(self.config.logging.level.upper())
        if self.args.get("logging.level") and isinstance(self.args["logging.level"], list):
            for lvl in self.args["logging.level"]:
                log_level = min(len(levels) - 1, max(log_level + lvl, 0))

        self.args["logging__level"] = levels[log_level]

        if root_path:
            self.args["base_dir"] = root_path

        self.config.update(self.args)
        self.validate()

        self._init_logger()

    def validate(self):
        try:
            self.config.validators.validate_all()
        except ValidationError as e:
            raise ansibledoctor.exception.ConfigError("Configuration error", e.message) from e

    def is_role(self):
        self.config.role_name = self.config.get(
            "role_name", os.path.basename(self.config.base_dir)
        )
        return os.path.isdir(os.path.join(self.config.base_dir, "tasks"))

    def get_annotations_definition(self, automatic=True):
        annotations = {}
        if automatic:
            for k, item in self.ANNOTATIONS.items():
                if item.get("automatic"):
                    annotations[k] = item
        return annotations

    def get_annotations_names(self, automatic=True):
        annotations = []
        if automatic:
            for k, item in self.ANNOTATIONS.items():
                if item.get("automatic"):
                    annotations.append(k)
        return annotations

    def _init_logger(self):
        styles = structlog.dev.ConsoleRenderer.get_default_level_styles()
        styles["debug"] = colorama.Fore.BLUE

        processors = [
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.StackInfoRenderer(),
            structlog.dev.set_exc_info,
            structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        ]

        if self.config.logging.json:
            processors.append(ErrorStringifier())
            processors.append(structlog.processors.JSONRenderer())
        else:
            processors.append(MultilineConsoleRenderer(level_styles=styles))

        try:
            structlog.configure(
                processors=processors,
                wrapper_class=structlog.make_filtering_bound_logger(
                    logging.getLevelName(self.config.get("logging.level")),
                ),
            )
            structlog.contextvars.unbind_contextvars()
        except KeyError as e:
            raise ansibledoctor.exception.ConfigError(f"Can not set log level: {e!s}") from e


class ErrorStringifier:
    """A processor that converts exceptions to a string representation."""

    def __call__(self, _, __, event_dict):
        if "error" not in event_dict:
            return event_dict

        err = event_dict.get("error")

        if isinstance(err, Exception):
            event_dict["error"] = f"{err.__class__.__name__}: {err}"

        return event_dict


class MultilineConsoleRenderer(structlog.dev.ConsoleRenderer):
    """A processor for printing multiline strings."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def __call__(self, _, __, event_dict):
        err = None

        if "error" in event_dict:
            err = event_dict.pop("error")

        event_dict = super().__call__(_, __, event_dict)

        if not err:
            return event_dict

        sio = StringIO()
        sio.write(event_dict)

        if isinstance(err, Exception):
            sio.write(
                f"\n{colorama.Fore.RED}{err.__class__.__name__}:"
                f"{colorama.Style.RESET_ALL} {str(err).strip()}"
            )
        else:
            sio.write(f"\n{err.strip()}")

        return sio.getvalue()


class SingleConfig(Config, metaclass=Singleton):
    """Singleton config class."""

    pass