Agent/examples/auto_put_ad_mini/tools/sls_setup.py

"""阿里云 SLS 日志上报接入(2026-06-11)。

数据流:
  Python logger → 本地 FileHandler(全量保留)
                → QueuedLogHandler(aliyun-log-python-sdk 自带,异步队列 + 批量 flush)
                                                 → SLS Logstore

设计原则:
  · 配置缺失 → SLS_ENABLED=False → 不上报,主链路不受影响(降级)
  · 上报失败 → SDK 内部 retry + 异常吞掉(不抛出),主链路不受影响
  · INFO+ 全量上报(用户 2026-06-11 决策,日量级约 5k-20k 条)
"""

from __future__ import annotations

import logging
from typing import Optional

logger = logging.getLogger(__name__)


def attach_sls_handler(root_logger: Optional[logging.Logger] = None) -> bool:
    """给 root_logger 挂上 SLS QueuedLogHandler(若 SLS_ENABLED)。

    Returns:
        True 已挂上 / False 跳过(配置缺失 / SDK 缺失 / 异常)
    """
    if root_logger is None:
        root_logger = logging.getLogger()

    try:
        from config import (
            SLS_ENABLED,
            SLS_ENDPOINT,
            SLS_ACCESS_KEY_ID,
            SLS_ACCESS_KEY_SECRET,
            SLS_PROJECT,
            SLS_LOGSTORE,
            SLS_LOG_LEVEL,
        )
    except ImportError as e:
        logger.warning("[sls] config 缺 SLS_* 变量,跳过:%s", e)
        return False

    if not SLS_ENABLED:
        logger.info("[sls] SLS_ENABLED=False(凭证未完整配置),跳过上报")
        return False

    try:
        from aliyun.log import QueuedLogHandler
    except ImportError:
        logger.warning("[sls] aliyun-log-python-sdk 未安装,跳过上报(.venv/bin/pip3 install aliyun-log-python-sdk)")
        return False

    try:
        sls_handler = QueuedLogHandler(
            end_point=SLS_ENDPOINT,
            access_key_id=SLS_ACCESS_KEY_ID,
            access_key=SLS_ACCESS_KEY_SECRET,
            project=SLS_PROJECT,
            log_store=SLS_LOGSTORE,
            # 自动提取 record 标准字段
            fields=["levelname", "name", "filename", "lineno", "thread", "process"],
            # message 中 k=v 形式自动解析为独立字段(便于 SLS 索引/查询)
            extract_kv=True,
        )
        sls_handler.setLevel(getattr(logging, SLS_LOG_LEVEL, logging.INFO))
        sls_handler.setFormatter(logging.Formatter(
            "%(asctime)s | %(levelname)s | %(name)s | %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S",
        ))
        root_logger.addHandler(sls_handler)
        logger.info(
            "[sls] 已挂载 QueuedLogHandler → %s/%s/%s (level=%s)",
            SLS_ENDPOINT, SLS_PROJECT, SLS_LOGSTORE, SLS_LOG_LEVEL,
        )
        return True
    except Exception as e:
        logger.warning("[sls] 挂载失败(降级为不上报,主链路不受影响):%s", e)
        return False


def smoke_test() -> bool:
    """烟囱测试:发一条 INFO + 一条 WARNING + 一条 ERROR,等 3 秒让队列 flush。

    用法:python -c "from tools.sls_setup import smoke_test; smoke_test()"
    然后去 SLS 控制台 logstore=info-log 查看是否收到。
    """
    import time

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
        datefmt="%H:%M:%S",
    )
    ok = attach_sls_handler()
    if not ok:
        logger.error("[smoke_test] SLS 未挂上,无法测试")
        return False

    test_logger = logging.getLogger("sls_smoke_test")
    test_logger.info("[smoke_test] hello SLS - INFO 消息 event=smoke_info")
    test_logger.warning("[smoke_test] hello SLS - WARNING 消息 event=smoke_warn")
    test_logger.error("[smoke_test] hello SLS - ERROR 消息 event=smoke_error code=999")
    logger.info("[smoke_test] 已发 3 条,等 3s 让 QueuedLogHandler flush")
    time.sleep(3)
    logger.info("[smoke_test] 测试完成,请去 SLS 控制台 logstore 查看")
    return True