如何写一个脚本解析日志模板

wen 实用脚本 2026-07-30 2

本文目录导读：

如何写一个脚本解析日志模板

基础日志解析器
高级日志解析器（支持多格式）
实际应用示例 - 处理不同日志格式
配置驱动的解析器
实用工具函数
使用建议

我来为您提供一个完整的日志解析脚本框架,包含多种常见模式：

基础日志解析器

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import re
import json
from datetime import datetime
from collections import defaultdict
import argparse
class LogParser:
    """通用日志解析器"""
    def __init__(self, log_pattern=None):
        self.log_pattern = log_pattern or r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}) (\w+) (.+)'
        self.logs = []
    def parse_file(self, file_path):
        """解析日志文件"""
        parsed_logs = []
        with open(file_path, 'r', encoding='utf-8') as f:
            for line_num, line in enumerate(f, 1):
                parsed = self.parse_line(line.strip(), line_num)
                if parsed:
                    parsed_logs.append(parsed)
        self.logs = parsed_logs
        return parsed_logs
    def parse_line(self, line, line_num=0):
        """解析单行日志"""
        match = re.match(self.log_pattern, line)
        if match:
            return {
                'line_number': line_num,
                'raw': line,
                'timestamp': match.group(1),
                'level': match.group(2),
                'message': match.group(3)
            }
        return None
# 使用示例
if __name__ == "__main__":
    parser = LogParser()
    logs = parser.parse_file('app.log')
    for log in logs[:5]:
        print(f"[{log['timestamp']}] {log['level']}: {log['message']}")

高级日志解析器（支持多格式）

import re
from typing import List, Dict, Optional
class AdvancedLogParser:
    """支持多种日志格式的高级解析器"""
    # 预定义日志模式
    PATTERNS = {
        'apache': r'(\S+) (\S+) (\S+) \[([^\]]+)\] "(\S+) (\S+) (\S+)" (\d+) (\d+)',
        'nginx': r'(\S+) - (\S+) \[([^\]]+)\] "(\S+) (\S+) (\S+)" (\d+) (\d+) "([^"]*)" "([^"]*)"',
        'syslog': r'(\w{3}\s+\d+\s+\d{2}:\d{2}:\d{2}) (\S+) (\S+) (.+)',
        'custom': r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}) \[(\w+)\] (\w+) - (.+)'
    }
    def __init__(self, log_type='custom'):
        self.log_type = log_type
        self.pattern = self.PATTERNS.get(log_type)
        self.fields = {
            'timestamp': 0,
            'level': 1,
            'message': -1
        }
    def parse(self, log_line: str) -> Optional[Dict]:
        """解析单行日志"""
        match = re.match(self.pattern, log_line)
        if match:
            result = {}
            for field, index in self.fields.items():
                result[field] = match.group(index) if index > 0 else match.group(0)
            # 增加额外解析
            result['length'] = len(log_line)
            result['parsed_time'] = datetime.now()
            return result
        return None
    def parse_file_with_stats(self, file_path: str) -> Dict:
        """解析文件并生成统计"""
        stats = defaultdict(int)
        parsed_logs = []
        with open(file_path, 'r') as f:
            for line in f:
                parsed = self.parse(line.strip())
                if parsed:
                    parsed_logs.append(parsed)
                    stats[parsed.get('level', 'unknown')] += 1
        return {
            'logs': parsed_logs,
            'total': len(parsed_logs),
            'stats': dict(stats)
        }

实际应用示例 - 处理不同日志格式

# 3.1 Apache日志解析
def parse_apache_log(log_line):
    """解析Apache访问日志"""
    pattern = r'(\S+) (\S+) (\S+) \[([^\]]+)\] "(\S+) (\S+) (\S+)" (\d+) (\d+)'
    match = re.match(pattern, log_line)
    if match:
        return {
            'ip': match.group(1),
            'identity': match.group(2),
            'user': match.group(3),
            'timestamp': match.group(4),
            'method': match.group(5),
            'path': match.group(6),
            'protocol': match.group(7),
            'status': int(match.group(8)),
            'size': int(match.group(9))
        }
    return None
# 3.2 自定义应用日志
class AppLogParser:
    def __init__(self):
        self.pattern = r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \[(\w+)\] (\w+) \((.+)\): (.+)'
    def parse_line(self, line):
        match = re.match(self.pattern, line)
        if match:
            return {
                'timestamp': match.group(1),
                'module': match.group(2),
                'level': match.group(3),
                'thread': match.group(4),
                'message': match.group(5)
            }
        return None
    def filter_by_level(self, logs, level='ERROR'):
        return [log for log in logs if log.get('level') == level]
    def extract_errors(self, file_path):
        errors = []
        with open(file_path, 'r') as f:
            for line in f:
                parsed = self.parse_line(line.strip())
                if parsed and parsed['level'] == 'ERROR':
                    errors.append(parsed)
        return errors

配置驱动的解析器

import yaml
import json
class ConfigurableLogParser:
    """基于配置文件的可配置解析器"""
    def __init__(self, config_file=None):
        self.config = self.load_config(config_file) if config_file else {}
        self.compiled_patterns = {}
    def load_config(self, config_file):
        """加载解析配置"""
        with open(config_file, 'r') as f:
            if config_file.endswith('.yaml'):
                return yaml.safe_load(f)
            elif config_file.endswith('.json'):
                return json.load(f)
    def compile_patterns(self):
        """编译正则表达式"""
        for name, config in self.config.get('patterns', {}).items():
            self.compiled_patterns[name] = {
                'pattern': re.compile(config['pattern']),
                'fields': config['fields']
            }
    def parse_with_config(self, line, pattern_name=None):
        """根据配置解析日志"""
        if pattern_name and pattern_name in self.compiled_patterns:
            config = self.compiled_patterns[pattern_name]
            match = config['pattern'].match(line)
            if match:
                result = {}
                for field, index in config['fields'].items():
                    result[field] = match.group(index)
                return result
        return None
# YAML配置示例
"""
patterns:
  application_log:
    pattern: '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \[(\w+)\] (.+)'
    fields:
      timestamp: 1
      level: 2
      message: 3
"""

实用工具函数

def log_analyzer(file_path, level='ERROR'):
    """快速日志分析工具"""
    with open(file_path, 'r') as f:
        lines = f.readlines()
    errors = []
    error_pattern = re.compile(rf'\b{level}\b', re.IGNORECASE)
    for i, line in enumerate(lines, 1):
        if error_pattern.search(line):
            errors.append({
                'line': i,
                'content': line.strip()
            })
    print(f"找到 {len(errors)} 个 {level} 日志:")
    for error in errors[:10]:  # 显示前10条
        print(f"行 {error['line']}: {error['content'][:100]}...")
    return errors
def export_to_json(parsed_logs, output_file):
    """导出为JSON格式"""
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(parsed_logs, f, indent=2, ensure_ascii=False)
    print(f"导出到 {output_file}")
# 命令行接口
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='日志解析脚本')
    parser.add_argument('file', help='日志文件路径')
    parser.add_argument('--format', choices=['apache', 'nginx', 'custom'], default='custom')
    parser.add_argument('--level', default='ERROR', help='过滤级别')
    parser.add_argument('--output', help='输出文件')
    args = parser.parse_args()
    # 执行解析
    log_parser = AdvancedLogParser(log_type=args.format)
    result = log_parser.parse_file_with_stats(args.file)
    # 过滤指定级别
    filtered = [log for log in result['logs'] if log.get('level') == args.level]
    # 输出结果
    if args.output:
        export_to_json(filtered, args.output)
    else:
        for log in filtered[:5]:
            print(log)

使用建议

选择合适的方法：
- 简单场景：使用正则表达式直接解析
- 复杂场景：使用配置驱动的方式
- 大量日志：考虑流式处理
性能优化：
- 预编译正则表达式
- 使用生成器处理大文件
- 考虑并行处理
扩展建议：
- 添加日志聚合统计
- 实现实时监控告警
- 集成到ELK等日志系统

这个脚本框架可以根据您的具体日志格式进行调整和扩展。