fault-log-analyzer/src/fault_log_analyzer/models.py

"""数据模型定义。

依据 docs/01-design/database-design.md 中 fault_log / root_cause /
fault_type / fault_filter_rule / event 表结构，提供进程内数据模型。
核心实现零第三方依赖，全部使用标准库 dataclass。
"""

from __future__ import annotations

from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _to_dt(value: Any) -> datetime:
    """将字符串/时间戳转换为带时区的 datetime。

    支持 ISO8601（带 Z 或空格分隔）、Unix 时间戳以及 syslog 风格
    ``Jan  1 10:18:00``（无年份，缺省使用当前年份）。
    """
    if value is None:
        return _now()
    if isinstance(value, datetime):
        if value.tzinfo is None:
            return value.replace(tzinfo=timezone.utc)
        return value
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value, tz=timezone.utc)

    text = str(value).strip()
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"

    dt: Optional[datetime] = None
    try:
        dt = datetime.fromisoformat(text)
    except ValueError:
        try:
            dt = datetime.fromisoformat(text.replace(" ", "T"))
        except ValueError:
            # syslog 风格：Jan  1 10:18:00（无年份，缺省当前年份）
            normalized = " ".join(text.split())
            dt = datetime.strptime(normalized, "%b %d %H:%M:%S")
            dt = dt.replace(year=datetime.now().year)

    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt


@dataclass
class FaultLog:
    """故障日志（归类后），对应 fault_log 表。"""

    fault_log_id: str
    host_id: str
    message: str
    level: str = "ERROR"
    fingerprint: str = ""
    fault_type: Optional[str] = None
    cluster_id: Optional[str] = None
    service: Optional[str] = None
    trace_id: Optional[str] = None
    occurred_at: datetime = field(default_factory=_now)
    count: int = 1

    def to_dict(self) -> Dict[str, Any]:
        data = asdict(self)
        data["occurred_at"] = self.occurred_at.isoformat()
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "FaultLog":
        payload = dict(data)
        payload["occurred_at"] = _to_dt(payload.get("occurred_at"))
        return cls(**{k: v for k, v in payload.items() if k in cls.__dataclass_fields__})


@dataclass
class RootCause:
    """根因结论，对应 root_cause 表。"""

    fault_log_id: str
    cause_type: str
    evidence: List[Dict[str, Any]] = field(default_factory=list)
    confidence: float = 0.0
    analysis_at: datetime = field(default_factory=_now)

    def to_dict(self) -> Dict[str, Any]:
        data = asdict(self)
        data["analysis_at"] = self.analysis_at.isoformat()
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "RootCause":
        payload = dict(data)
        payload["analysis_at"] = _to_dt(payload.get("analysis_at"))
        return cls(**{k: v for k, v in payload.items() if k in cls.__dataclass_fields__})


@dataclass
class FaultType:
    """故障类型，对应 fault_type 表。"""

    fault_type: str
    name: str
    description: str = ""
    pattern: str = ""
    severity: str = "warning"
    enabled: bool = True

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "FaultType":
        return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})


@dataclass
class FaultFilterRule:
    """故障日志过滤规则，对应 fault_filter_rule 表。"""

    name: str
    level: str = "ERROR"
    pattern: str = ""
    exclude_pattern: str = ""
    enabled: bool = True

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "FaultFilterRule":
        return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})


@dataclass
class Event:
    """检测事件（用于根因分析的关联输入），对应 event 表。"""

    event_id: str
    host_id: str
    metric: str
    value: float
    threshold: float
    operator: str = "gt"
    status: str = "firing"
    severity: str = "warning"
    fired_at: datetime = field(default_factory=_now)

    def to_dict(self) -> Dict[str, Any]:
        data = asdict(self)
        data["fired_at"] = self.fired_at.isoformat()
        return data

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Event":
        payload = dict(data)
        payload["fired_at"] = _to_dt(payload.get("fired_at"))
        return cls(**{k: v for k, v in payload.items() if k in cls.__dataclass_fields__})


@dataclass
class ParsedLog:
    """解析后的结构化日志条目（捕获管道内部表示）。"""

    timestamp: datetime = field(default_factory=_now)
    host_id: str = ""
    service: str = ""
    level: str = "INFO"
    message: str = ""
    trace_id: Optional[str] = None
    raw: str = ""

    def to_dict(self) -> Dict[str, Any]:
        data = asdict(self)
        data["timestamp"] = self.timestamp.isoformat()
        return data