logging 日志模块

import logging

# 层级继承演示
root_logger = logging.getLogger()              # root，默认 WARNING
app_logger = logging.getLogger('app')           # app
app_logger.setLevel(logging.DEBUG)              # 显式设置 DEBUG
db_logger = logging.getLogger('app.db')         # 未设置级别，继承 app 的 DEBUG
pool_logger = logging.getLogger('app.db.pool')  # 未设置级别，继承 app 的 DEBUG

# 验证有效级别
print(pool_logger.getEffectiveLevel())  # 10 (DEBUG)，从 app 继承

import logging

# basicConfig 只在第一次调用时生效，后续调用无效
logging.basicConfig(
    level=logging.DEBUG,                                    # 设置最低级别为 DEBUG
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',  # 日志格式
    datefmt='%Y-%m-%d %H:%M:%S',                            # 时间格式
    filename='app.log',                                     # 输出到文件（不指定则输出到控制台）
    filemode='a',                                           # 追加模式（'w' 为覆盖）
    encoding='utf-8',                                       # 文件编码（Python 3.9+）
)

# 记录不同级别的日志
logging.debug('调试信息：变量 x = 42')       # 开发调试用
logging.info('程序启动完成')                  # 确认正常运行
logging.warning('磁盘空间不足 20%')           # 潜在问题
logging.error('数据库连接超时')               # 功能异常
logging.critical('系统内存耗尽，即将退出')     # 致命错误

2026-06-04 10:30:00 - root - DEBUG - 调试信息：变量 x = 42
2026-06-04 10:30:00 - root - INFO - 程序启动完成
2026-06-04 10:30:00 - root - WARNING - 磁盘空间不足 20%
2026-06-04 10:30:00 - root - ERROR - 数据库连接超时
2026-06-04 10:30:00 - root - CRITICAL - 系统内存耗尽，即将退出

import logging

# 1. 创建 Logger（每个模块一个，用 __name__ 命名）
logger = logging.getLogger(__name__)       # Logger 名称 = 模块名
logger.setLevel(logging.DEBUG)             # Logger 级别设为最低，由 Handler 控制

# 2. 创建控制台 Handler —— 开发时看 INFO 及以上
console_handler = logging.StreamHandler()  # 输出到 sys.stderr
console_handler.setLevel(logging.INFO)     # 控制台只显示 INFO 及以上

# 3. 创建文件 Handler —— 记录所有 DEBUG 及以上
file_handler = logging.FileHandler(
    'debug.log',                           # 日志文件路径
    mode='a',                              # 追加模式
    encoding='utf-8',                      # 指定编码，避免中文乱码
)
file_handler.setLevel(logging.DEBUG)       # 文件记录所有级别

# 4. 创建 Formatter —— 定义输出格式
console_fmt = logging.Formatter(
    '%(asctime)s [%(levelname)s] %(message)s',       # 简洁格式
    datefmt='%H:%M:%S',                               # 只显示时分秒
)
file_fmt = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - '
    '%(filename)s:%(lineno)d - %(message)s',          # 详细格式
    datefmt='%Y-%m-%d %H:%M:%S',
)

# 5. 将 Formatter 绑定到 Handler
console_handler.setFormatter(console_fmt)
file_handler.setFormatter(file_fmt)

# 6. 将 Handler 添加到 Logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)

# 7. 记录日志
logger.debug('这条只在文件中')       # 控制台级别不够，只在文件
logger.info('这条控制台和文件都有')   # 两个 Handler 都满足
logger.error('错误！控制台和文件都有')

10:30:00 [INFO] 这条控制台和文件都有
10:30:00 [ERROR] 错误！控制台和文件都有

2026-06-04 10:30:00 - __main__ - DEBUG - test.py:28 - 这条只在文件中
2026-06-04 10:30:00 - __main__ - INFO - test.py:29 - 这条控制台和文件都有
2026-06-04 10:30:00 - __main__ - ERROR - test.py:30 - 错误！控制台和文件都有

import logging
from logging.handlers import RotatingFileHandler

logger = logging.getLogger('app')

# 创建按大小轮转的 Handler
# maxBytes: 单个文件最大字节数（这里 10MB）
# backupCount: 保留的备份文件数量（超出则删除最旧的）
handler = RotatingFileHandler(
    filename='app.log',                    # 主日志文件
    maxBytes=10 * 1024 * 1024,             # 10MB 后轮转
    backupCount=5,                         # 保留 5 个备份
    encoding='utf-8',                      # 编码
)

handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger.addHandler(handler)

# 轮转后的文件命名：app.log → app.log.1 → app.log.2 → ... → app.log.5
# app.log 始终是当前日志，app.log.1 是最近的备份

import logging
from logging.handlers import TimedRotatingFileHandler

logger = logging.getLogger('app')

# when 参数取值：
# 'S' - 秒, 'M' - 分钟, 'H' - 小时, 'D' - 天
# 'midnight' - 每天午夜, 'W0'-'W6' - 每周指定日（W0=周一）
handler = TimedRotatingFileHandler(
    filename='app.log',                    # 主日志文件
    when='midnight',                       # 每天午夜轮转
    interval=1,                            # 每 1 个 when 单位轮转一次
    backupCount=30,                        # 保留 30 天的日志
    encoding='utf-8',
)

handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger.addHandler(handler)

# 轮转后的文件命名：app.log → app.log.2026-06-03 → app.log.2026-06-02 → ...

import logging

# % 风格（默认，最常用）
fmt_percent = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')

# { 风格（str.format，更现代）
fmt_brace = logging.Formatter('{asctime} - {levelname} - {message}', style='{')

# $ 风格（string.Template，较少用）
fmt_dollar = logging.Formatter('$asctime - $levelname - $message', style='$')

import logging
import logging.config

LOGGING_CONFIG = {
    'version': 1,                          # 配置版本，必须为 1
    'disable_existing_loggers': False,      # 不禁用已有 Logger

    # ── 格式器 ──
    'formatters': {
        'verbose': {                        # 详细格式（文件用）
            'format': '%(asctime)s - %(name)s - %(levelname)s - '
                      '%(filename)s:%(lineno)d - %(message)s',
            'datefmt': '%Y-%m-%d %H:%M:%S',
        },
        'simple': {                         # 简洁格式（控制台用）
            'format': '%(asctime)s [%(levelname)s] %(message)s',
            'datefmt': '%H:%M:%S',
        },
        'json': {                           # JSON 格式（结构化日志）
            'class': 'pythonjsonlogger.jsonlogger.JsonFormatter',
            'format': '%(asctime)s %(name)s %(levelname)s %(message)s',
        },
    },

    # ── 处理器 ──
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',       # 控制台输出
            'level': 'INFO',                         # 控制台只看 INFO 及以上
            'formatter': 'simple',                   # 使用简洁格式
            'stream': 'ext://sys.stdout',            # 输出到 stdout
        },
        'file_debug': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'DEBUG',                        # 文件记录所有级别
            'formatter': 'verbose',                  # 使用详细格式
            'filename': 'logs/debug.log',
            'maxBytes': 10485760,                    # 10MB
            'backupCount': 5,
            'encoding': 'utf-8',
        },
        'file_error': {
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'level': 'ERROR',                        # 只记录 ERROR 及以上
            'formatter': 'verbose',
            'filename': 'logs/error.log',
            'when': 'midnight',                      # 每天轮转
            'backupCount': 30,                       # 保留 30 天
            'encoding': 'utf-8',
        },
    },

    # ── 记录器 ──
    'loggers': {
        'app': {                                     # 应用顶层 Logger
            'level': 'DEBUG',
            'handlers': ['console', 'file_debug', 'file_error'],
            'propagate': False,                      # 不向 root 传播，避免重复
        },
        'app.db': {                                  # 数据库模块
            'level': 'INFO',                         # 数据库模块只记 INFO 及以上
            'propagate': True,                       # 传播到 app（app 的 Handler 也会处理）
        },
    },

    # ── 根记录器 ──
    'root': {
        'level': 'WARNING',
        'handlers': ['console'],
    },
}

# 加载配置
logging.config.dictConfig(LOGGING_CONFIG)

# 使用
logger = logging.getLogger('app')
logger.info('应用启动')

# logging_config.yaml
version: 1
disable_existing_loggers: false

formatters:
  verbose:
    format: "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s"
    datefmt: "%Y-%m-%d %H:%M:%S"
  simple:
    format: "%(asctime)s [%(levelname)s] %(message)s"
    datefmt: "%H:%M:%S"

handlers:
  console:
    class: logging.StreamHandler
    level: INFO
    formatter: simple
    stream: ext://sys.stdout
  file_debug:
    class: logging.handlers.RotatingFileHandler
    level: DEBUG
    formatter: verbose
    filename: logs/debug.log
    maxBytes: 10485760
    backupCount: 5
    encoding: utf-8

loggers:
  app:
    level: DEBUG
    handlers: [console, file_debug]
    propagate: false

root:
  level: WARNING
  handlers: [console]

# 加载 YAML 配置
import logging.config
import yaml

with open('logging_config.yaml', 'r', encoding='utf-8') as f:
    config = yaml.safe_load(f)

logging.config.dictConfig(config)

logger = logging.getLogger('app')
logger.info('YAML 配置加载成功')

import logging

# ── 自定义 Filter：只记录包含特定关键字的日志 ──
class KeywordFilter(logging.Filter):
    """只放行消息中包含指定关键字的日志"""
    def __init__(self, keyword):
        super().__init__()
        self.keyword = keyword

    def filter(self, record):
        # 返回 True 放行，False 拦截
        return self.keyword in record.getMessage()

# ── 自定义 Filter：只记录特定模块的日志 ──
class ModuleFilter(logging.Filter):
    """只放行指定 Logger 名称开头的日志"""
    def __init__(self, prefix):
        super().__init__()
        self.prefix = prefix

    def filter(self, record):
        return record.name.startswith(self.prefix)

# ── 使用示例 ──
logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setLevel(logging.DEBUG)

# 添加关键字过滤器：只记录包含"重要"的消息
handler.addFilter(KeywordFilter('重要')

handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger.addHandler(handler)

logger.debug('普通调试信息')        # 被过滤，不输出
logger.info('这是一条重要通知')     # 包含"重要"，输出
logger.warning('磁盘空间不足')      # 被过滤，不输出
logger.error('重要：服务不可用')    # 包含"重要"，输出

2026-06-04 10:30:00 - app - INFO - 这是一条重要通知
2026-06-04 10:30:00 - app - ERROR - 重要：服务不可用

import logging

class RequestLogger(logging.LoggerAdapter):
    """自动注入 request_id 和 user_id 的日志适配器"""
    def process(self, msg, kwargs):
        # 从 extra 中取出上下文，拼接到消息前
        request_id = self.extra.get('request_id', 'N/A')
        user_id = self.extra.get('user_id', 'anonymous')
        return f'[{request_id}][{user_id}] {msg}', kwargs

# 基础 Logger 配置
logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')
logger.addHandler(handler)

# 创建带上下文的适配器
req_logger = RequestLogger(logger, {'request_id': 'REQ-001', 'user_id': 'alice'})

req_logger.info('用户登录')           # [REQ-001][alice] 用户登录
req_logger.warning('密码错误 3 次')   # [REQ-001][alice] 密码错误 3 次

2026-06-04 10:30:00 [INFO] [REQ-001][alice] 用户登录
2026-06-04 10:30:00 [WARNING] [REQ-001][alice] 密码错误 3 次

import logging
import logging.handlers
import queue
import threading
import time

# 1. 创建日志队列
log_queue = queue.Queue(-1)  # -1 表示无大小限制

# 2. 创建 QueueHandler —— 线程只需把日志放入队列，立即返回
queue_handler = logging.handlers.QueueHandler(log_queue)

# 3. 配置 Logger 使用 QueueHandler
logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG)
logger.addHandler(queue_handler)

# 4. 创建实际写入的 Handler
file_handler = logging.FileHandler('app.log', encoding='utf-8')
file_handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(threadName)s - %(name)s - %(levelname)s - %(message)s'
)

# 5. 创建 QueueListener —— 专门从队列取日志并写入文件
listener = logging.handlers.QueueListener(
    log_queue,                              # 日志队列
    file_handler,                           # 实际 Handler（可多个）
    respect_handler_level=True,             # 尊重 Handler 级别设置
)
listener.start()                            # 启动监听线程

# 6. 多线程写日志 —— 全部异步，不阻塞工作线程
def worker(worker_id):
    for i in range(100):
        logger.info(f'Worker-{worker_id} 处理任务 {i}')
        time.sleep(0.01)

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# 7. 程序退出前停止监听器
listener.stop()

"""
生产级日志配置模板
- 控制台：INFO 及以上，简洁格式
- debug.log：DEBUG 及以上，详细格式，按大小轮转
- error.log：ERROR 及以上，详细格式，按天轮转保留 30 天
- 结构化 JSON 日志：供 ELK/Splunk 等日志平台消费
"""
import logging
import logging.config
import os

def setup_logging(log_dir='logs', app_name='app'):
    """初始化生产级日志配置"""
    os.makedirs(log_dir, exist_ok=True)     # 确保日志目录存在

    config = {
        'version': 1,
        'disable_existing_loggers': False,

        'formatters': {
            'console': {
                'format': '%(asctime)s [%(levelname)s] %(name)s: %(message)s',
                'datefmt': '%H:%M:%S',
            },
            'file': {
                'format': (
                    '%(asctime)s - %(name)s - %(levelname)s - '
                    '%(filename)s:%(lineno)d - %(funcName)s - %(message)s'
                ),
                'datefmt': '%Y-%m-%d %H:%M:%S',
            },
        },

        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
                'formatter': 'console',
                'stream': 'ext://sys.stdout',
            },
            'debug_file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'DEBUG',
                'formatter': 'file',
                'filename': os.path.join(log_dir, f'{app_name}_debug.log'),
                'maxBytes': 10 * 1024 * 1024,     # 10MB
                'backupCount': 5,
                'encoding': 'utf-8',
            },
            'error_file': {
                'class': 'logging.handlers.TimedRotatingFileHandler',
                'level': 'ERROR',
                'formatter': 'file',
                'filename': os.path.join(log_dir, f'{app_name}_error.log'),
                'when': 'midnight',
                'backupCount': 30,
                'encoding': 'utf-8',
            },
        },

        'loggers': {
            app_name: {
                'level': 'DEBUG',
                'handlers': ['console', 'debug_file', 'error_file'],
                'propagate': False,          # 关键：不向 root 传播，避免重复输出
            },
        },

        'root': {
            'level': 'WARNING',
            'handlers': ['console'],
        },
    }

    logging.config.dictConfig(config)
    return logging.getLogger(app_name)

# 使用
logger = setup_logging(log_dir='logs', app_name='myapp')
logger.info('应用启动')
logger.debug('调试信息')
logger.error('错误信息')

import logging
import json
from datetime import datetime

# ── 方法一：自定义 JSON Formatter ──
class JsonFormatter(logging.Formatter):
    """将日志格式化为 JSON，便于日志平台解析"""
    def format(self, record):
        log_obj = {
            'timestamp': datetime.fromtimestamp(record.created).isoformat(),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
            'process': record.process,
            'thread': record.threadName,
        }
        # 如果有异常信息，添加到 JSON
        if record.exc_info:
            log_obj['exception'] = self.formatException(record.exc_info)
        # 如果有 extra 字段，合并到 JSON
        if hasattr(record, 'extra_data'):
            log_obj['data'] = record.extra_data
        return json.dumps(log_obj, ensure_ascii=False)  # ensure_ascii=False 支持中文

# ── 使用 ──
logger = logging.getLogger('app')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter()
logger.addHandler(handler)

# 普通日志
logger.info('用户登录', extra={'extra_data': {'user_id': 'alice', 'ip': '192.168.1.1'}})

# 异常日志
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception('计算异常')

{"timestamp":"2026-06-04T10:30:00.123456","level":"INFO","logger":"app","message":"用户登录","module":"test","function":"<module>","line":28,"process":12345,"thread":"MainThread","data":{"user_id":"alice","ip":"192.168.1.1"}}
{"timestamp":"2026-06-04T10:30:00.234567","level":"ERROR","logger":"app","message":"计算异常","module":"test","function":"<module>","line":33,"process":12345,"thread":"MainThread","exception":"Traceback (most recent call last):\n  File \"test.py\", line 31, in <module>\n    result = 1 / 0\nZeroDivisionError: division by zero"}

import logging

logger = logging.getLogger('app')

# ── 错误做法：用 error() 记录异常 ──
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.error('除零错误')              # 丢失堆栈信息！

# ── 正确做法一：用 exception() ──
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception('除零错误')           # 自动附加完整堆栈

# ── 正确做法二：用 error() + exc_info=True ──
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.error('除零错误', exc_info=True)  # 等价于 exception()

# ── 正确做法三：记录异常但不输出堆栈 ──
try:
    result = 1 / 0
except ZeroDivisionError as e:
    logger.error(f'除零错误: {e}')         # 只记录异常消息，不输出堆栈

# ── 正确做法四：延迟格式化（性能优化） ──
# 使用 % 风格而非 f-string，只在日志真正输出时才格式化
try:
    result = 1 / 0
except ZeroDivisionError as e:
    logger.exception('计算错误: %s', e)    # 延迟格式化，避免无谓的字符串拼接

# ── myapp/__init__.py ──
import logging

# 包级别 Logger，只做配置入口
logger = logging.getLogger('myapp')

# ── myapp/db.py ──
import logging

# 每个模块用自己的 Logger，名称自动对应 myapp.db
logger = logging.getLogger(__name__)  # 等价于 logging.getLogger('myapp.db')

def connect():
    logger.info('数据库连接成功')     # 会传播到 myapp 的 Handler

# ── myapp/api.py ──
import logging

logger = logging.getLogger(__name__)  # 等价于 logging.getLogger('myapp.api')

def handle_request():
    logger.debug('处理请求')          # 会传播到 myapp 的 Handler

# ── main.py ──
import logging
import logging.config
from myapp import db, api

# 只在入口配置一次
logging.config.dictConfig({...})      # 配置 myapp Logger 的 Handler

# 之后所有模块的日志自动传播到 myapp 的 Handler
# 无需在每个模块重复配置

import logging

# ── 错误示例：重复输出 ──
logging.basicConfig(level=logging.INFO)           # root 有一个 StreamHandler
logger = logging.getLogger('app')
logger.setLevel(logging.INFO)
console_handler = logging.StreamHandler()          # app 又加了一个 StreamHandler
logger.addHandler(console_handler)
# 结果：logger.info('test') 输出 2 次！
# 1次：app 自己的 console_handler
# 1次：传播到 root 的 basicConfig Handler

# ── 解决方案一：关闭传播 ──
logger.propagate = False                           # 不向 root 传播

# ── 解决方案二：不给 root 配 Handler ──
# 不使用 basicConfig，只给 app 配 Handler

# ── 解决方案三：检查是否重复添加 Handler ──
# 每次模块被 import 时，addHandler 会重复添加
# 防御性写法：
if not logger.handlers:                            # 只在无 Handler 时添加
    logger.addHandler(console_handler)

import logging

# ── 解决方案：使用 force=True（Python 3.8+）──
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
    force=True,                                    # 强制重新配置，移除已有 Handler
)

# ── 或者：直接操作 root Logger ──
root = logging.getLogger()
for h in root.handlers[:]:                         # 先清除所有 Handler
    root.removeHandler(h)
root.addHandler(logging.StreamHandler()

import logging

# ── 解决方案：所有文件 Handler 显式指定 encoding ──
handler = logging.FileHandler('app.log', encoding='utf-8')           # FileHandler
handler = logging.handlers.RotatingFileHandler('app.log', encoding='utf-8', ...)  # 轮转
handler = logging.handlers.TimedRotatingFileHandler('app.log', encoding='utf-8', ...)  # 定时轮转

# ── basicConfig 方式 ──
logging.basicConfig(filename='app.log', encoding='utf-8', ...)       # Python 3.9+

import logging

# 1. 延迟格式化：只在日志真正输出时才拼接字符串
logger.debug('值: %s', expensive_func())   # 如果 DEBUG 级别未开启，expensive_func() 不会被调用
# 对比：
logger.debug(f'值: {expensive_func()}')    # 即使 DEBUG 未开启，f-string 也会执行

# 2. 级别守卫：避免昂贵的计算
if logger.isEnabledFor(logging.DEBUG):
    logger.debug('详细状态: %s', get_full_state())  # 只在 DEBUG 开启时才调用

# 3. 异步日志：QueueHandler 解耦 I/O
# 参见 3.10 节

# 4. 生产环境设 INFO 或 WARNING
# DEBUG 级别会产生大量日志，只在排查问题时临时开启

import logging

logging.basicConfig(
    format='%(asctime)s %(filename)s:%(lineno)d %(message)s',
    level=logging.DEBUG,
)

# ── 不推荐：封装函数导致位置失真 ──
def log(level, content):
    if level == 'info':
        logging.info(content)    # filename 和 lineno 指向这里，而非调用者

log('info', '测试')              # 显示 log 函数的行号，而非此行

# ── 推荐：直接使用 Logger 对象 ──
logger = logging.getLogger(__name__)
logger.info('测试')              # filename 和 lineno 正确指向调用位置

# ── 如果必须封装，使用 stacklevel（Python 3.8+）──
def log(level, content):
    getattr(logging, level)(content, stacklevel=2)  # 向上跳一层，指向真正的调用者

# ── 解决方案一：每个进程写不同文件 ──
import os
handler = logging.FileHandler(f'app_{os.getpid()}.log', encoding='utf-8')

# ── 解决方案二：使用 QueueHandler + 多进程队列 ──
from multiprocessing import Queue
log_queue = Queue()
queue_handler = logging.handlers.QueueHandler(log_queue)
# 主进程用 QueueListener 消费队列写入文件

# ── 解决方案三：使用 SocketHandler 发送到集中式日志服务 ──
handler = logging.handlers.SocketHandler('log-server', 9020)

# ── 解决方案四：使用 WatchedFileHandler + 外部 logrotate ──
handler = logging.handlers.WatchedFileHandler('app.log')
# 配合系统 logrotate 工具管理轮转