|
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