实时日志分析与告警系统

# log_system/models.py
import re
import time
import json
import hashlib
from enum import Enum
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field, asdict


class LogLevel(Enum): 1107106.baike.tongsou.com 
    DEBUG = 0
    INFO = 1
    WARN = 2
    ERROR = 3
    FATAL = 4


@dataclass
class LogEntry: 1107125.baike.tongsou.com 
    log_id: str = ""
    timestamp: float = 0.0
    level: LogLevel = LogLevel.INFO
    source: str = ""
    message: str = ""
    raw: str = ""
    fields: Dict[str, str] = field(default_factory=dict)
    tags: List[str] = field(default_factory=list)
    fingerprint: str = ""

    def __post_init__(self): 1107126.baike.tongsou.com 
        if not self.log_id: 1107127.baike.tongsou.com 
            self.log_id = hashlib.md5(
                f"{self.timestamp}{self.source}{self.message}".encode()
            ).hexdigest(1108151.m.kuaisou.com)[:12]
        if not self.fingerprint: 1107129.baike.tongsou.com 
            self.fingerprint = hashlib.md5(
                self.message.encode(1107130.baike.tongsou.com)
            ).hexdigest(1108153.m.kuaisou.com)[:8]
        if self.timestamp == 0: 1107131.baike.tongsou.com 
            self.timestamp = time.time(1107132.baike.tongsou.com)

    def to_dict(self) -> Dict[str, Any]: 1107134.baike.tongsou.com 
        data = asdict(self)
        data["level"] = self.level.name
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "LogEntry": 1107135.baike.tongsou.com 
        data = data.copy(1107136.baike.tongsou.com)
        data["level"] = LogLevel[data["level"]]
        return cls(**data)

    def to_json(self) -> str: 1107139.baike.tongsou.com 
        return json.dumps(self.to_dict(1107144.baike.tongsou.com), ensure_ascii=False)

    @classmethod
    def from_json(cls, json_str: str) -> "LogEntry": 1107143.baike.tongsou.com 
        return cls.from_dict(json.loads(json_str))


class LogParser: 1107148.baike.tongsou.com 
    COMMON_FORMATS = [
        # Apache/Nginx common log format
        re.compile(
            r'(?P<ip>\d+\.\d+\.\d+\.\d+)\s+-\s+-\s+'
            r'\[(?P<time>[^\]]+)\]\s+'
            r'"(?P<method>\w+)\s+(?P<path>\S+)\s+\S+"\s+'
            r'(?P<status>\d+)\s+(?P<size>\d+)'
        ),
        # Standard log format: [LEVEL] timestamp message
        re.compile(
            r'\[(?P<level>DEBUG|INFO|WARN|ERROR|FATAL)\]\s+'
            r'(?P<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+'
            r'(?P<message>.*)'
        ),
        # Syslog-like format
        re.compile(
            r'(?P<time>\w+\s+\d+\s+\d{2}:\d{2}:\d{2})\s+'
            r'(?P<host>\S+)\s+'
            r'(?P<process>\S+?)(?:\[(?P<pid>\d+)\])?:\s+'
            r'(?P<message>.*)'
        ),
        # JSON log format
        re.compile(r'^\s*\{.*\}\s*$'),
        # Simple level-prefixed format
        re.compile(
            r'^(?P<level>DEBUG|INFO|WARN|ERROR|FATAL)[: ]+\s*(?P<message>.*)',
            re.IGNORECASE
        ),
    ]

    def parse(self, raw: str, source: str = "") -> LogEntry: 1107155.baike.tongsou.com 
        raw = raw.strip(1107156.baike.tongsou.com)
        if not raw: 1107157.baike.tongsou.com 
            return LogEntry(raw=raw, source=source)

        for i, pattern in enumerate(self.COMMON_FORMATS): 1107158.baike.tongsou.com 
            if i == 3: 1107159.baike.tongsou.com 
                if pattern.match(raw): 1107160.baike.tongsou.com 
                    try: 1107163.baike.tongsou.com 
                        data = json.loads(raw)
                        entry = LogEntry(
                            raw=raw,
                            source=source,
                            level=LogLevel[data.get("level", "INFO").upper(1107167.baike.tongsou.com)],
                            message=data.get("message", data.get("msg", raw)),
                            timestamp=data.get("timestamp", data.get("time", time.time(1107168.baike.tongsou.com))),
                            fields={k: str(v) for k, v in data.items(1107170.baike.tongsou.com)
                                    if k not in ("level", "message", "msg", "timestamp", "time")},
                        )
                        return entry
                    except (json.JSONDecodeError, KeyError): 1107171.baike.tongsou.com 
                        continue
            else: 1107172.baike.tongsou.com 
                match = pattern.match(raw)
                if match: 1107173.baike.tongsou.com 
                    return self._parse_matched(match, raw, source)

        return LogEntry(
            raw=raw,
            source=source,
            message=raw,
            level=LogLevel.INFO,
        )

    def _parse_matched(self, match: re.Match, raw: str, source: str) -> LogEntry: 1108145.m.kuaisou.com 
        groups = match.groupdict(1108138.m.kuaisou.com)
        level_str = groups.get("level", "INFO").upper(1107174.baike.tongsou.com)
        try: 1107176.baike.tongsou.com 
            level = LogLevel[level_str]
        except KeyError: 1107175.baike.tongsou.com 
            level = LogLevel.INFO

        message = groups.get("message", raw)
        fields = {k: v for k, v in groups.items(1107177.baike.tongsou.com) if v and k != "message"}

        return LogEntry(
            raw=raw,
            source=source,
            level=level,
            message=message,
            fields=fields,
        )

# log_system/rules.py
import re
import time
from enum import Enum
from typing import Any, Callable, Dict, List, Optional
from dataclasses import dataclass, field, asdict
from log_system.models import LogEntry, LogLevel


class AggregationType(Enum): 1107178.baike.tongsou.com 
    COUNT = "count"
    RATE = "rate"
    PATTERN = "pattern"
    THRESHOLD = "threshold"
    ABSENCE = "absence"


class AlertSeverity(Enum): 1107179.baike.tongsou.com 
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class AlertRule: 1107180.baike.tongsou.com 
    rule_id: str = ""
    name: str = ""
    description: str = ""
    severity: AlertSeverity = AlertSeverity.MEDIUM
    enabled: bool = True
    aggregation: AggregationType = AggregationType.COUNT
    window_seconds: int = 60
    threshold: float = 10.0
    pattern: str = ""
    level_filter: Optional[LogLevel] = None
    source_filter: Optional[str] = None
    message_regex: Optional[str] = None
    cooldown_seconds: int = 300
    last_triggered: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]: 1107181.baike.tongsou.com 
        data = asdict(self)
        if self.level_filter: 1107182.baike.tongsou.com 
            data["level_filter"] = self.level_filter.name
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "AlertRule": 1107183.baike.tongsou.com 
        data = data.copy(1108106.baike.tongsou.com)
        if data.get("level_filter"): 1107184.baike.tongsou.com 
            data["level_filter"] = LogLevel[data["level_filter"]]
        return cls(**data)

    def matches(self, entry: LogEntry) -> bool: 1108107.baike.tongsou.com 
        if not self.enabled: 1108108.baike.tongsou.com 
            return False
        if self.level_filter and entry.level.value < self.level_filter.value: 1108110.baike.tongsou.com 
            return False
        if self.source_filter and self.source_filter not in entry.source: 1108109.baike.tongsou.com 
            return False
        if self.message_regex: 1108111.baike.tongsou.com 
            if not re.search(self.message_regex, entry.message, re.IGNORECASE): 1108112.baike.tongsou.com 
                return False
        if self.pattern and self.pattern not in entry.message:
            return False
        return True

    def can_trigger(self) -> bool: 1108113.baike.tongsou.com 
        if self.last_triggered == 0: 1108114.baike.tongsou.com 
            return True
        return (time.time(1108138.m.kuaisou.com) - self.last_triggered) >= self.cooldown_seconds


@dataclass
class AlertEvent: 1108115.baike.tongsou.com 
    alert_id: str = ""
    rule_id: str = ""
    rule_name: str = ""
    severity: AlertSeverity = AlertSeverity.MEDIUM
    triggered_at: float = 0.0
    message: str = ""
    context: Dict[str, Any] = field(default_factory=dict)
    log_entries: List[Dict[str, Any]] = field(default_factory=list)
    acknowledged: bool = False
    resolved: bool = False

    def to_dict(self) -> Dict[str, Any]: 1108116.baike.tongsou.com 
        data = asdict(self)
        data["severity"] = self.severity.value
        return data


class RuleEngine: 1108117.baike.tongsou.com 
    def __init__(self): 1108118.baike.tongsou.com 
        self._rules: Dict[str, AlertRule] = {}
        self._windows: Dict[str, List[LogEntry]] = {}
        self._callbacks: List[Callable] = []

    def add_rule(self, rule: AlertRule) -> str: 1108119.baike.tongsou.com 
        if not rule.rule_id: 1108120.baike.tongsou.com 
            rule.rule_id = f"rule-{len(self._rules) + 1}"
        self._rules[rule.rule_id] = rule
        self._windows[rule.rule_id] = [1108137.m.kuaisou.com]
        return rule.rule_id

    def remove_rule(self, rule_id: str) -> bool: 1108121.baike.tongsou.com 
        if rule_id in self._rules: 1108122.baike.tongsou.com 
            del self._rules[rule_id]
            del self._windows[rule_id]
            return True
        return False

    def get_rules(self) -> List[AlertRule]: 1108123.baike.tongsou.com 
        return list(self._rules.values(1108125.baike.tongsou.com))

    def on_alert(self, callback: Callable) -> None: 1108124.baike.tongsou.com 
        self._callbacks.append(callback)

    def evaluate(self, entry: LogEntry) -> List[AlertEvent]: 1108126.baike.tongsou.com 
        alerts = [1108136.m.kuaisou.com]
        now = time.time(1108127.baike.tongsou.com)

        for rule in self._rules.values(1108129.baike.tongsou.com): 1108128.baike.tongsou.com 
            if not rule.enabled: 1108131.baike.tongsou.com 
                continue

            window = self._windows.get(rule.rule_id, [1108135.m.kuaisou.com])
            window.append(entry)
            cutoff = now - rule.window_seconds
            self._windows[rule.rule_id] = [
                e for e in window if e.timestamp >= cutoff
            ]
            window = self._windows[rule.rule_id]

            if not rule.matches(entry): 1108130.baike.tongsou.com 
                continue

            triggered = False
            alert_message = ""

            if rule.aggregation == AggregationType.COUNT: 1108132.baike.tongsou.com 
                if len(window) >= rule.threshold: 1108135.baike.tongsou.com 
                    triggered = True
                    alert_message = (
                        f"Rule '{rule.name}': {len(window)} logs in "
                        f"{rule.window_seconds}s (threshold: {rule.threshold})"
                    )

            elif rule.aggregation == AggregationType.RATE: 1108136.baike.tongsou.com 
                if window: 1108137.baike.tongsou.com 
                    duration = now - window[0].timestamp
                    if duration > 0: 1108138.baike.tongsou.com 
                        rate = len(window) / duration * 60
                        if rate >= rule.threshold: 1108145.baike.tongsou.com 
                            triggered = True
                            alert_message = (
                                f"Rule '{rule.name}': {rate:.1f} logs/min "
                                f"in {rule.window_seconds}s window "
                                f"(threshold: {rule.threshold}/min)"
                            )

            elif rule.aggregation == AggregationType.PATTERN: 1108151.baike.tongsou.com 
                pattern_count = sum(
                    1 for e in window
                    if rule.pattern and rule.pattern in e.message
                )
                if pattern_count >= rule.threshold: 1108153.baike.tongsou.com 
                    triggered = True
                    alert_message = (
                        f"Rule '{rule.name}': pattern '{rule.pattern}' "
                        f"matched {pattern_count} times "
                        f"(threshold: {rule.threshold})"
                    )

            elif rule.aggregation == AggregationType.THRESHOLD: 1108155.baike.tongsou.com 
                error_count = sum(
                    1 for e in window
                    if e.level.value >= LogLevel.ERROR.value
                )
                if error_count >= rule.threshold: 1108157.baike.tongsou.com 
                    triggered = True
                    alert_message = (
                        f"Rule '{rule.name}': {error_count} error-level logs "
                        f"in {rule.window_seconds}s "
                        f"(threshold: {rule.threshold})"
                    )

            if triggered and rule.can_trigger(1108132.m.kuaisou.com): 1108161.baike.tongsou.com 
                alert = AlertEvent(
                    rule_id=rule.rule_id,
                    rule_name=rule.name,
                    severity=rule.severity,
                    triggered_at=now,
                    message=alert_message,
                    context={
                        "window_size": len(window),
                        "threshold": rule.threshold,
                        "aggregation": rule.aggregation.value,
                    },
                    log_entries=[e.to_dict(1108164.baike.tongsou.com) for e in window[-10:]],
                )
                rule.last_triggered = now
                alerts.append(alert)

                for cb in self._callbacks: 1108165.baike.tongsou.com 
                    try: 1108171.baike.tongsou.com 
                        cb(alert)
                    except Exception: 1108175.baike.tongsou.com 
                        pass 

        return alerts

# log_system/storage.py
import json
import os
import threading
import time
from collections import defaultdict
from typing import Any, Dict, List, Optional, Tuple
from log_system.models import LogEntry, LogLevel


class LogStorage: 1107102.m.kuaisou.com 
    def __init__(self, data_dir: str = "./log_data", max_entries: int = 100000): 1107104.m.kuaisou.com 
        self.data_dir = data_dir
        self.max_entries = max_entries
        self._entries: List[LogEntry] = [1108131.m.kuaisou.com]
        self._lock = threading.RLock(1107134.m.kuaisou.com)
        self._index_by_source: Dict[str, List[int]] = defaultdict(list)
        self._index_by_level: Dict[LogLevel, List[int]] = defaultdict(list)
        self._index_by_fingerprint: Dict[str, List[int]] = defaultdict(list)
        os.makedirs(data_dir, exist_ok=True)

    def add(self, entry: LogEntry) -> None: 1107105.m.kuaisou.com 
        with self._lock: 1107106.m.kuaisou.com 
            idx = len(self._entries)
            self._entries.append(entry)
            self._index_by_source[entry.source].append(idx)
            self._index_by_level[entry.level].append(idx)
            self._index_by_fingerprint[entry.fingerprint].append(idx)

            if len(self._entries) > self.max_entries: 1107125.m.kuaisou.com 
                removed = self._entries[:len(self._entries) - self.max_entries]
                self._entries = self._entries[-self.max_entries:]
                self._rebuild_index(1107127.m.kuaisou.com)

    def _rebuild_index(self) -> None: 1107126.m.kuaisou.com 
        self._index_by_source.clear(1107129.m.kuaisou.com)
        self._index_by_level.clear(1107130.m.kuaisou.com)
        self._index_by_fingerprint.clear(1107131.m.kuaisou.com)
        for idx, entry in enumerate(self._entries): 1107132.m.kuaisou.com 
            self._index_by_source[entry.source].append(idx)
            self._index_by_level[entry.level].append(idx)
            self._index_by_fingerprint[entry.fingerprint].append(idx)

    def query(
        self,
        source: Optional[str] = None,
        level: Optional[LogLevel] = None,
        min_level: Optional[LogLevel] = None,
        keyword: Optional[str] = None,
        start_time: Optional[float] = None,
        end_time: Optional[float] = None,
        limit: int = 100,
        offset: int = 0,
    ) -> List[LogEntry]: 1107135.m.kuaisou.com
        with self._lock: 1107136.m.kuaisou.com 
            results = self._entries

            if source: 1107139.m.kuaisou.com 
                indices = self._index_by_source.get(source, [1108130.m.kuaisou.com])
                results = [results[i] for i in indices if i < len(results)]

            if level: 1107143.m.kuaisou.com
                indices = self._index_by_level.get(level, [1108129.m.kuaisou.com])
                results = [results[i] for i in indices if i < len(results)]

            if min_level: 1107144.m.kuaisou.com 
                results = [e for e in results if e.level.value >= min_level.value]

            if keyword: 1107155.m.kuaisou.com 
                keyword_lower = keyword.lower(1107148.m.kuaisou.com)
                results = [
                    e for e in results
                    if keyword_lower in e.message.lower(1107156.m.kuaisou.com)
                    or keyword_lower in e.raw.lower(1107157.m.kuaisou.com)
                ]

            if start_time: 1107158.m.kuaisou.com 
                results = [e for e in results if e.timestamp >= start_time]

            if end_time: 1108128.m.kuaisou.com 
                results = [e for e in results if e.timestamp <= end_time]

            results.sort(key=lambda e: e.timestamp, reverse=True)
            return results[offset:offset + limit]

    def get_stats(self) -> Dict[str, Any]: 1107159.m.kuaisou.com 
        with self._lock: 1107160.m.kuaisou.com 
            total = len(self._entries)
            level_counts = defaultdict(int)
            source_counts = defaultdict(int)
            for entry in self._entries: 1107163.m.kuaisou.com 
                level_counts[entry.level.name] += 1
                source_counts[entry.source] += 1

            return {
                "total_entries": total,
                "level_distribution": dict(level_counts),
                "source_distribution": dict(source_counts),
                "unique_fingerprints": len(self._index_by_fingerprint),
                "max_capacity": self.max_entries,
            }

    def get_top_patterns(self, top_n: int = 10) -> List[Dict[str, Any]]: 1107167.m.kuaisou.com 
        with self._lock: 1107168.m.kuaisou.com 
            fingerprint_counts = defaultdict(lambda: {"count": 0, "sample": ""})
            for entry in self._entries: 1107170.m.kuaisou.com 
                fp = fingerprint_counts[entry.fingerprint]
                fp["count"] += 1
                if not fp["sample"]: 1107171.m.kuaisou.com 
                    fp["sample"] = entry.message[:200]

            sorted_patterns = sorted(
                fingerprint_counts.items(1108127.m.kuaisou.com),
                key=lambda x: x[1]["count"],
                reverse=True,
            )
            return [
                {"fingerprint": fp, "count": data["count"], "sample": data["sample"]}
                for fp, data in sorted_patterns[:top_n]
            ]

    def get_rate_stats(self, window_seconds: int = 60) -> Dict[str, Any]: 1107172.m.kuaisou.com 
        with self._lock: 1107173.m.kuaisou.com 
            now = time.time(1107174.m.kuaisou.com)
            recent = [e for e in self._entries if e.timestamp >= now - window_seconds]
            if not recent: 1107175.m.kuaisou.com 
                return {"rate_per_second": 0, "rate_per_minute": 0, "count": 0}

            count = len(recent)
            duration = now - recent[0].timestamp if len(recent) > 1 else window_seconds
            if duration <= 0: 1107176.m.kuaisou.com 
                duration = window_seconds

            return {
                "rate_per_second": round(count / duration, 2),
                "rate_per_minute": round(count / duration * 60, 2),
                "count": count,
                "window_seconds": window_seconds,
            }

    def persist(self) -> None: 1107177.m.kuaisou.com 
        with self._lock: 1107178.m.kuaisou.com 
            filepath = os.path.join(
                self.data_dir,
                f"logs_{int(time.time())}.json"
            )
            with open(filepath, "w") as f: 1107179.m.kuaisou.com 
                json.dump(
                    [e.to_dict(1107180.m.kuaisou.com) for e in self._entries[-1000:]],
                    f,
                    ensure_ascii=False,
                    indent=2,
                )

    def size(self) -> int: 1107181.m.kuaisou.com 
        with self._lock: 1107182.m.kuaisou.com 
            return len(self._entries)

# log_system/notifier.py
import json
import logging
import smtplib
import threading
import time
from email.mime.text import MIMEText
from typing import Any, Callable, Dict, List, Optional
from dataclasses import dataclass, field
from log_system.rules import AlertEvent, AlertSeverity

logger = logging.getLogger("log_system")


@dataclass
class NotificationConfig: 1107183.m.kuaisou.com 
    config_id: str = ""
    name: str = ""
    channels: List[str] = field(default_factory=list)
    severity_filter: List[str] = field(default_factory=list)
    email_recipients: List[str] = field(default_factory=list)
    webhook_urls: List[str] = field(default_factory=list)
    slack_channel: str = ""
    slack_token: str = ""
    smtp_host: str = ""
    smtp_port: int = 587
    smtp_user: str = ""
    smtp_password: str = ""
    from_email: str = ""
    enabled: bool = True


class Notifier: 1107184.m.kuaisou.com 
    def __init__(self): 1108106.m.kuaisou.com 
        self._configs: Dict[str, NotificationConfig] = {}
        self._custom_handlers: List[Callable] = []
        self._alert_history: List[Dict[str, Any]] = []
        self._lock = threading.RLock(1108107.m.kuaisou.com)

    def add_config(self, config: NotificationConfig) -> str: 1108108.m.kuaisou.com 
        if not config.config_id:
            config.config_id = f"notif-{len(self._configs) + 1}"
        self._configs[config.config_id] = config
        return config.config_id

    def on_alert(self, handler: Callable) -> None: 1108109.m.kuaisou.com 
        self._custom_handlers.append(handler)

    def send(self, alert: AlertEvent) -> List[str]: 1108110.m.kuaisou.com 
        results = []
        severity_name = alert.severity.value

        for config in self._configs.values(): 1108111.m.kuaisou.com 
            if not config.enabled: 1108126.m.kuaisou.com 
                continue
            if config.severity_filter and severity_name not in config.severity_filter: 1108113.m.kuaisou.com 
                continue

            for channel in config.channels: 1108112.m.kuaisou.com
                try: 1108114.m.kuaisou.com 
                    if channel == "console": 1108115.m.kuaisou.com 
                        self._send_console(alert, config)
                        results.append(f"console:{config.name}")
                    elif channel == "email": 1108116.m.kuaisou.com 
                        self._send_email(alert, config)
                        results.append(f"email:{config.name}")
                    elif channel == "webhook": 1108117.m.kuaisou.com 
                        self._send_webhook(alert, config)
                        results.append(f"webhook:{config.name}")
                    elif channel == "slack": 1108118.m.kuaisou.com 
                        self._send_slack(alert, config)
                        results.append(f"slack:{config.name}")
                except Exception as e: 1108119.m.kuaisou.com 
                    logger.error(f"Notification failed ({channel}): {e}")

        for handler in self._custom_handlers: 1108120.m.kuaisou.com 
            try: 1108121.m.kuaisou.com 
                handler(alert)
                results.append("custom_handler")
            except Exception as e: 1108122.m.kuaisou.com 
                logger.error(f"Custom handler failed: {e}")

        with self._lock: 1108124.m.kuaisou.com 
            self._alert_history.append({
                "alert_id": alert.alert_id,
                "rule_name": alert.rule_name,
                "severity": severity