request_logging_middleware.py

import datetime
import sys
from typing import Optional

import colorama
import pyramid.config
from loguru import logger
from pyramid.request import Request
from pyramid.response import Response

__set = False

logfile: Optional[str] = None


def logging_middleware_factor(handler, _):
    def tween(request: Request):
        t0 = datetime.datetime.now()
        response: Optional[Response] = None

        # noinspection PyBroadException
        try:
            response: Response = handler(request)
            return response

        except UnicodeEncodeError:
            # Tired of hackers causing 500 errors:
            # UnicodeEncodeError: generic WSGI request
            # Unhandled: 'latin-1' codec can't encode characters in position 19-20: ordinal not in range(256)
            response = create_stand_in_response()
            response.status_code = 400
            response.text = 'No.'
            return response

        finally:
            try:
                if response is None:
                    response = create_stand_in_response()

                dt = datetime.datetime.now() - t0
                ms = int(dt.total_seconds() * 1000)
                msg = get_log_text(request, response, ms)
                logger.info(msg)
            except Exception as e:
                print('Error in logging middleware', e)

    return tween


def create_stand_in_response() -> Response:
    response = Response()
    response.content_length = 0
    response.status_code = 500
    return response


def get_log_text(request: Request, resp: Response, time_in_ms: int) -> str:
    # '11ms code 200 => "GET /episodes/rss HTTP/1.0" [pid: <9>] [12/Jan/2024:13:55:11 -0800] 201.182.16.1 860627
    # bytes from "PodcastAddict/v5 (+https://podcastaddict.com/; Android podcast app)"'

    green = colorama.Fore.GREEN
    yellow = colorama.Fore.YELLOW
    red = colorama.Fore.RED
    purple = colorama.Fore.MAGENTA
    white = colorama.Fore.LIGHTWHITE_EX
    clear = colorama.Fore.RESET

    time_color = green  # Green
    if time_in_ms > 150:
        time_color = yellow  # yellow
    elif time_in_ms > 300:
        time_color = red  # red

    status_color = green  # Green
    if 300 <= resp.status_code < 400:
        status_color = purple
    if 400 <= resp.status_code < 500:
        status_color = yellow
    elif 500 <= resp.status_code < 600:
        status_color = red

    length = 0.0
    len_size = 'bytes'
    bytes_color = white
    if resp.content_length:
        length = float(resp.content_length)

    bytes_color = bytes_color if length < 1024 else (yellow if length < 1024 * 250 else red)

    if length > 1024:
        length = float(length) / 1024.0
        len_size = 'KB'
    if length > 1024:
        length = float(length) / 1024.0
        len_size = 'MB'

    n = datetime.datetime.now()
    time_text = (
        f'{n.year}-{"{:02d}".format(n.month)}-{"{:02d}".format(n.day)} '
        f'{"{:02d}".format(n.hour)}:{"{:02d}".format(n.minute)}:{"{:02d}".format(n.second)}'
    )

    verb_color = green
    if request.method.upper() != 'GET':
        verb_color = purple
    return (
        f'{time_color}{time_in_ms:,}ms{clear} '
        + f'{status_color}HTTP {resp.status_code}{clear} '
        + '=> '
        + f'{verb_color}"{request.method.upper()} {request.path}"{clear} '
        + f'{bytes_color}{length:,.2f} {len_size}{clear} '
        + f'at {white}{time_text}{clear} '
        + f'from {purple}{get_client_ip(request)}{clear} '
        + f'via {yellow}{request.user_agent}{clear}'
    )


def get_client_ip(request: Request):
    if 'X-Real-IP' in request.headers and request.headers['X-Real-IP']:
        return request.headers['X-Real-IP']
    else:
        return request.client_addr  # noqa: FURB126


def configure(config: pyramid.config.Configurator):
    global __set
    if __set:
        return

    # https://loguru.readthedocs.io/en/stable/resources/recipes.html#creating-independent-loggers-with-separate-set-of-handlers
    __set = True
    config.add_tween('.infrastructure.request_logging_tween.logging_middleware_factor')

    logger.level('INFO')
    logger.remove(0)
    color = {'colorize': True, 'format': '<white>{message}</white>'}
    if logfile:
        logger.add(logfile, rotation='1 month', **color)
    else:
        logger.add(sys.stdout, **color)