copy from main tesi folder
This commit is contained in:
43
utils/ColoredFormatter.py
Normal file
43
utils/ColoredFormatter.py
Normal file
@@ -0,0 +1,43 @@
|
||||
import logging
|
||||
|
||||
class ColoredFormatter(logging.Formatter):
|
||||
"""Colored formatter for the logging package."""
|
||||
|
||||
def __init__(
|
||||
self, fmt=None, datefmt=None, style="%", validate=True, *, defaults=None
|
||||
):
|
||||
"""Colored formatter for the logging package."""
|
||||
fmt = fmt or "%(levelname)s: %(message)s"
|
||||
super().__init__(fmt, datefmt, style, validate, defaults=defaults)
|
||||
colors = {
|
||||
"red": "\x1b[31;20m",
|
||||
"bold_red": "\x1b[31;1m",
|
||||
"green": "\x1b[32;20m",
|
||||
"bold_green": "\x1b[32;1m",
|
||||
"yellow": "\x1b[33;20m",
|
||||
"bold_yellow": "\x1b[33;1m",
|
||||
"blue": "\x1b[34;20m",
|
||||
"bold_blue": "\x1b[34;1m",
|
||||
"grey": "\x1b[37;20m",
|
||||
"bold_grey": "\x1b[37;1m",
|
||||
"reset": "\x1b[0m",
|
||||
}
|
||||
self._default_formatter = logging.Formatter(fmt)
|
||||
self._formatters = {
|
||||
100: logging.Formatter(colors["bold_blue"] + fmt + colors["reset"]),
|
||||
logging.DEBUG: logging.Formatter(colors["grey"] + fmt + colors["reset"]),
|
||||
logging.INFO: logging.Formatter(colors["green"] + fmt + colors["reset"]),
|
||||
logging.WARNING: logging.Formatter(
|
||||
colors["yellow"] + fmt + colors["reset"]
|
||||
),
|
||||
logging.ERROR: logging.Formatter(colors["red"] + fmt + colors["reset"]),
|
||||
logging.CRITICAL: logging.Formatter(
|
||||
colors["bold_red"] + fmt + colors["reset"]
|
||||
),
|
||||
}
|
||||
|
||||
def format(self, record):
|
||||
"""Override of logging.Formatter.format"""
|
||||
return self._formatters.get(record.levelno, self._default_formatter).format(
|
||||
record
|
||||
)
|
24
utils/PostHandler.py
Normal file
24
utils/PostHandler.py
Normal file
@@ -0,0 +1,24 @@
|
||||
import json
|
||||
import logging
|
||||
from http.server import BaseHTTPRequestHandler
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def __init__(self, custom_func, driver, logger, *args, **kwargs):
|
||||
self._custom_func = custom_func
|
||||
self.logger = logger
|
||||
self.driver = driver
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def do_POST(self):
|
||||
if self.path == '/webrtc-internals-exporter':
|
||||
content_length = int(self.headers['Content-Length'])
|
||||
post_data = self.rfile.read(content_length)
|
||||
self.logger.log(logging.DEBUG, f"POST request,\nPath: {self.path}\nHeaders:\n{self.headers}\n\nBody:\n{post_data.decode('utf-8')}")
|
||||
self._custom_func(self.driver, json.loads(post_data.decode('utf-8')))
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'POST request received')
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'404 Not Found')
|
Reference in New Issue
Block a user