"""设备数据模型""" from sqlalchemy import Column, BigInteger, String, Integer, Float, Text, TIMESTAMP, ForeignKey, JSON from sqlalchemy.orm import relationship from sqlalchemy.sql import func from app.core.database import Base class OLTDevice(Base): __tablename__ = "olt_devices" id = Column(BigInteger, primary_key=True, index=True) ip_address = Column(String(45), nullable=False) username = Column(String(100), nullable=False) password = Column(Text, nullable=False) slot_command = Column(String(50), nullable=False) region = Column(String(100), index=True) location = Column(String(200)) description = Column(Text) created_at = Column(TIMESTAMP, server_default=func.now()) updated_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now()) class ONUDevice(Base): __tablename__ = "onu_devices" id = Column(BigInteger, primary_key=True, index=True) mac_address = Column(String(17), nullable=False, index=True) olt_id = Column(BigInteger, ForeignKey("olt_devices.id")) slot_number = Column(Integer) port_number = Column(Integer) port_id = Column(String(20)) # 完整端口标识,如 "1/0/2:4" # OLT 返回的额外信息 loid = Column(String(50)) # LOID model = Column(String(100)) # 设备型号 distance_m = Column(Integer) # 距离(米) region = Column(String(100), index=True) school_name = Column(String(200), index=True) building = Column(String(100)) place_type = Column(String(50)) # 场所类型 room_number = Column(String(50)) notes = Column(Text) # 备注 tags = Column(Text) # 标签(逗号分隔),如 "重点设备,考试用" latitude = Column(Float, nullable=True) # 纬度 longitude = Column(Float, nullable=True) # 经度 created_at = Column(TIMESTAMP, server_default=func.now()) updated_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now()) # 关联状态历史 status_history = relationship("DeviceStatusHistory", back_populates="device", lazy="selectin") class DeviceStatusHistory(Base): __tablename__ = "device_status_history" id = Column(BigInteger, primary_key=True, index=True) onu_device_id = Column(BigInteger, ForeignKey("onu_devices.id")) status = Column(String(20), nullable=False) # online, offline distance_m = Column(Integer) # 距离(米) checked_at = Column(TIMESTAMP, nullable=False) response_data = Column(Text) created_at = Column(TIMESTAMP, server_default=func.now()) device = relationship("ONUDevice", back_populates="status_history") class DuplicateMac(Base): """重复 MAC 地址记录(同一 MAC 出现在多个端口)""" __tablename__ = "duplicate_macs" id = Column(BigInteger, primary_key=True, index=True) olt_id = Column(BigInteger, ForeignKey("olt_devices.id"), nullable=False) mac_address = Column(String(17), nullable=False, index=True) # 所有出现的端口列表,JSON 格式: [{"port_id": "1/0/1:1", "status": "online"}, ...] ports = Column(JSON, nullable=False) first_seen_at = Column(TIMESTAMP, server_default=func.now()) last_seen_at = Column(TIMESTAMP, server_default=func.now(), onupdate=func.now()) class DeviceDailySnapshot(Base): """设备每日状态快照(用于趋势图,避免全量扫描历史表)""" __tablename__ = "device_daily_snapshots" id = Column(BigInteger, primary_key=True, index=True) snapshot_date = Column(String(10), nullable=False, index=True) # YYYY-MM-DD total = Column(Integer, nullable=False, default=0) online = Column(Integer, nullable=False, default=0) offline = Column(Integer, nullable=False, default=0) created_at = Column(TIMESTAMP, server_default=func.now()) class NewDevice(Base): """新发现设备(OLT 扫描到但尚未补全信息的设备)""" __tablename__ = "new_devices" id = Column(BigInteger, primary_key=True, index=True) onu_device_id = Column(BigInteger, ForeignKey("onu_devices.id"), nullable=False, unique=True) olt_id = Column(BigInteger, ForeignKey("olt_devices.id"), nullable=False) discovered_at = Column(TIMESTAMP, server_default=func.now()) class OpticalPowerHistory(Base): """光功率历史记录""" __tablename__ = "optical_power_history" id = Column(BigInteger, primary_key=True, index=True) onu_device_id = Column(BigInteger, ForeignKey("onu_devices.id"), nullable=False, index=True) power_in = Column(String(20)) # 接收光功率 dBm power_out = Column(String(20)) # 发送光功率 dBm recorded_at = Column(TIMESTAMP, nullable=False, server_default=func.now()) class DeviceReplacement(Base): """设备更换记录""" __tablename__ = "device_replacements" id = Column(BigInteger, primary_key=True, index=True) onu_device_id = Column(BigInteger, ForeignKey("onu_devices.id"), nullable=False, index=True) old_mac = Column(String(17), nullable=False) new_mac = Column(String(17), nullable=False) reason = Column(Text) operator_id = Column(String(100)) # 操作人 user_id operator_name = Column(String(100)) replaced_at = Column(TIMESTAMP, nullable=False, server_default=func.now()) created_at = Column(TIMESTAMP, server_default=func.now())