PingWatch/backend/app/models/device.py

import enum
from dataclasses import dataclass
from datetime import datetime
from typing import TYPE_CHECKING

from sqlalchemy import Boolean, Column, DateTime, Enum, Float, Integer, String
from sqlalchemy.orm import DeclarativeBase

if TYPE_CHECKING:
    from app.config import Settings


class Base(DeclarativeBase):
    pass


class DeviceTypeEnum(str, enum.Enum):
    server = "server"         # 服务器
    olt = "olt"               # OLT
    switch = "switch"         # 交换机
    firewall = "firewall"     # 防火墙
    other = "other"           # 其他


@dataclass(frozen=True)
class DeviceMonitoringPolicy:
    """Resolved monitoring thresholds for one device.

    Nullable device fields intentionally inherit the runtime defaults so a
    configuration change can apply to devices that do not need exceptions.
    """

    probe_packets_per_round: int
    offline_consecutive_rounds: int
    degraded_window_rounds: int
    degraded_loss_percent: float
    recovery_consecutive_clean_rounds: int

    @staticmethod
    def _resolve_override(
        override: object,
        default: int | float,
        minimum: int | float,
        maximum: int | float,
    ) -> int | float:
        """Use a bounded persisted override or retain the validated default."""
        if (
            isinstance(override, bool)
            or not isinstance(override, (int, float))
            or override < minimum
            or override > maximum
        ):
            return default
        return override

    @classmethod
    def from_device(
        cls,
        device: "Device",
        settings: "Settings",
    ) -> "DeviceMonitoringPolicy":
        """Resolve device overrides without changing the stored device."""
        return cls(
            probe_packets_per_round=(
                cls._resolve_override(
                    device.probe_packets_per_round,
                    settings.probe_packets_per_round,
                    1,
                    10,
                )
            ),
            offline_consecutive_rounds=(
                cls._resolve_override(
                    device.offline_consecutive_rounds,
                    settings.offline_consecutive_rounds,
                    1,
                    10,
                )
            ),
            degraded_window_rounds=(
                cls._resolve_override(
                    device.degraded_window_rounds,
                    settings.degraded_window_rounds,
                    2,
                    60,
                )
            ),
            degraded_loss_percent=(
                cls._resolve_override(
                    device.degraded_loss_percent,
                    settings.degraded_loss_percent,
                    1,
                    100,
                )
            ),
            recovery_consecutive_clean_rounds=(
                cls._resolve_override(
                    device.recovery_consecutive_clean_rounds,
                    settings.recovery_consecutive_clean_rounds,
                    1,
                    20,
                )
            ),
        )


class Device(Base):
    __tablename__ = "devices"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(128), nullable=False, comment="设备名称")
    ip = Column(String(45), nullable=False, index=True, comment="IP 地址")
    device_type = Column(Enum(DeviceTypeEnum), nullable=False, default=DeviceTypeEnum.other, comment="设备类型")
    location = Column(String(256), default="", comment="物理位置/地址")
    project_name = Column(String(256), default="", comment="所属项目")
    tags = Column(String(512), default="", comment="标签，逗号分隔")

    # Ping 设置
    ping_interval = Column(Integer, default=30, comment="ping 间隔（秒）")
    alert_threshold = Column(Integer, default=5, comment="连续失败次数判离线")
    is_enabled = Column(Boolean, default=True, comment="是否启用监控")

    # 设备级监测策略覆盖；NULL 时使用运行环境中的全局默认值。
    probe_packets_per_round = Column(
        Integer,
        nullable=True,
        comment="单轮探测发包数覆盖",
    )
    offline_consecutive_rounds = Column(
        Integer,
        nullable=True,
        comment="连续全丢包离线轮数覆盖",
    )
    degraded_window_rounds = Column(
        Integer,
        nullable=True,
        comment="故障丢包滑动窗口轮数覆盖",
    )
    degraded_loss_percent = Column(
        Float,
        nullable=True,
        comment="故障丢包率阈值覆盖",
    )
    recovery_consecutive_clean_rounds = Column(
        Integer,
        nullable=True,
        comment="连续零丢包恢复轮数覆盖",
    )

    # 运行状态
    current_status = Column(
        String(16),
        default="unknown",
        comment="当前状态: online/degraded/offline/unknown",
    )
    consecutive_failures = Column(Integer, default=0, comment="当前连续失败次数")
    last_ping_time = Column(DateTime, nullable=True, comment="最后一次 ping 时间")
    last_online_time = Column(DateTime, nullable=True, comment="最后一次在线时间")
    last_offline_time = Column(DateTime, nullable=True, comment="最后一次离线时间")

    created_at = Column(DateTime, default=datetime.now, comment="创建时间")
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")

    def __repr__(self):
        return f"<Device(id={self.id}, name={self.name}, ip={self.ip})>"