Agent/examples/auto_put_ad_mini/tools/sls_setup.py

"""阿里云 SLS 日志上报 — 工具集。

数据流:
  Python logger → logging_setup.setup_logging()
                 → 本地 FileHandler（全量保留）
                 → QueuedLogHandler（异步队列 + 批量 flush）
                                                → SLS info-log / error-log

设计原则:
  · 配置缺失 → logging_setup 内部自动降级 → 不上报，主链路不受影响
  · 上报失败 → SDK 内部 retry + SafeQueuedLogHandler 吞异常，主链路不受影响
  · INFO/WARNING 进 info-log，ERROR/CRITICAL 进 error-log

注意:
  · 生产环境日志初始化入口是 ``logging_setup.setup_logging()``，已内置 SLS 挂载
  · 本模块仅提供 smoke_test 等辅助工具，不重复实现日志初始化
"""

from __future__ import annotations

import logging
import os
import sys

logger = logging.getLogger(__name__)


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

    用法（从项目根目录）:
        cd examples/auto_put_ad_mini
        python -c "from tools.sls_setup import smoke_test; smoke_test()"

    然后去 SLS 控制台 logstore=info-log / error-log 查看是否收到。
    """

    # 确保 examples/auto_put_ad_mini 在 sys.path 中，兼容不同工作目录
    _project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    if _project_root not in sys.path:
        sys.path.insert(0, _project_root)

    # 加载 .env 中的 SLS 凭证（logging_setup 不自行 load_dotenv，由调用方负责）
    from dotenv import load_dotenv
    load_dotenv(os.path.join(_project_root, ".env"))

    from logging_setup import setup_logging

    # setup_logging() 内部已调用 attach_sls_handler()，无需重复挂载
    setup_logging(level="INFO", capture_output=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")
    import time
    time.sleep(3)
    logger.info("[smoke_test] 测试完成，请去 SLS 控制台 logstore 查看")
    return True