129 lines
3.6 KiB
Python
129 lines
3.6 KiB
Python
from functools import partial
|
|
import logging
|
|
import os
|
|
import time
|
|
import typing as t
|
|
|
|
import loguru
|
|
import loguru._logger
|
|
from chatchat.settings import Settings
|
|
|
|
|
|
def build_logger(log_file: str = "chatchat"):
|
|
"""
|
|
build a logger with colorized output and a log file, for example:
|
|
|
|
logger = build_logger("api")
|
|
logger.info("<green>some message</green>")
|
|
|
|
user can set basic_settings.log_verbose=True to output debug logs
|
|
"""
|
|
logger = loguru.logger.opt(colors=True)
|
|
logger.opt = partial(loguru.logger.opt, colors=True)
|
|
|
|
if log_file:
|
|
if not log_file.endswith(".log"):
|
|
log_file = f"{log_file}.log"
|
|
if not os.path.isabs(log_file):
|
|
log_file = str((Settings.basic_settings.LOG_PATH / log_file).resolve())
|
|
logger.add(log_file, colorize=False)
|
|
|
|
logger.error = logger.opt(exception=True).error
|
|
|
|
_debug = logger.debug
|
|
|
|
def debug(*args, **kwds):
|
|
if (Settings.basic_settings.log_verbose
|
|
and _debug is not debug):
|
|
_debug(*args, **kwds)
|
|
|
|
logger.debug = debug
|
|
|
|
return logger
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LoggerNameFilter(logging.Filter):
|
|
def filter(self, record):
|
|
# return record.name.startswith("{}_core") or record.name in "ERROR" or (
|
|
# record.name.startswith("uvicorn.error")
|
|
# and record.getMessage().startswith("Uvicorn running on")
|
|
# )
|
|
return True
|
|
|
|
|
|
def get_log_file(log_path: str, sub_dir: str):
|
|
"""
|
|
sub_dir should contain a timestamp.
|
|
"""
|
|
log_dir = os.path.join(log_path, sub_dir)
|
|
# Here should be creating a new directory each time, so `exist_ok=False`
|
|
os.makedirs(log_dir, exist_ok=False)
|
|
return os.path.join(log_dir, f"{sub_dir}.log")
|
|
|
|
|
|
def get_config_dict(
|
|
log_level: str, log_file_path: str, log_backup_count: int, log_max_bytes: int
|
|
) -> dict:
|
|
# for windows, the path should be a raw string.
|
|
log_file_path = (
|
|
log_file_path.encode("unicode-escape").decode()
|
|
if os.name == "nt"
|
|
else log_file_path
|
|
)
|
|
log_level = log_level.upper()
|
|
config_dict = {
|
|
"version": 1,
|
|
"disable_existing_loggers": False,
|
|
"formatters": {
|
|
"formatter": {
|
|
"format": (
|
|
"%(asctime)s %(name)-12s %(process)d %(levelname)-8s %(message)s"
|
|
)
|
|
},
|
|
},
|
|
"filters": {
|
|
"logger_name_filter": {
|
|
"()": __name__ + ".LoggerNameFilter",
|
|
},
|
|
},
|
|
"handlers": {
|
|
"stream_handler": {
|
|
"class": "logging.StreamHandler",
|
|
"formatter": "formatter",
|
|
"level": log_level,
|
|
# "stream": "ext://sys.stdout",
|
|
# "filters": ["logger_name_filter"],
|
|
},
|
|
"file_handler": {
|
|
"class": "logging.handlers.RotatingFileHandler",
|
|
"formatter": "formatter",
|
|
"level": log_level,
|
|
"filename": log_file_path,
|
|
"mode": "a",
|
|
"maxBytes": log_max_bytes,
|
|
"backupCount": log_backup_count,
|
|
"encoding": "utf8",
|
|
},
|
|
},
|
|
"loggers": {
|
|
"chatchat_core": {
|
|
"handlers": ["stream_handler", "file_handler"],
|
|
"level": log_level,
|
|
"propagate": False,
|
|
}
|
|
},
|
|
"root": {
|
|
"level": log_level,
|
|
"handlers": ["stream_handler", "file_handler"],
|
|
},
|
|
}
|
|
return config_dict
|
|
|
|
|
|
def get_timestamp_ms():
|
|
t = time.time()
|
|
return int(round(t * 1000))
|