|
1 | 1 | import logging |
| 2 | +import sys |
2 | 3 |
|
3 | 4 | import colorlog |
4 | 5 |
|
5 | | - |
6 | | -def get_logger(name: str, level: int = logging.INFO) -> logging.Logger: |
7 | | - # Force configure the root logger with a NullHandler to prevent duplicate logs |
8 | | - logging.basicConfig(handlers=[logging.NullHandler()], force=True) |
9 | | - |
10 | | - formatter = colorlog.ColoredFormatter( |
11 | | - "%(white)s%(asctime)s - %(name)s - %(log_color)s%(levelname)s%(reset)s%(white)s - %(message_log_color)s%(message)s", |
12 | | - log_colors={ |
| 6 | +formatter = colorlog.ColoredFormatter( |
| 7 | + "%(white)s%(asctime)s - %(name)s - %(log_color)s%(levelname)s%(reset)s%(white)s - %(message_log_color)s%(message)s", |
| 8 | + log_colors={ |
| 9 | + "DEBUG": "white", |
| 10 | + "INFO": "green", |
| 11 | + "WARNING": "yellow", |
| 12 | + "ERROR": "red", |
| 13 | + "CRITICAL": "red,bg_white", |
| 14 | + }, |
| 15 | + secondary_log_colors={ |
| 16 | + "message": { |
13 | 17 | "DEBUG": "cyan", |
14 | | - "INFO": "green", |
| 18 | + "INFO": "white", |
15 | 19 | "WARNING": "yellow", |
16 | 20 | "ERROR": "red", |
17 | 21 | "CRITICAL": "red,bg_white", |
18 | | - }, |
19 | | - secondary_log_colors={ |
20 | | - "message": { |
21 | | - "DEBUG": "cyan", |
22 | | - "INFO": "blue", |
23 | | - "WARNING": "yellow", |
24 | | - "ERROR": "red", |
25 | | - "CRITICAL": "red,bg_white", |
26 | | - } |
27 | | - }, |
28 | | - ) |
| 22 | + } |
| 23 | + }, |
| 24 | +) |
| 25 | + |
| 26 | + |
| 27 | +class StdOutFilter(logging.Filter): |
| 28 | + def filter(self, record): |
| 29 | + return record.levelno < logging.ERROR |
| 30 | + |
| 31 | + |
| 32 | +class StdErrFilter(logging.Filter): |
| 33 | + def filter(self, record): |
| 34 | + return record.levelno >= logging.ERROR |
| 35 | + |
| 36 | + |
| 37 | +# Create handlers |
| 38 | +stdout_handler = logging.StreamHandler(sys.stdout) # Logs to stdout |
| 39 | +stdout_handler.setFormatter(formatter) |
| 40 | +stdout_handler.addFilter(StdOutFilter()) |
| 41 | + |
| 42 | +stderr_handler = logging.StreamHandler(sys.stderr) # Logs to stderr |
| 43 | +stderr_handler.setFormatter(formatter) |
| 44 | +stderr_handler.addFilter(StdErrFilter()) |
| 45 | + |
| 46 | + |
| 47 | +def get_logger(name: str, level: int = logging.INFO) -> logging.Logger: |
| 48 | + logger = _setup_logger(name, level) |
| 49 | + _setup_exception_logging(logger) |
| 50 | + return logger |
| 51 | + |
| 52 | + |
| 53 | +def _setup_logger(name: str, level: int = logging.INFO) -> logging.Logger: |
| 54 | + # Force configure the root logger with a NullHandler to prevent duplicate logs |
| 55 | + logging.basicConfig(handlers=[logging.NullHandler()], force=True) |
29 | 56 | logger = logging.getLogger(name) |
30 | 57 | if logger.hasHandlers(): |
31 | 58 | for h in logger.handlers: |
32 | 59 | logger.removeHandler(h) |
33 | 60 |
|
34 | | - handler = colorlog.StreamHandler() |
35 | | - handler.setFormatter(formatter) |
36 | | - logger.addHandler(handler) |
| 61 | + logger.addHandler(stdout_handler) |
| 62 | + logger.addHandler(stderr_handler) |
| 63 | + |
37 | 64 | # Ensure the logger propagates to the root logger |
38 | 65 | logger.propagate = True |
39 | 66 | # Set the level on the logger itself |
40 | 67 | logger.setLevel(level) |
41 | 68 | return logger |
| 69 | + |
| 70 | + |
| 71 | +def _setup_exception_logging(logger: logging.Logger) -> None: |
| 72 | + def log_exception(exc_type, exc_value, exc_traceback): |
| 73 | + logger.exception("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)) |
| 74 | + |
| 75 | + # Set the log_exception function as the exception hook |
| 76 | + sys.excepthook = log_exception |
0 commit comments