|
|
import logging |
|
|
import re |
|
|
import sys |
|
|
from datetime import timedelta |
|
|
from pathlib import Path |
|
|
|
|
|
import sentry_sdk.integrations.logging as sentry |
|
|
from loguru import logger as _logger |
|
|
|
|
|
from hibiapi.utils.config import Config |
|
|
|
|
|
LOG_FILE = Config["log"]["file"].get_optional(Path) |
|
|
LOG_LEVEL = Config["log"]["level"].as_str().strip().upper() |
|
|
LOG_FORMAT = Config["log"]["format"].as_str().strip() |
|
|
|
|
|
|
|
|
class LoguruHandler(logging.Handler): |
|
|
_tag_escape_re = re.compile(r"</?((?:[fb]g\s)?[^<>\s]*)>") |
|
|
|
|
|
@classmethod |
|
|
def escape_tag(cls, string: str) -> str: |
|
|
return cls._tag_escape_re.sub(r"\\\g<0>", string) |
|
|
|
|
|
def emit(self, record: logging.LogRecord): |
|
|
try: |
|
|
level = logger.level(record.levelname).name |
|
|
except ValueError: |
|
|
level = record.levelno |
|
|
|
|
|
frame, depth, message = logging.currentframe(), 2, record.getMessage() |
|
|
while frame.f_code.co_filename == logging.__file__: |
|
|
frame = frame.f_back |
|
|
depth += 1 |
|
|
|
|
|
logger.opt(depth=depth, exception=record.exc_info, colors=True).log( |
|
|
level, f"<e>{self.escape_tag(message)}</e>" |
|
|
) |
|
|
|
|
|
|
|
|
logger = _logger.opt(colors=True) |
|
|
logger.remove() |
|
|
logger.add( |
|
|
sys.stdout, |
|
|
level=LOG_LEVEL, |
|
|
format=LOG_FORMAT, |
|
|
filter=lambda record: record["level"].no < logging.WARNING, |
|
|
) |
|
|
logger.add( |
|
|
sys.stderr, |
|
|
level=LOG_LEVEL, |
|
|
filter=lambda record: record["level"].no >= logging.WARNING, |
|
|
format=LOG_FORMAT, |
|
|
) |
|
|
logger.add(sentry.BreadcrumbHandler(), level=LOG_LEVEL) |
|
|
logger.add(sentry.EventHandler(), level="ERROR") |
|
|
|
|
|
if LOG_FILE is not None: |
|
|
LOG_FILE.parent.mkdir(parents=True, exist_ok=True) |
|
|
|
|
|
logger.add( |
|
|
str(LOG_FILE), |
|
|
level=LOG_LEVEL, |
|
|
encoding="utf-8", |
|
|
rotation=timedelta(days=1), |
|
|
) |
|
|
|
|
|
logger.level(LOG_LEVEL) |
|
|
|
