Logger SDK - 智能日志框架集成

🎯 核心特性

1. 智能框架检测

自动检测并复用项目中已有的日志框架，无需手动配置：

• ✅ loguru - 最流行的第三方日志库
• ✅ structlog - 结构化日志库
• ✅ logging - Python 标准库
• ✅ 自定义 logger - 通过环境变量指定

2. 框架无关的 trace_id

统一的 trace_id 管理，支持所有日志框架：

• 使用 threading.local() 支持多线程
• 使用 contextvars 支持异步/协程
• 自动注入到不同框架的日志中

3. 零侵入式集成

不破坏现有日志配置，完全向后兼容。

---

📦 安装

# 基础功能（仅依赖标准库）
pip install -r requirements.txt

# 可选：安装第三方日志框架
pip install loguru      # 如果想使用 loguru
pip install structlog   # 如果想使用 structlog

---

🚀 快速开始

场景 1: 项目中没有日志框架

# 直接使用，会自动创建一个 logging 实例
from huace_aigc_logs_client import logger, log_function

@log_function()
def my_function():
    logger.info("自动创建的日志实例")

场景 2: 项目中已有 loguru

# step 1: 先配置 loguru（在项目入口处）
from loguru import logger
logger.add("app.log", rotation="500 MB")

# step 2: 导入 logger_sdk（自动检测并复用 loguru）
from huace_aigc_logs_client import logger, log_function

@log_function()
def my_function():
    # trace_id 会自动注入到 loguru 的 extra 中
    logger.info("使用 loguru，自动包含 trace_id")

场景 3: 项目中已有 structlog

# step 1: 先配置 structlog
import structlog
structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)

# step 2: 导入 logger_sdk（自动检测并复用 structlog）
from huace_aigc_logs_client import logger, log_function

@log_function()
def my_function():
    # trace_id 会自动注入到结构化日志中
    logger.info("structured_event", user_id=123)

场景 4: 指定现有的 logger

# step 1: 设置环境变量
import os
os.environ['EXISTING_LOGGER_NAME'] = 'my_app_logger'

# step 2: 配置你的 logger
import logging
my_logger = logging.getLogger('my_app_logger')
my_logger.addHandler(logging.StreamHandler())

# step 3: 导入 logger_sdk（会复用 my_app_logger）
from huace_aigc_logs_client import logger, log_function

---

🔧 环境变量配置

环境变量	说明	默认值	示例
EXISTING_LOGGER_NAME	指定要复用的 logger 名称	无	my_app_logger
LOG_LEVEL	日志级别	INFO	DEBUG, INFO, WARNING
SERVICE_NAME	服务名称（用于新建 logger）	app-logger	my-service
LOG_SERVICE_URL	远程日志服务地址	无	http://log-server:8080
LOG_SERVICE_ACCESS_KEY	🔐 远程日志服务访问密钥（用于安全认证，客户端和服务端必须一致）	无	your-secret-key-123
LOG_FORCE_READ_STREAM_RESPONSE	是否强制记录读取流响应的日志（适用于需要调试流式响应的场景，可能会影响性能）	false	true

---

📖 使用指南

装饰器使用

1. 默认行为 - 复用 trace_id

from huace_aigc_logs_client import log_function, logger

@log_function()  # 默认复用现有 trace_id
def process_data(data):
    logger.info(f"处理数据: {data}")
    return validate_data(data)

@log_function()  # 继续复用相同的 trace_id
def validate_data(data):
    logger.info("验证数据")
    return True

日志输出：

2024-01-01 10:00:00 [INFO] [abc-123-def] 调用函数 process_data 开始 [复用trace_id: abc-123-def]
2024-01-01 10:00:00 [INFO] [abc-123-def] 处理数据: {...}
2024-01-01 10:00:00 [INFO] [abc-123-def] 调用函数 validate_data 开始 [复用trace_id: abc-123-def]
2024-01-01 10:00:00 [INFO] [abc-123-def] 验证数据

2. 强制生成新的 trace_id

@log_function(new_trace_id=True)  # 强制生成新的 trace_id
def independent_task():
    logger.info("执行独立任务")
    return "done"

日志输出：

2024-01-01 10:00:01 [INFO] [xyz-789-ghi] 调用函数 independent_task 开始 [新trace_id: xyz-789-ghi]
2024-01-01 10:00:01 [INFO] [xyz-789-ghi] 执行独立任务

异步函数支持

from huace_aigc_logs_client import async_log_function, logger
import asyncio

@async_log_function()
async def async_process(data):
    logger.info(f"异步处理: {data}")
    await asyncio.sleep(0.1)
    return "done"

@async_log_function(new_trace_id=True)
async def background_job():
    logger.info("后台异步任务")
    return "completed"

Web 框架集成

Flask

from flask import Flask
from huace_aigc_logs_client import init_flask_app_logger, logger

app = Flask(__name__)
init_flask_app_logger(app)  # 自动为每个请求生成 trace_id

@app.route('/api/test')
def test():
    logger.info("处理请求")  # 自动包含请求的 trace_id
    return {"status": "ok"}

FastAPI

from fastapi import FastAPI
from huace_aigc_logs_client import init_fastapi_app_logger, logger

app = FastAPI()
init_fastapi_app_logger(app)

@app.get("/api/test")
async def test():
    logger.info("处理异步请求")
    return {"status": "ok"}

---

🔍 检测优先级

logger_sdk 按以下优先级检测和选择日志框架：

1. loguru - 如果已配置且有 handlers
2. structlog - 如果已配置（structlog.is_configured() == True）
3. 环境变量指定的 logger - 通过 EXISTING_LOGGER_NAME 指定
4. root logger - 如果已经配置了 handlers
5. 新建 logging - 以上都没有，则创建新的

---

💡 最佳实践

1. 在项目入口处初始化日志

# main.py
# 如果使用 loguru
from loguru import logger
logger.add("logs/app_{time}.log", rotation="500 MB", retention="30 days")

# 然后导入其他模块
from huace_aigc_logs_client import logger as sdk_logger
from my_app import app

2. 使用环境变量管理配置

# .env
LOG_LEVEL=DEBUG
SERVICE_NAME=my-service
EXISTING_LOGGER_NAME=my_logger
LOG_SERVICE_URL=http://log-server:8080/api/logs
# 安全配置（生产环境必须配置）
LOG_SERVICE_ACCESS_KEY=your-secret-key-here

3. 🔐 远程日志安全配置（重要）

为了保护日志服务不被未授权访问，强烈建议配置 Access Key：

服务端配置 (日志服务器):

# 设置服务端 access key
export LOG_SERVICE_ACCESS_KEY="your-secret-key-here"

# 启动日志服务
python app.py

客户端配置 (应用端):

# 设置相同的 access key
export LOG_SERVICE_URL="http://log-server:8080"
export LOG_SERVICE_ACCESS_KEY="your-secret-key-here"
export SERVICE_NAME="my-service"

# 启动应用
python main.py

Python 代码中配置:

import os

# 在应用启动前设置
os.environ['LOG_SERVICE_URL'] = 'http://log-server:8080'
os.environ['LOG_SERVICE_ACCESS_KEY'] = 'your-secret-key-here'
os.environ['SERVICE_NAME'] = 'my-service'

# 然后导入 SDK
from huace_aigc_logs_client import logger, setup_remote_logging

# 启用远程日志
setup_remote_logging()

logger.info("远程日志已启用，带 access key 认证")

安全提示:

• ✅ 强烈推荐在生产环境中启用 Access Key
• ✅ 使用足够复杂的密钥（至少 32 位随机字符）
• ✅ 定期轮换密钥
• ✅ 不要在代码中硬编码密钥，使用环境变量或密钥管理服务
• ❌ 不要将密钥提交到版本控制系统

验证机制:

• 客户端在每次上报日志时，会在 HTTP 请求头中携带 X-Access-Key
• 服务端验证 Access Key 是否匹配
• 验证失败返回 401 Unauthorized 错误
• 不配置 Access Key 时，系统仍可正常工作（向后兼容）

4. 统一导入

# 在项目中统一从 logger_sdk 导入
from huace_aigc_logs_client import (
    # 核心组件
    logger,                  # 日志实例
    logger_framework,        # 查看当前使用的框架
    
    # trace_id 管理
    get_trace_id,           # 获取当前 trace_id
    set_trace_id,           # 设置 trace_id
    clear_trace_id,         # 清除 trace_id
    generate_trace_id,      # 生成新的 trace_id
    
    # 装饰器
    log_function,           # 同步函数装饰器
    async_log_function,     # 异步函数装饰器
    
    # 远程日志
    setup_remote_logging,   # 启用远程日志上报
    
    # Web 框架集成
    init_flask_app_logger,
    init_fastapi_app_logger,
    init_django_app_logger,
    init_tornado_app_logger,
    init_sanic_app_logger,
)

5. 混合使用场景

@log_function()
def handle_request(request_id):
    # 主流程使用相同的 trace_id
    data = fetch_data(request_id)
    result = process_data(data)
    
    # 触发独立的后台任务（使用新的 trace_id）
    trigger_background_job(result)
    
    return result

@log_function()
def fetch_data(request_id):
    # 复用主流程的 trace_id
    pass

@log_function(new_trace_id=True)
def trigger_background_job(data):
    # 使用独立的 trace_id
    pass

---

📊 trace_id 注入机制

不同日志框架的 trace_id 注入方式：

框架	注入方式	日志格式
logging	TraceIDFilter	%(trace_id)s 格式化参数
loguru	extra 参数	{extra[trace_id]}
structlog	processor	自动添加到 event_dict

---

🧪 测试

运行示例：

# 基本示例
python app-demo/example_logger_integration.py

# 装饰器使用示例
python app-demo/example_log_function_usage.py

---

🤔 常见问题

Q1: 如何查看当前使用的日志框架？

from huace_aigc_logs_client import logger_framework
print(f"当前框架: {logger_framework}")  # 输出: logging / loguru / structlog

Q2: 为什么没有检测到我的 loguru？

确保在导入 logger_sdk 之前先配置 loguru：

# ✅ 正确
from loguru import logger
logger.add("app.log")  # 先配置
from huace_aigc_logs_client import logger as sdk_logger

# ❌ 错误
from huace_aigc_logs_client import logger  # logger_sdk 先加载，检测不到
from loguru import logger  # 后配置无效

Q3: 可以同时使用多个日志实例吗？

可以，logger_sdk 只是提供一个便捷的单例，你仍然可以使用项目中的其他 logger：

from huace_aigc_logs_client import logger as sdk_logger
from loguru import logger as my_logger

sdk_logger.info("使用 logger_sdk")  # 带 trace_id
my_logger.info("使用原始 loguru")   # 原始配置

Q4: 远程日志上报失败怎么办？

检查以下配置：

import os

# 1. 检查环境变量是否正确设置
print("LOG_SERVICE_URL:", os.getenv('LOG_SERVICE_URL'))
print("LOG_SERVICE_ACCESS_KEY:", os.getenv('LOG_SERVICE_ACCESS_KEY', '未设置'))
print("SERVICE_NAME:", os.getenv('SERVICE_NAME'))

# 2. 确认调用了 setup_remote_logging()
from huace_aigc_logs_client import setup_remote_logging
setup_remote_logging()

# 3. 检查网络连接
import requests
try:
    response = requests.get(os.getenv('LOG_SERVICE_URL') + '/health')
    print("日志服务状态:", response.json())
except Exception as e:
    print("连接失败:", e)

常见错误:

• ❌ 401 Unauthorized: Access Key 不匹配，检查客户端和服务端的 LOG_SERVICE_ACCESS_KEY 是否一致
• ❌ 429 Too Many Requests: 请求过于频繁，触发限流保护
• ❌ 连接超时: 检查网络和服务端地址是否正确

Q5: Access Key 是否必须配置？

不是必须的，但强烈建议在生产环境配置：

• 不配置: 系统正常工作，但日志服务可被任何人访问（不安全）
• 配置: 只有持有正确密钥的客户端才能上报日志（安全）

# 开发环境（可选）
export LOG_SERVICE_URL="http://localhost:5000"

# 生产环境（强烈推荐）
export LOG_SERVICE_URL="http://log-server:8080"
export LOG_SERVICE_ACCESS_KEY="production-secret-key-2024"

Q6: 如何生成安全的 Access Key？

import secrets
import string

def generate_access_key(length=32):
    """生成安全的随机密钥"""
    alphabet = string.ascii_letters + string.digits
    return ''.join(secrets.choice(alphabet) for _ in range(length))

# 生成密钥
access_key = generate_access_key()
print(f"生成的 Access Key: {access_key}")
# 示例输出: aB3dE7fG9hK2mN4pQ6sT8vW0xY5zC1u7

Q7: 如何禁用自动检测，强制使用 logging？

设置环境变量：

import os
os.environ['FORCE_LOGGING'] = '1'  # 即将支持

---

📝 更新日志

v1.0.1 (2026-02-06)

• 🔐 新增 Access Key 安全认证机制
  • 支持通过 LOG_SERVICE_ACCESS_KEY 环境变量配置访问密钥
  • 客户端自动在请求头中携带 X-Access-Key
  • 服务端验证密钥，未授权访问返回 401
  • 向后兼容，不配置时系统仍正常工作
• 📚 完善 SDK 文档和使用示例
• 🐛 修复远程日志上报的若干问题

v1.0.0 (2024-01-01)

• ✨ 支持自动检测 loguru、structlog
• ✨ 支持通过环境变量指定现有 logger
• ✨ 统一的 trace_id 管理
• ✨ 同步和异步函数装饰器
• ✨ 支持 Flask、FastAPI、Django、Tornado、Sanic
• ✨ 远程日志批量上报功能
• ✨ 零侵入式集成，不破坏现有配置

---

🔗 相关链接

• 📖 完整文档
• 🐛 问题反馈
• 💬 讨论交流

---

🤝 贡献指南

欢迎提交 Pull Request 或 Issue！

1. Fork 本仓库
2. 创建特性分支 (git checkout -b feature/AmazingFeature)
3. 提交更改 (git commit -m 'Add some AmazingFeature')
4. 推送到分支 (git push origin feature/AmazingFeature)
5. 开启 Pull Request

---

📄 License

MIT License