tw2/qingjian_cs/app/logger.py

import logging
import sys


class _StreamColorizer:
    RESET = "\033[0m"
    C_AI_THINK = "\033[33m"   # yellow
    C_INBOUND = "\033[36m"    # cyan
    C_OUTBOUND = "\033[32m"   # green

    def __init__(self, stream):
        self.stream = stream

    def write(self, data):
        if not data:
            return
        out = str(data)
        if "Unsupported block type thinking" in out:
            return
        if "\x1b[" in out:
            self.stream.write(out)
            return
        if "(thinking):" in out:
            out = f"{self.C_AI_THINK}{out}{self.RESET}"
        elif "[收消息]" in out:
            out = f"{self.C_INBOUND}{out}{self.RESET}"
        elif "[发送]" in out:
            out = f"{self.C_OUTBOUND}{out}{self.RESET}"
        self.stream.write(out)

    def flush(self):
        self.stream.flush()

    def isatty(self):
        return getattr(self.stream, "isatty", lambda: False)()

    @property
    def encoding(self):
        return getattr(self.stream, "encoding", "utf-8")


_stream_color_installed = False


def install_stream_colorizer() -> None:
    global _stream_color_installed
    if _stream_color_installed:
        return
    try:
        sys.stdout = _StreamColorizer(sys.stdout)
        sys.stderr = _StreamColorizer(sys.stderr)
        _stream_color_installed = True
    except Exception:
        pass


class _ColorFormatter(logging.Formatter):
    RESET = "\033[0m"
    C_INFO = "\033[36m"      # cyan
    C_WARN = "\033[33m"      # yellow
    C_ERR = "\033[31m"       # red
    C_EVENT = "\033[35m"     # magenta
    C_DRAW = "\033[34m"      # blue
    C_PRICE = "\033[32m"     # green

    def format(self, record: logging.LogRecord) -> str:
        base = super().format(record)
        msg = str(record.getMessage() or "")
        if msg.startswith("[活动日志]"):
            return f"{self.C_EVENT}{base}{self.RESET}"
        if msg.startswith("[作图]"):
            return f"{self.C_DRAW}{base}{self.RESET}"
        if msg.startswith("[价格]"):
            return f"{self.C_PRICE}{base}{self.RESET}"
        if record.levelno >= logging.ERROR:
            return f"{self.C_ERR}{base}{self.RESET}"
        if record.levelno >= logging.WARNING:
            return f"{self.C_WARN}{base}{self.RESET}"
        return f"{self.C_INFO}{base}{self.RESET}"


def setup_logger() -> logging.Logger:
    install_stream_colorizer()
    logger = logging.getLogger("qingjian_cs")
    if logger.handlers:
        return logger
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler(sys.stdout)
    formatter = _ColorFormatter("[%(asctime)s] %(levelname)s: %(message)s", "%H:%M:%S")
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.propagate = False

    # 降低 AgentScope 内部推理/格式器日志噪音，保留本项目活动日志。
    logging.getLogger("agentscope").setLevel(logging.ERROR)
    logging.getLogger("agentscope.formatter").setLevel(logging.ERROR)
    logging.getLogger("agentscope.agent").setLevel(logging.ERROR)
    fmt_logger = logging.getLogger("_openai_formatter")
    fmt_logger.setLevel(logging.CRITICAL)
    fmt_logger.propagate = False
    fmt_logger.disabled = True
    fmt_logger.handlers.clear()
    fmt_logger.addHandler(logging.NullHandler())
    logging.getLogger("_react_agent").setLevel(logging.ERROR)
    logging.getLogger("httpx").setLevel(logging.ERROR)
    logging.getLogger("urllib3").setLevel(logging.ERROR)

    # 兜底：把当前已注册的同类噪声 logger 一并禁掉
    for name in list(logging.root.manager.loggerDict.keys()):
        if "_openai_formatter" in name:
            lg = logging.getLogger(name)
            lg.setLevel(logging.CRITICAL)
            lg.propagate = False
            lg.disabled = True
            lg.handlers.clear()
            lg.addHandler(logging.NullHandler())

    return logger