87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
import logging
|
|
import os
|
|
from datetime import datetime
|
|
|
|
|
|
class _AnsiColorFormatter(logging.Formatter):
|
|
RESET = "\033[0m"
|
|
MESSAGE_TEXT_REPLACEMENTS = (
|
|
("[PROMPT->AI 前置提示词]", "[AI提示词]"),
|
|
("[PROMPT->AI", "[AI提示词"),
|
|
("[THINK/TOOL_CALL]", "[AI思考-工具调用]"),
|
|
("[THINK/TOOL_RETURN]", "[AI思考-工具返回]"),
|
|
("[THINK/RAW_OUTPUT]", "[AI思考-原始输出]"),
|
|
("[REPLY->CUSTOMER]", "[AI回复客户]"),
|
|
("[ACTIVITY]", "[活动日志]"),
|
|
("[AI质检]", "[AI质检]"),
|
|
)
|
|
MESSAGE_COLOR_RULES = (
|
|
("[PROMPT->AI", "\033[94m"),
|
|
("[THINK/", "\033[96m"),
|
|
("[REPLY->CUSTOMER]", "\033[92m"),
|
|
("Agent 回复", "\033[92m"),
|
|
("[ACTIVITY]", "\033[95m"),
|
|
("[AI质检]", "\033[97m"),
|
|
("收到新消息", "\033[36m"),
|
|
("发送成功", "\033[32m"),
|
|
("防抖等待", "\033[93m"),
|
|
)
|
|
COLORS = {
|
|
logging.DEBUG: "\033[36m",
|
|
logging.INFO: "\033[32m",
|
|
logging.WARNING: "\033[33m",
|
|
logging.ERROR: "\033[31m",
|
|
logging.CRITICAL: "\033[35m",
|
|
}
|
|
|
|
def __init__(self, fmt: str, datefmt: str | None = None, use_color: bool = True):
|
|
super().__init__(fmt=fmt, datefmt=datefmt)
|
|
self.use_color = use_color
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
msg = super().format(record)
|
|
if not self.use_color:
|
|
for old, new in self.MESSAGE_TEXT_REPLACEMENTS:
|
|
msg = msg.replace(old, new)
|
|
return msg
|
|
raw_msg = record.getMessage()
|
|
for old, new in self.MESSAGE_TEXT_REPLACEMENTS:
|
|
msg = msg.replace(old, new)
|
|
for key, color in self.MESSAGE_COLOR_RULES:
|
|
if key in raw_msg:
|
|
return f"{color}{msg}{self.RESET}"
|
|
color = self.COLORS.get(record.levelno, "")
|
|
if not color:
|
|
return msg
|
|
return f"{color}{msg}{self.RESET}"
|
|
|
|
|
|
def setup_logger():
|
|
from logging.handlers import RotatingFileHandler
|
|
from config.config import LOG_DIR, LOG_MAX_BYTES, LOG_BACKUP_COUNT
|
|
|
|
logger = logging.getLogger("cs_agent")
|
|
if getattr(logger, "_cs_logger_configured", False):
|
|
return logger
|
|
logger.setLevel(logging.INFO)
|
|
logger.propagate = False
|
|
fmt = logging.Formatter("[%(asctime)s] %(message)s", datefmt="%H:%M:%S")
|
|
use_color = (os.getenv("LOG_COLOR", "1").lower() in ("1", "true", "yes")) and not bool(os.getenv("NO_COLOR"))
|
|
|
|
ch = logging.StreamHandler()
|
|
ch.setFormatter(_AnsiColorFormatter("[%(asctime)s] %(message)s", datefmt="%H:%M:%S", use_color=use_color))
|
|
logger.addHandler(ch)
|
|
|
|
LOG_DIR.mkdir(exist_ok=True)
|
|
today = datetime.now().strftime("%Y-%m-%d")
|
|
fh = RotatingFileHandler(
|
|
LOG_DIR / f"chat_{today}.log",
|
|
maxBytes=LOG_MAX_BYTES,
|
|
backupCount=LOG_BACKUP_COUNT,
|
|
encoding="utf-8",
|
|
)
|
|
fh.setFormatter(fmt)
|
|
logger.addHandler(fh)
|
|
logger._cs_logger_configured = True
|
|
return logger
|