try:
    with open(log_file, 'r') as f:
        process_log_lines(f)
except FileNotFoundError:
    logger.error(f"日志文件不存在: {log_file}")
    send_alert(f"日志文件缺失: {log_file}", level="CRITICAL")
except PermissionError:
    logger.error(f"無(wú)權(quán)限訪問(wèn)日志文件: {log_file}")
    send_alert(f"日志文件訪問(wèn)被拒絕: {log_file}", level="CRITICAL")
except IOError as e:
    logger.error(f"讀取日志文件失敗: {str(e)}")
    handle_io_error(log_file, e)

try:
    parsed_log = parse_log_line(log_line)
    if not parsed_log:
        logger.warning(f"無(wú)法解析日志行: {log_line}")
        increment_parse_error_count()
except LogParseError as e:
    logger.error(f"日志解析錯(cuò)誤: {str(e)}")
    handle_parse_error(log_line, e)
except Exception as e:
    logger.error(f"日志解析異常: {str(e)}")
    handle_unexpected_parse_error(log_line, e)

try:
    alert_sender.send(alert_message)
except AlertSendError as e:
    logger.error(f"告警發(fā)送失敗: {str(e)}")
    # 重試機(jī)制
    retry_send_alert(alert_message, max_retries=3)
except NetworkError as e:
    logger.error(f"網(wǎng)絡(luò)連接失敗: {str(e)}")
    handle_network_failure(alert_message, e)
except Exception as e:
    logger.error(f"告警發(fā)送異常: {str(e)}")

try:
    config = load_config(config_file)
    validate_config(config)
except yaml.YAMLError as e:
    logger.error(f"配置文件YAML格式錯(cuò)誤: {str(e)}")
    raise ConfigError(f"配置文件格式無(wú)效: {str(e)}")
except json.JSONDecodeError as e:
    logger.error(f"配置文件JSON格式錯(cuò)誤: {str(e)}")
    raise ConfigError(f"配置文件格式無(wú)效: {str(e)}")
except ValidationError as e:
    logger.error(f"配置驗(yàn)證失敗: {str(e)}")
    raise ConfigError(f"配置無(wú)效: {str(e)}")
except Exception as e:
    logger.error(f"配置加載異常: {str(e)}")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
日志分析和告警工具
功能：實(shí)時(shí)監(jiān)控、分析日志并發(fā)送告警
作者：Cline
版本：1.0
"""

import argparse
import sys
import json
import yaml
import logging
import os
import time
import threading
import re
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional, Callable
from collections import defaultdict, deque
import smtplib
import requests
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('log_analyzer.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

class LogAnalyzerError(Exception):
    """日志分析器異常類(lèi)"""
    pass

class ConfigError(Exception):
    """配置異常類(lèi)"""
    pass

class AlertSendError(Exception):
    """告警發(fā)送異常類(lèi)"""
    pass

class LogEntry:
    """日志條目類(lèi)"""
    def __init__(self, raw_line: str, parsed_data: Dict = None):
        self.raw_line = raw_line
        self.parsed_data = parsed_data or {}
        self.timestamp = parsed_data.get('timestamp') or datetime.now()
        self.level = parsed_data.get('level', 'INFO')
        self.source = parsed_data.get('source', 'unknown')
        self.message = parsed_data.get('message', raw_line)
        
    def to_dict(self):
        """轉(zhuǎn)換為字典"""
        return {
            'raw_line': self.raw_line,
            'parsed_data': self.parsed_data,
            'timestamp': self.timestamp.isoformat() if isinstance(self.timestamp, datetime) else self.timestamp,
            'level': self.level,
            'source': self.source,
            'message': self.message
        }

class LogParser:
    """日志解析器"""
    def __init__(self, format_config: Dict):
        self.format_config = format_config
        self.pattern = re.compile(format_config.get('pattern', '.*'))
        self.fields = format_config.get('fields', [])
        
    def parse(self, log_line: str) -> Optional[LogEntry]:
        """解析日志行"""
        try:
            match = self.pattern.match(log_line.strip())
            if not match:
                return None
                
            parsed_data = {}
            for i, field in enumerate(self.fields):
                if i < len(match.groups()):
                    parsed_data[field] = match.group(i + 1)
                    
            # 處理時(shí)間戳
            if 'timestamp' in parsed_data:
                try:
                    timestamp_str = parsed_data['timestamp']
                    # 嘗試多種時(shí)間格式
                    for fmt in [
                        '%Y-%m-%d %H:%M:%S',
                        '%d/%b/%Y:%H:%M:%S %z',
                        '%Y-%m-%dT%H:%M:%S',
                        '%Y-%m-%d %H:%M:%S.%f'
                    ]:
                        try:
                            parsed_data['timestamp'] = datetime.strptime(timestamp_str, fmt)
                            break
                        except ValueError:
                            continue
                except Exception:
                    pass
                    
            return LogEntry(log_line, parsed_data)
        except Exception as e:
            logger.warning(f"解析日志行失敗: {str(e)}")
            return LogEntry(log_line)

class Rule:
    """告警規(guī)則"""
    def __init__(self, rule_config: Dict):
        self.id = rule_config.get('id')
        self.name = rule_config.get('name', 'Unnamed Rule')
        self.description = rule_config.get('description', '')
        self.enabled = rule_config.get('enabled', True)
        self.pattern = rule_config.get('pattern')
        self.regex = re.compile(rule_config.get('regex', '.*')) if rule_config.get('regex') else None
        self.level = rule_config.get('level', 'INFO')
        self.threshold = rule_config.get('threshold', 1)
        self.window = rule_config.get('window', 60)  # 秒
        self.cooldown = rule_config.get('cooldown', 300)  # 秒
        self.actions = rule_config.get('actions', [])
        self.last_triggered = None
        self.match_count = 0
        self.match_history = deque(maxlen=1000)  # 保留最近1000個(gè)匹配記錄
        
    def should_trigger(self, log_entry: LogEntry) -> bool:
        """判斷是否應(yīng)該觸發(fā)告警"""
        if not self.enabled:
            return False
            
        # 檢查冷卻時(shí)間
        if self.last_triggered and (datetime.now() - self.last_triggered).seconds < self.cooldown:
            return False
            
        # 檢查日志級(jí)別
        if log_entry.level < self.level:
            return False
            
        # 檢查模式匹配
        if self.pattern and self.pattern not in log_entry.message:
            return False
            
        # 檢查正則表達(dá)式
        if self.regex and not self.regex.search(log_entry.message):
            return False
            
        # 記錄匹配
        self.match_history.append({
            'timestamp': datetime.now(),
            'log_entry': log_entry.to_dict()
        })
        
        # 檢查閾值
        if self.threshold > 1:
            # 統(tǒng)計(jì)時(shí)間窗口內(nèi)的匹配次數(shù)
            window_start = datetime.now() - timedelta(seconds=self.window)
            recent_matches = [m for m in self.match_history 
                            if m['timestamp'] >= window_start]
            
            if len(recent_matches) < self.threshold:
                return False
                
        return True
        
    def trigger(self, log_entry: LogEntry) -> Dict:
        """觸發(fā)告警"""
        self.last_triggered = datetime.now()
        self.match_count += 1
        
        alert_data = {
            'rule_id': self.id,
            'rule_name': self.name,
            'timestamp': datetime.now().isoformat(),
            'log_entry': log_entry.to_dict(),
            'match_count': self.match_count,
            'recent_matches': list(self.match_history)[-10:]  # 最近10個(gè)匹配
        }
        
        return alert_data

class AlertSender:
    """告警發(fā)送器"""
    def __init__(self, config: Dict):
        self.config = config
        self.senders = {
            'email': self._send_email,
            'webhook': self._send_webhook,
            'slack': self._send_slack,
            'console': self._send_console
        }
        
    def send(self, alert_data: Dict, action_config: Dict):
        """發(fā)送告警"""
        action_type = action_config.get('type', 'console')
        sender = self.senders.get(action_type)
        
        if not sender:
            raise AlertSendError(f"不支持的告警類(lèi)型: {action_type}")
            
        try:
            sender(alert_data, action_config)
        except Exception as e:
            raise AlertSendError(f"發(fā)送告警失敗: {str(e)}")
            
    def _send_email(self, alert_data: Dict, config: Dict):
        """發(fā)送郵件告警"""
        try:
            msg = MIMEMultipart()
            msg['From'] = config.get('sender')
            msg['To'] = ', '.join(config.get('recipients', []))
            msg['Subject'] = f"日志告警 - {alert_data['rule_name']}"
            
            body = f"""
日志告警通知

規(guī)則名稱(chēng): {alert_data['rule_name']}
規(guī)則ID: {alert_data['rule_id']}
觸發(fā)時(shí)間: {alert_data['timestamp']}
匹配次數(shù): {alert_data['match_count']}

日志內(nèi)容:
{alert_data['log_entry']['raw_line']}

---
日志分析器
            """
            msg.attach(MIMEText(body, 'plain'))
            
            server = smtplib.SMTP(config.get('smtp_server'), config.get('smtp_port', 587))
            server.starttls()
            server.login(config.get('sender'), config.get('password'))
            server.send_message(msg)
            server.quit()
            
            logger.info(f"郵件告警已發(fā)送: {alert_data['rule_name']}")
            
        except Exception as e:
            logger.error(f"發(fā)送郵件告警失敗: {str(e)}")
            raise
            
    def _send_webhook(self, alert_data: Dict, config: Dict):
        """發(fā)送Webhook告警"""
        try:
            response = requests.post(
                config.get('url'),
                json=alert_data,
                headers=config.get('headers', {}),
                timeout=config.get('timeout', 30)
            )
            response.raise_for_status()
            
            logger.info(f"Webhook告警已發(fā)送: {alert_data['rule_name']}")
            
        except Exception as e:
            logger.error(f"發(fā)送Webhook告警失敗: {str(e)}")
            raise
            
    def _send_slack(self, alert_data: Dict, config: Dict):
        """發(fā)送Slack告警"""
        try:
            payload = {
                'text': f"日志告警: {alert_data['rule_name']}",
                'attachments': [{
                    'color': 'danger',
                    'fields': [
                        {
                            'title': '規(guī)則名稱(chēng)',
                            'value': alert_data['rule_name'],
                            'short': True
                        },
                        {
                            'title': '觸發(fā)時(shí)間',
                            'value': alert_data['timestamp'],
                            'short': True
                        },
                        {
                            'title': '日志內(nèi)容',
                            'value': alert_data['log_entry']['raw_line'],
                            'short': False
                        }
                    ]
                }]
            }
            
            response = requests.post(
                config.get('webhook_url'),
                json=payload,
                timeout=config.get('timeout', 30)
            )
            response.raise_for_status()
            
            logger.info(f"Slack告警已發(fā)送: {alert_data['rule_name']}")
            
        except Exception as e:
            logger.error(f"發(fā)送Slack告警失敗: {str(e)}")
            raise
            
    def _send_console(self, alert_data: Dict, config: Dict):
        """發(fā)送控制臺(tái)告警"""
        print(f"[ALERT] {alert_data['rule_name']}: {alert_data['log_entry']['message']}")

class LogFileMonitor:
    """日志文件監(jiān)控器"""
    def __init__(self, file_path: str, parser: LogParser):
        self.file_path = file_path
        self.parser = parser
        self.file_handle = None
        self.file_position = 0
        self.callbacks = []
        
    def add_callback(self, callback: Callable):
        """添加回調(diào)函數(shù)"""
        self.callbacks.append(callback)
        
    def start(self):
        """開(kāi)始監(jiān)控"""
        try:
            # 打開(kāi)文件并移動(dòng)到末尾
            self.file_handle = open(self.file_path, 'r', encoding='utf-8', errors='ignore')
            self.file_handle.seek(0, 2)  # 移動(dòng)到文件末尾
            self.file_position = self.file_handle.tell()
            
            # 啟動(dòng)監(jiān)控線程
            thread = threading.Thread(target=self._monitor_loop)
            thread.daemon = True
            thread.start()
            
            logger.info(f"開(kāi)始監(jiān)控日志文件: {self.file_path}")
            
        except Exception as e:
            logger.error(f"啟動(dòng)日志監(jiān)控失敗: {str(e)}")
            raise LogAnalyzerError(f"啟動(dòng)監(jiān)控失敗: {str(e)}")
            
    def stop(self):
        """停止監(jiān)控"""
        if self.file_handle:
            self.file_handle.close()
            self.file_handle = None
        logger.info(f"停止監(jiān)控日志文件: {self.file_path}")
        
    def _monitor_loop(self):
        """監(jiān)控循環(huán)"""
        while self.file_handle:
            try:
                # 檢查文件是否有新內(nèi)容
                current_position = self.file_handle.tell()
                self.file_handle.seek(0, 2)
                file_size = self.file_handle.tell()
                self.file_handle.seek(current_position)
                
                if file_size > current_position:
                    # 讀取新內(nèi)容
                    new_lines = self.file_handle.readlines()
                    for line in new_lines:
                        self._process_line(line)
                        
                # 更新文件位置
                self.file_position = self.file_handle.tell()
                
                time.sleep(0.1)  # 短暫休眠
                
            except Exception as e:
                logger.error(f"日志監(jiān)控異常: {str(e)}")
                time.sleep(1)
                
    def _process_line(self, line: str):
        """處理日志行"""
        try:
            log_entry = self.parser.parse(line)
            if log_entry:
                for callback in self.callbacks:
                    try:
                        callback(log_entry)
                    except Exception as e:
                        logger.error(f"回調(diào)執(zhí)行失敗: {str(e)}")
        except Exception as e:
            logger.warning(f"處理日志行失敗: {str(e)}")

class DatabaseManager:
    """數(shù)據(jù)庫(kù)管理器"""
    def __init__(self, db_path: str = 'log_analyzer.db'):
        self.db_path = db_path
        self.init_database()
        
    def init_database(self):
        """初始化數(shù)據(jù)庫(kù)"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            # 創(chuàng)建告警記錄表
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS alerts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT NOT NULL,
                    rule_id TEXT NOT NULL,
                    rule_name TEXT NOT NULL,
                    log_entry TEXT,
                    match_count INTEGER
                )
            ''')
            
            # 創(chuàng)建統(tǒng)計(jì)信息表
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS statistics (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT NOT NULL,
                    rule_id TEXT NOT NULL,
                    match_count INTEGER,
                    processed_lines INTEGER
                )
            ''')
            
            conn.commit()
            conn.close()
            logger.info("數(shù)據(jù)庫(kù)初始化完成")
            
        except Exception as e:
            logger.error(f"數(shù)據(jù)庫(kù)初始化失敗: {str(e)}")
            
    def save_alert(self, alert_data: Dict):
        """保存告警記錄"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                INSERT INTO alerts 
                (timestamp, rule_id, rule_name, log_entry, match_count)
                VALUES (?, ?, ?, ?, ?)
            ''', (
                alert_data.get('timestamp', ''),
                alert_data.get('rule_id', ''),
                alert_data.get('rule_name', ''),
                json.dumps(alert_data.get('log_entry', {})),
                alert_data.get('match_count', 0)
            ))
            
            conn.commit()
            conn.close()
            logger.info(f"告警記錄已保存: {alert_data.get('rule_name', '')}")
            
        except Exception as e:
            logger.error(f"保存告警記錄失敗: {str(e)}")
            
    def get_alerts(self, limit: int = 100) -> List[Dict]:
        """獲取告警記錄"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                SELECT timestamp, rule_id, rule_name, log_entry, match_count
                FROM alerts
                ORDER BY timestamp DESC
                LIMIT ?
            ''', (limit,))
            
            rows = cursor.fetchall()
            conn.close()
            
            alerts = []
            for row in rows:
                alerts.append({
                    'timestamp': row[0],
                    'rule_id': row[1],
                    'rule_name': row[2],
                    'log_entry': json.loads(row[3]) if row[3] else {},
                    'match_count': row[4]
                })
                
            return alerts
            
        except Exception as e:
            logger.error(f"獲取告警記錄失敗: {str(e)}")
            return []

class LogAnalyzer:
    """日志分析器主類(lèi)"""
    def __init__(self, config_file: str = None):
        self.config_file = config_file
        self.config = {}
        self.rules = []
        self.parsers = {}
        self.monitors = []
        self.alert_sender = None
        self.db_manager = None
        self.running = False
        self.processed_lines = 0
        self.match_statistics = defaultdict(int)
        
        # 加載配置
        self.load_config()
        
        # 初始化組件
        self._init_components()
        
    def load_config(self):
        """加載配置文件"""
        if not self.config_file or not os.path.exists(self.config_file):
            logger.info("未指定配置文件或文件不存在，使用默認(rèn)配置")
            self.config = self._create_default_config()
            return
            
        try:
            with open(self.config_file, 'r', encoding='utf-8') as f:
                if self.config_file.endswith('.yaml') or self.config_file.endswith('.yml'):
                    self.config = yaml.safe_load(f)
                else:
                    self.config = json.load(f)
                    
            logger.info(f"成功加載配置文件: {self.config_file}")
            
        except Exception as e:
            logger.error(f"加載配置文件失敗: {str(e)}")
            raise ConfigError(f"配置加載失敗: {str(e)}")
            
    def _create_default_config(self) -> Dict:
        """創(chuàng)建默認(rèn)配置"""
        return {
            "parsers": {
                "default": {
                    "pattern": "(.*)",
                    "fields": ["message"]
                }
            },
            "rules": [
                {
                    "id": "error_detector",
                    "name": "錯(cuò)誤檢測(cè)器",
                    "description": "檢測(cè)包含ERROR關(guān)鍵字的日志",
                    "enabled": True,
                    "pattern": "ERROR",
                    "level": "ERROR",
                    "threshold": 1,
                    "window": 60,
                    "cooldown": 300,
                    "actions": [
                        {"type": "console"}
                    ]
                }
            ],
            "sources": [
                {
                    "path": "/var/log/syslog",
                    "parser": "default",
                    "enabled": False
                }
            ],
            "actions": {
                "email": {
                    "type": "email",
                    "sender": "loganalyzer@example.com",
                    "password": "your_password",
                    "smtp_server": "smtp.example.com",
                    "smtp_port": 587,
                    "recipients": ["admin@example.com"]
                }
            }
        }
        
    def _init_components(self):
        """初始化組件"""
        # 初始化解析器
        for name, parser_config in self.config.get('parsers', {}).items():
            self.parsers[name] = LogParser(parser_config)
            
        # 初始化規(guī)則
        for rule_config in self.config.get('rules', []):
            self.rules.append(Rule(rule_config))
            
        # 初始化告警發(fā)送器
        self.alert_sender = AlertSender(self.config.get('actions', {}))
        
        # 初始化數(shù)據(jù)庫(kù)管理器
        self.db_manager = DatabaseManager()
        
    def start(self):
        """啟動(dòng)日志分析器"""
        if self.running:
            logger.warning("日志分析器已在運(yùn)行")
            return
            
        logger.info("啟動(dòng)日志分析器...")
        self.running = True
        
        # 啟動(dòng)日志監(jiān)控器
        for source_config in self.config.get('sources', []):
            if not source_config.get('enabled', False):
                continue
                
            path = source_config.get('path')
            parser_name = source_config.get('parser', 'default')
            
            if not os.path.exists(path):
                logger.warning(f"日志文件不存在: {path}")
                continue
                
            if parser_name not in self.parsers:
                logger.warning(f"解析器不存在: {parser_name}")
                continue
                
            try:
                parser = self.parsers[parser_name]
                monitor = LogFileMonitor(path, parser)
                monitor.add_callback(self._process_log_entry)
                monitor.start()
                self.monitors.append(monitor)
                logger.info(f"已啟動(dòng)日志監(jiān)控: {path}")
                
            except Exception as e:
                logger.error(f"啟動(dòng)日志監(jiān)控失敗 {path}: {str(e)}")
                
        logger.info("日志分析器啟動(dòng)完成")
        
    def stop(self):
        """停止日志分析器"""
        logger.info("停止日志分析器...")
        self.running = False
        
        # 停止所有監(jiān)控器
        for monitor in self.monitors:
            try:
                monitor.stop()
            except Exception as e:
                logger.error(f"停止監(jiān)控器失敗: {str(e)}")
                
        self.monitors.clear()
        logger.info("日志分析器已停止")
        
    def _process_log_entry(self, log_entry: LogEntry):
        """處理日志條目"""
        self.processed_lines += 1
        
        # 應(yīng)用所有規(guī)則
        for rule in self.rules:
            try:
                if rule.should_trigger(log_entry):
                    # 觸發(fā)告警
                    alert_data = rule.trigger(log_entry)
                    
                    # 保存到數(shù)據(jù)庫(kù)
                    if self.db_manager:
                        self.db_manager.save_alert(alert_data)
                        
                    # 更新統(tǒng)計(jì)信息
                    self.match_statistics[rule.id] += 1
                    
                    # 發(fā)送告警
                    for action_config in rule.actions:
                        try:
                            self.alert_sender.send(alert_data, action_config)
                        except Exception as e:
                            logger.error(f"發(fā)送告警失敗: {str(e)}")
                            
            except Exception as e:
                logger.error(f"應(yīng)用規(guī)則 {rule.name} 失敗: {str(e)}")
                
    def get_statistics(self) -> Dict:
        """獲取統(tǒng)計(jì)信息"""
        return {
            'processed_lines': self.processed_lines,
            'match_statistics': dict(self.match_statistics),
            'active_monitors': len(self.monitors),
            'active_rules': len([r for r in self.rules if r.enabled])
        }
        
    def add_rule(self, rule_config: Dict):
        """動(dòng)態(tài)添加規(guī)則"""
        try:
            rule = Rule(rule_config)
            self.rules.append(rule)
            logger.info(f"已添加規(guī)則: {rule.name}")
        except Exception as e:
            logger.error(f"添加規(guī)則失敗: {str(e)}")
            
    def remove_rule(self, rule_id: str):
        """移除規(guī)則"""
        self.rules = [r for r in self.rules if r.id != rule_id]
        logger.info(f"已移除規(guī)則: {rule_id}")
        
    def get_alerts(self, limit: int = 100) -> List[Dict]:
        """獲取告警記錄"""
        if self.db_manager:
            return self.db_manager.get_alerts(limit)
        return []

def create_sample_config():
    """創(chuàng)建示例配置文件"""
    sample_config = {
        "parsers": {
            "nginx": {
                "pattern": r'(\S+) - - \[(.*?)\] "(\S+) (\S+) (\S+)" (\d+) (\d+) "(.*?)" "(.*?)"',
                "fields": ["ip", "timestamp", "method", "url", "protocol", "status", "size", "referer", "user_agent"]
            },
            "syslog": {
                "pattern": r'<(\d+)>(\d+) (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+[+-]\d{2}:\d{2}) (\S+) ([^ ]+) ([^ ]+) (.*)',
                "fields": ["priority", "version", "timestamp", "hostname", "app_name", "proc_id", "message"]
            }
        },
        "rules": [
            {
                "id": "nginx_404_errors",
                "name": "Nginx 404錯(cuò)誤檢測(cè)",
                "description": "檢測(cè)Nginx訪問(wèn)日志中的404錯(cuò)誤",
                "enabled": True,
                "regex": r'" 404 ',
                "level": "WARNING",
                "threshold": 5,
                "window": 60,
                "cooldown": 300,
                "actions": [
                    {"type": "console"},
                    {"type": "email"}
                ]
            },
            {
                "id": "security_login_failures",
                "name": "安全登錄失敗檢測(cè)",
                "description": "檢測(cè)認(rèn)證失敗的日志",
                "enabled": True,
                "pattern": "authentication failure",
                "level": "ERROR",
                "threshold": 3,
                "window": 300,
                "cooldown": 600,
                "actions": [
                    {"type": "console"},
                    {"type": "webhook", "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"}
                ]
            }
        ],
        "sources": [
            {
                "path": "/var/log/nginx/access.log",
                "parser": "nginx",
                "enabled": False
            },
            {
                "path": "/var/log/syslog",
                "parser": "syslog",
                "enabled": False
            }
        ],
        "actions": {
            "email": {
                "type": "email",
                "sender": "loganalyzer@example.com",
                "password": "your_app_password",
                "smtp_server": "smtp.gmail.com",
                "smtp_port": 587,
                "recipients": ["admin@example.com", "security@example.com"]
            },
            "slack_webhook": {
                "type": "webhook",
                "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
                "timeout": 30
            }
        }
    }
    
    with open('log_analyzer_sample_config.json', 'w', encoding='utf-8') as f:
        json.dump(sample_config, f, indent=2, ensure_ascii=False)
    logger.info("示例配置文件已創(chuàng)建: log_analyzer_sample_config.json")

def main():
    parser = argparse.ArgumentParser(description='日志分析和告警工具')
    parser.add_argument('-c', '--config', help='配置文件路徑')
    parser.add_argument('--start', action='store_true', help='啟動(dòng)日志分析器')
    parser.add_argument('--sample-config', action='store_true', help='創(chuàng)建示例配置文件')
    parser.add_argument('--stats', action='store_true', help='顯示統(tǒng)計(jì)信息')
    parser.add_argument('--alerts', action='store_true', help='顯示最近告警')
    parser.add_argument('--limit', type=int, default=10, help='告警記錄數(shù)量限制')
    
    args = parser.parse_args()
    
    if args.sample_config:
        create_sample_config()
        return
        
    analyzer = LogAnalyzer(args.config)
    
    if args.stats:
        stats = analyzer.get_statistics()
        print(json.dumps(stats, indent=2, ensure_ascii=False))
        return
        
    if args.alerts:
        alerts = analyzer.get_alerts(args.limit)
        print(json.dumps(alerts, indent=2, ensure_ascii=False))
        return
        
    if args.start:
        try:
            analyzer.start()
            # 保持程序運(yùn)行
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            logger.info("收到中斷信號(hào)，正在停止分析器...")
        finally:
            analyzer.stop()
    else:
        parser.print_help()

if __name__ == '__main__':
    main()

pip install pyyaml requests

python log_analyzer.py --sample-config

python log_analyzer.py --config log_analyzer_config.json --start

python log_analyzer.py --stats

python log_analyzer.py --alerts --limit 20

{
  "parsers": {
    "nginx": {
      "pattern": "(\\S+) - - \\[(.*?)\\] \"(\\S+) (\\S+) (\\S+)\" (\\d+) (\\d+) \"(.*?)\" \"(.*?)\"",
      "fields": ["ip", "timestamp", "method", "url", "protocol", "status", "size", "referer", "user_agent"]
    },
    "syslog": {
      "pattern": "<(\\d+)>(\\d+) (\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d+[+-]\\d{2}:\\d{2}) (\\S+) ([^ ]+) ([^ ]+) (.*)",
      "fields": ["priority", "version", "timestamp", "hostname", "app_name", "proc_id", "message"]
    }
  },
  "rules": [
    {
      "id": "nginx_404_errors",
      "name": "Nginx 404錯(cuò)誤檢測(cè)",
      "description": "檢測(cè)Nginx訪問(wèn)日志中的404錯(cuò)誤",
      "enabled": true,
      "regex": "\" 404 ",
      "level": "WARNING",
      "threshold": 5,
      "window": 60,
      "cooldown": 300,
      "actions": [
        {"type": "console"},
        {"type": "email"}
      ]
    },
    {
      "id": "security_login_failures",
      "name": "安全登錄失敗檢測(cè)",
      "description": "檢測(cè)認(rèn)證失敗的日志",
      "enabled": true,
      "pattern": "authentication failure",
      "level": "ERROR",
      "threshold": 3,
      "window": 300,
      "cooldown": 600,
      "actions": [
        {"type": "console"},
        {"type": "webhook", "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"}
      ]
    }
  ],
  "sources": [
    {
      "path": "/var/log/nginx/access.log",
      "parser": "nginx",
      "enabled": false
    },
    {
      "path": "/var/log/syslog",
      "parser": "syslog",
      "enabled": false
    }
  ],
  "actions": {
    "email": {
      "type": "email",
      "sender": "loganalyzer@example.com",
      "password": "your_app_password",
      "smtp_server": "smtp.gmail.com",
      "smtp_port": 587,
      "recipients": ["admin@example.com", "security@example.com"]
    },
    "slack_webhook": {
      "type": "webhook",
      "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
      "timeout": 30
    }
  }
}