zpark-ems/core/backend/app/models/ai_ops.py

"""AI运维智能体数据模型 - 设备健康评分、异常检测、诊断报告、预测性维护、运营洞察"""
from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, ForeignKey, Text, JSON
from sqlalchemy.sql import func
from app.core.database import Base


class DeviceHealthScore(Base):
    """设备健康评分"""
    __tablename__ = "device_health_scores"

    id = Column(Integer, primary_key=True, autoincrement=True)
    device_id = Column(Integer, ForeignKey("devices.id"), nullable=False, index=True)
    timestamp = Column(DateTime(timezone=True), server_default=func.now(), index=True)
    health_score = Column(Float, nullable=False)  # 0-100
    status = Column(String(20), default="healthy")  # healthy, warning, critical
    factors = Column(JSON)  # {power_stability, efficiency, alarm_frequency, uptime, temperature}
    trend = Column(String(20), default="stable")  # improving, stable, degrading
    created_at = Column(DateTime(timezone=True), server_default=func.now())


class AnomalyDetection(Base):
    """异常检测记录"""
    __tablename__ = "anomaly_detections"

    id = Column(Integer, primary_key=True, autoincrement=True)
    device_id = Column(Integer, ForeignKey("devices.id"), nullable=False, index=True)
    detected_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
    anomaly_type = Column(String(50), nullable=False)  # power_drop, efficiency_loss, abnormal_temperature, communication_loss, pattern_deviation
    severity = Column(String(20), default="warning")  # info, warning, critical
    description = Column(Text)
    metric_name = Column(String(50))
    expected_value = Column(Float)
    actual_value = Column(Float)
    deviation_percent = Column(Float)
    status = Column(String(20), default="detected")  # detected, investigating, resolved, false_positive
    resolution_notes = Column(Text)
    created_at = Column(DateTime(timezone=True), server_default=func.now())


class DiagnosticReport(Base):
    """AI诊断报告"""
    __tablename__ = "diagnostic_reports"

    id = Column(Integer, primary_key=True, autoincrement=True)
    device_id = Column(Integer, ForeignKey("devices.id"), nullable=False, index=True)
    generated_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
    report_type = Column(String(20), default="routine")  # routine, triggered, comprehensive
    findings = Column(JSON)  # [{finding, severity, detail}]
    recommendations = Column(JSON)  # [{action, priority, detail}]
    estimated_impact = Column(JSON)  # {energy_loss_kwh, cost_impact_yuan}
    status = Column(String(20), default="generated")  # generated, reviewed, action_taken
    created_at = Column(DateTime(timezone=True), server_default=func.now())


class MaintenancePrediction(Base):
    """预测性维护"""
    __tablename__ = "maintenance_predictions"

    id = Column(Integer, primary_key=True, autoincrement=True)
    device_id = Column(Integer, ForeignKey("devices.id"), nullable=False, index=True)
    predicted_at = Column(DateTime(timezone=True), server_default=func.now())
    component = Column(String(100))
    failure_mode = Column(String(200))
    probability = Column(Float)  # 0-1
    predicted_failure_date = Column(DateTime(timezone=True))
    recommended_action = Column(Text)
    urgency = Column(String(20), default="medium")  # low, medium, high, critical
    estimated_downtime_hours = Column(Float)
    estimated_repair_cost = Column(Float)
    status = Column(String(20), default="predicted")  # predicted, scheduled, completed, false_alarm
    created_at = Column(DateTime(timezone=True), server_default=func.now())


class OpsInsight(Base):
    """运营洞察"""
    __tablename__ = "ops_insights"

    id = Column(Integer, primary_key=True, autoincrement=True)
    insight_type = Column(String(50), nullable=False)  # efficiency_trend, cost_anomaly, performance_comparison, seasonal_pattern
    title = Column(String(200), nullable=False)
    description = Column(Text)
    data = Column(JSON)
    impact_level = Column(String(20), default="medium")  # low, medium, high
    actionable = Column(Boolean, default=False)
    recommended_action = Column(Text)
    generated_at = Column(DateTime(timezone=True), server_default=func.now())
    valid_until = Column(DateTime(timezone=True))
    created_at = Column(DateTime(timezone=True), server_default=func.now())


class AIAnalysisResult(Base):
    """AI智能分析结果"""
    __tablename__ = "ai_analysis_results"

    id = Column(Integer, primary_key=True, autoincrement=True)
    scope = Column(String(20), default="station")  # device, station, all
    device_id = Column(Integer, ForeignKey("devices.id"), nullable=True)
    analysis_type = Column(String(20), default="diagnostic")  # diagnostic, insight, custom
    prompt_used = Column(Text)
    result_text = Column(Text)
    model_used = Column(String(100))
    provider_used = Column(String(20))  # stepfun, zhipu
    tokens_used = Column(Integer)
    duration_ms = Column(Integer)
    created_by = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime(timezone=True), server_default=func.now())