Source code for prozorro_sale.tools.logger

"""
Package to provide a logging with specific output accepted for CBD3.
"""
import logging
import os
from logging import config, LogRecord
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse

import yaml
from aiohttp.abc import AbstractAccessLogger, BaseRequest
from aiohttp.web_log import AccessLogger

from prozorro_sale.tools.context_variables import REQUEST_ID, PROCEDURE_ID, AUCTION_ID
from prozorro_sale.tools.environment import path_discover

CONFIG_PATH = os.getenv("LOGGING_CONFIG_PATH", '/config/logging.yml')

FILTER = None


class RequestIdFilter(logging.Filter):
    """Logger filter for request id"""

    def filter(self, record: LogRecord) -> int:
        request_id = REQUEST_ID.get('unknown')
        record.x_request_id = request_id
        return True


class ProcedureIdFilter(logging.Filter):
    """Logger filter for procedure id"""

    def filter(self, record: logging.LogRecord) -> bool:
        if procedure_id := PROCEDURE_ID.get():
            record.procedure_id = procedure_id
        return True


class AuctionIdFilter(logging.Filter):
    """Logger filter for procedure id"""

    def filter(self, record: logging.LogRecord) -> bool:
        if auction_id := AUCTION_ID.get():
            record.auction_id = auction_id
        return True


def get_filter():
    """Get logger filter class"""
    global FILTER
    if not FILTER:
        FILTER = RequestIdFilter()
    return FILTER


def get_custom_logger(name):
    """Get logger custom filter class"""
    logger = logging.getLogger(name)
    logger.addFilter(get_filter())
    return logger


LOG = get_custom_logger(__name__)

FILTER_MAPPING = {
    'requestid': RequestIdFilter,
    'procedureid': ProcedureIdFilter,
    'auctionid': AuctionIdFilter,
}


def configure_logging(config_path=None):
    """Configure logger from config file.

    Raises:
        Exception: Failed to configure logging from file
    """
    log_level = os.getenv('LOGGING_LEVEL', 'INFO')
    config_path = config_path or CONFIG_PATH
    config_path = path_discover(config_path)
    try:
        with open(config_path, 'r') as c:
            log_config = yaml.safe_load(c)
            log_config['logging']['root']['level'] = log_level
            out_filters = {}
            for filter_ in log_config['logging'].get('filters', []):
                if filter_ in FILTER_MAPPING:
                    out_filters.update({
                        filter_: {
                            '()': FILTER_MAPPING[filter_]
                        }
                    })
            log_config['logging']['filters'] = out_filters
            config.dictConfig(log_config['logging'])
        return
    except Exception as ex:
        LOG.exception(f'Failed to configure logging from file cause of {ex}')

    log_format = '%(asctime)s | %(name)s | %(levelname)s - %(message)s'
    logging.basicConfig(level=log_level, format=log_format, datefmt='%Y-%m-%dT%H:%M:%S %z')


class CustomAccessLogger(AbstractAccessLogger):
    """Custom Access Logger.
    """

    @staticmethod
    def _format_r(request: BaseRequest) -> str:
        """Format log message output.

        Args:
            request (BaseRequest)

        Returns:
            str: log message
        """
        if request is None:
            return '-'
        path_qs = urlparse(request.path_qs)
        query = parse_qs(path_qs.query, keep_blank_values=True)
        if 'acc_token' in query:
            query['acc_token'] = '#censored#'
        path_qs = path_qs._replace(query=urlencode(query, True))
        return '%s %s HTTP/%s.%s' % (request.method, urlunparse(path_qs),
                                     request.version.major,
                                     request.version.minor)

    def log(self, request, response, time):
        """logging method.

        Args:
            request (aiohttp.web.Request object)
            response (aiohttp.web.Response object)
            time (float) – Time taken to serve the request.

        """
        remote = request.headers.get('X-Forwarded-For', request.remote)
        request_info = self._format_r(request)
        refer = request.headers.get('Referer', "-")
        user_agent = request.headers.get('User-Agent', "-")
        time = AccessLogger._format_Tf(request, response, time)
        self.logger.info(f"\"{remote} {request_info} {response.status} "
                         f"{response.body_length} {refer} {user_agent} {time} \"")


class LogFormatter(logging.Formatter):
    """Log formatter.
    """

    def formatMessage(self, record: LogRecord) -> str:
        """
        Format output log message.

        Args:
            record (LogRecord)
        """
        record_dict = record.__dict__
        origin_fmt = self._style._fmt
        custom_attrs = [
            ('PROCEDURE-ID', 'procedure_id'),
            ('AUCTION-ID', 'auction_id'),
            ('X-Request-ID', 'x_request_id'),
        ]
        for title, param in custom_attrs:
            formatted_param = f"%({param})s"
            if record_dict.get(param) and formatted_param not in origin_fmt:
                self._style._fmt += f" | {title}: {formatted_param}"
        styled_mes = super().formatMessage(record)
        self._style._fmt = origin_fmt
        return styled_mes