# logger_config.py
import logging.config

LOGGING_CONFIG = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'standard': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            'datefmt': '%Y-%m-%d %H:%M:%S'
        },
        'simple': {
            'format': '%(levelname)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'DEBUG',
            'formatter': 'standard',
            'stream': 'ext://sys.stdout'
        },
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'INFO',
            'formatter': 'standard',
            'filename': 'app.log',
            'maxBytes': 10485760,  # 10MB
            'backupCount': 5
        }
    },
    'loggers': {
        '': {  # root logger
            'handlers': ['console', 'file'],
            'level': 'DEBUG',
            'propagate': False
        },
        'api': {
            'handlers': ['console', 'file'],
            'level': 'DEBUG',
            'propagate': False
        }
    }
}

def setup_logging():
    logging.config.dictConfig(LOGGING_CONFIG)

# main.py
from logger_config import setup_logging
import logging

setup_logging()
logger = logging.getLogger(__name__)

def main():
    logger.info("Application started")
    # 業(yè)務(wù)代碼...

import logging
logger = logging.getLogger(__name__)

def process_order(order_id):
    logger.debug(f"Processing order {order_id} - raw data: {order_data}")
    
    try:
        result = api_call(order_id)
        logger.info(f"Order {order_id} processed successfully")
        return result
    except TimeoutError:
        logger.warning(f"Order {order_id} processing timeout, retrying...")
        retry_process(order_id)
    except Exception as e:
        logger.error(f"Order {order_id} processing failed: {str(e)}", exc_info=True)
        raise

# 使用QueueHandler實(shí)現(xiàn)異步日志
import logging
import queue
from logging.handlers import QueueHandler, QueueListener
import threading

log_queue = queue.Queue(-1)  # 無(wú)限制隊(duì)列
queue_handler = QueueHandler(log_queue)

# 配置實(shí)際處理日志的handler（可配置多個(gè)）
file_handler = logging.FileHandler('async.log')
console_handler = logging.StreamHandler()

listener = QueueListener(log_queue, file_handler, console_handler)
listener.start()

# 應(yīng)用中使用
logger = logging.getLogger('async_logger')
logger.addHandler(queue_handler)
logger.setLevel(logging.DEBUG)

# 使用完畢后
listener.stop()

from logging import Filter

class DuplicateFilter(Filter):
    def __init__(self):
        self.msgs = set()

    def filter(self, record):
        msg = record.getMessage()
        if msg in self.msgs:
            return False
        self.msgs.add(msg)
        return True

# 使用示例
logger = logging.getLogger('dup_filter')
handler = logging.FileHandler('dup.log')
handler.addFilter(DuplicateFilter())
logger.addHandler(handler)

import json
import logging

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
            'message': record.getMessage(),
            'thread': record.threadName,
            'process': record.processName
        }
        
        if record.exc_info:
            log_record['exception'] = self.formatException(record.exc_info)
            
        return json.dumps(log_record)

# 使用示例
handler = logging.FileHandler('app.json')
handler.setFormatter(JsonFormatter())
logger = logging.getLogger('json_logger')
logger.addHandler(handler)
logger.setLevel(logging.INFO)

import logging
from contextvars import ContextVar

logger = logging.getLogger(__name__)
request_id_var = ContextVar('request_id', default=None)

class RequestIDFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True

# 配置過(guò)濾器
handler = logging.StreamHandler()
handler.addFilter(RequestIDFilter())
logger.addHandler(handler)

# 在Web框架中間件中設(shè)置
def request_middleware(request):
    request_id = generate_id()
    request_id_var.set(request_id)
    logger.info(f"Request started: {request_id}")

# 記錄接口處理時(shí)間
import time
import logging

logger = logging.getLogger('api_perf')

def api_endpoint(request):
    start_time = time.time()
    try:
        # 業(yè)務(wù)邏輯
        result = process_request(request)
        duration = time.time() - start_time
        logger.info(f"API {request.path} processed in {duration:.3f}s")
        return result
    except Exception as e:
        logger.error(f"API {request.path} failed: {str(e)}", exc_info=True)
        raise

# 記錄詳細(xì)錯(cuò)誤上下文
def transfer_money(from_account, to_account, amount):
    logger = logging.getLogger('transaction')
    logger.info(
        f"Starting transfer",
        extra={
            'from': from_account,
            'to': to_account,
            'amount': amount,
            'initial_balance': get_balance(from_account)
        }
    )
    
    try:
        # 轉(zhuǎn)賬邏輯
        result = execute_transfer()
        logger.info("Transfer completed", extra={'new_balance': get_balance(from_account)})
        return result
    except Exception as e:
        logger.error(
            "Transfer failed",
            exc_info=True,
            extra={'status': 'failed', 'retry_count': get_retry_count()}
        )
        raise

filter {
  if [level] == "ERROR" {
    mutate {
      add_field => { "alert" => "true" }
    }
  }
  
  if "database" in [message] {
    grok {
      match => { "message" => "Database error: %{GREEDYDATA:error_msg}" }
    }
  }
}

import re

class SensitiveDataFilter(logging.Filter):
    def filter(self, record):
        # 脫敏信用卡號(hào)
        record.message = re.sub(r'\d{12}\d{4}', '****-****-****-XXXX', record.message)
        # 脫敏郵箱
        record.message = re.sub(r'([\w.-]+)@([\w.-]+)', r'*@\2', record.message)
        return True

import os
import logging.config

def get_logging_config():
    env = os.getenv('APP_ENV', 'development')
    
    if env == 'production':
        return {
            'handlers': {
                'file': {
                    'class': 'logging.handlers.RotatingFileHandler',
                    'filename': '/var/log/app.log',
                            }
                        }
                    }
    else:
        return {
            'handlers': {
                'console': {
                    'class': 'logging.StreamHandler',
                    'stream': 'ext://sys.stdout'
                }
            }
        }

logging.config.dictConfig(get_logging_config())