Python logging模块：无需双处理器，从全级别日志过滤错误日志至文件

1. 过滤已生成日志中的Error条目 #

由于要求不能使用两个日志处理器，我们直接对已存在的全级别日志文件做事后过滤，而非在日志生成阶段分流。

假设你的全级别日志路径为all_levels.log，日志采用标准格式（例如包含- ERROR -标识，示例行：2024-05-20 14:30:00,456 - app - ERROR - 数据库连接失败），可以用以下代码过滤：

def filter_error_logs(source_path, target_path):
    with open(source_path, 'r', encoding='utf-8') as src, open(target_path, 'w', encoding='utf-8') as dst:
        for line in src:
            # 根据日志格式判断级别，可根据实际格式调整
            if ' - ERROR - ' in line:
                dst.write(line)

# 调用示例
filter_error_logs('all_levels.log', 'error_only.log')

如果日志是JSON格式（例如{"level": "ERROR", "message": "..."}），可以修改判断逻辑为解析JSON对象：

import json

def filter_error_logs(source_path, target_path):
    with open(source_path, 'r', encoding='utf-8') as src, open(target_path, 'w', encoding='utf-8') as dst:
        for line in src:
            try:
                log_entry = json.loads(line.strip())
                if log_entry.get('level') == 'ERROR':
                    dst.write(line)
            except json.JSONDecodeError:
                # 跳过无效的JSON行
                continue

2. 上传错误日志到AWS S3 #

使用boto3库完成上传，需先配置AWS凭证（可通过~/.aws/credentials文件、环境变量或IAM角色，根据部署环境选择）：

import boto3

def upload_to_s3(local_file, bucket_name, s3_key):
    s3_client = boto3.client('s3')
    try:
        s3_client.upload_file(local_file, bucket_name, s3_key)
        print(f"文件 {local_file} 已上传至S3存储桶 {bucket_name}，对象键: {s3_key}")
    except Exception as e:
        raise RuntimeError(f"S3上传失败: {str(e)}")

# 调用示例
upload_to_s3('error_only.log', 'your-bucket-name', 'logs/error_only.log')

3. 完整流程整合 #

将过滤与上传步骤合并为完整脚本：

def filter_error_logs(source_path, target_path):
    with open(source_path, 'r', encoding='utf-8') as src, open(target_path, 'w', encoding='utf-8') as dst:
        for line in src:
            if ' - ERROR - ' in line:
                dst.write(line)

def upload_to_s3(local_file, bucket_name, s3_key):
    import boto3
    s3_client = boto3.client('s3')
    s3_client.upload_file(local_file, bucket_name, s3_key)

if __name__ == "__main__":
    # 配置参数
    SOURCE_LOG = 'all_levels.log'
    ERROR_LOG = 'error_only.log'
    S3_BUCKET = 'your-s3-bucket'
    S3_OBJECT_KEY = 'error_logs/error_20240520.log'

    # 执行流程
    filter_error_logs(SOURCE_LOG, ERROR_LOG)
    upload_to_s3(ERROR_LOG, S3_BUCKET, S3_OBJECT_KEY)
    print("错误日志过滤及上传完成")

关键注意事项 #

• 日志格式适配：务必根据你实际的日志格式调整过滤逻辑，避免漏判或误判。
• 权限配置：确保运行脚本的实体拥有S3存储桶的PutObject权限，本地运行可通过aws configure配置凭证，AWS服务内运行建议使用IAM角色。
• 异常处理：可根据业务需求添加文件不存在、S3网络异常等场景的捕获处理。

内容的提问来源于stack exchange，提问作者user369287