wecom_it_smart_desk/backend/app/models/diagnostic.py

# =============================================================================
# 企微IT智能服务台 — 诊断相关模型
# =============================================================================
# 说明：包含3张表，支撑三层诊断闭环：
#   1. diagnostic_templates:  原子化诊断检查项模板库（管理员预置）
#   2. diagnostic_reports:    客户端/API采集的诊断报告存储
#   3. diagnostic_dispatches: 诊断下发记录，追踪 dispatch→execute→analyze→resolve 闭环
#
# 三层诊断架构：
#   Layer 1 — 火绒/联软API静默采集（check_type=api, api_source=huorong/lianruan）
#   Layer 2 — 客户端脚本兜底（check_type=script, script_template=PowerShell/zsh）
#   Layer 3 — AI分析报告 + 修复包推送（fix_template + risk_level分级审批）
# =============================================================================

import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional

from sqlalchemy import Boolean, DateTime, Index, Integer, JSON, String, Text
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class DiagnosticTemplate(Base):
    """诊断模板 — 原子化检查项。

    每条记录是一个独立的检查单元（如"ping网关""查DNS配置"），
    管理员在后台预置，AI只负责选择哪些检查项组合，不生成脚本内容。

    Attributes:
        id: 模板唯一标识（UUID）
        category: 问题类别（network/vpn/email/system/printer/security/office）
        name: 检查项名称（如"网关连通性检测"）
        check_type: 检查类型（api=服务端API采集 / script=客户端脚本采集）
        api_source: API来源（huorong/lianruan），仅check_type=api时有效
        api_method: 调用的API方法名（如get_terminal_detail），仅check_type=api时有效
        script_template: PowerShell/zsh脚本模板（参数化），仅check_type=script时有效
        fix_template: 对应的修复脚本模板（可选，部分检查项有配套修复）
        fix_risk_level: 修复风险等级（low/medium/high），决定审批流程
        target_condition: 触发此检查项的条件（如"dns_resolution=fail"）
        description: 检查项描述
        is_active: 是否启用
    """

    __tablename__ = "diagnostic_templates"

    id: Mapped[str] = mapped_column(
        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
    )
    category: Mapped[str] = mapped_column(
        String(50), nullable=False, comment="问题类别"
    )
    name: Mapped[str] = mapped_column(
        String(200), nullable=False, comment="检查项名称"
    )
    check_type: Mapped[str] = mapped_column(
        String(20), nullable=False, default="api",
        comment="检查类型: api/script"
    )
    api_source: Mapped[Optional[str]] = mapped_column(
        String(50), nullable=True, comment="API来源: huorong/lianruan"
    )
    api_method: Mapped[Optional[str]] = mapped_column(
        String(100), nullable=True, comment="调用的API方法名"
    )
    script_template: Mapped[Optional[str]] = mapped_column(
        Text, nullable=True, comment="脚本模板(PowerShell/zsh)"
    )
    fix_template: Mapped[Optional[str]] = mapped_column(
        Text, nullable=True, comment="修复脚本模板"
    )
    fix_risk_level: Mapped[str] = mapped_column(
        String(20), nullable=False, default="medium",
        comment="修复风险等级: low/medium/high"
    )
    target_condition: Mapped[Optional[str]] = mapped_column(
        String(200), nullable=True, comment="触发条件"
    )
    description: Mapped[Optional[str]] = mapped_column(
        Text, nullable=True, comment="检查项描述"
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, comment="是否启用"
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=datetime.now,
        comment="创建时间"
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=datetime.now,
        onupdate=datetime.now, comment="更新时间"
    )

    __table_args__ = (
        Index("idx_diag_tpl_category", "category"),
        Index("idx_diag_tpl_type", "check_type"),
        Index("idx_diag_tpl_active", "is_active"),
    )

    def __repr__(self) -> str:
        return f"<DiagnosticTemplate(id={self.id}, name={self.name}, type={self.check_type})>"


class DiagnosticDispatch(Base):
    """诊断下发记录 — 追踪每次诊断的完整生命周期。

    状态流转：dispatched → executed → analyzed → resolved

    Attributes:
        id: 下发记录ID
        conversation_id: 关联的会话ID
        employee_id: 员工ID
        template_ids: 下发的诊断模板ID列表（JSON数组）
        script_content: 实际生成的脚本内容（参数化后的最终版本）
        script_hash: 脚本SHA256哈希（审计追溯）
        upload_token: 一次性上传token（绑定session+employee+TTL）
        status: 状态（dispatched/executed/analyzed/resolved）
        report_id: 关联的诊断报告ID（报告上传后填入）
        fix_dispatched: 是否已下发修复包
        created_at: 下发时间
        completed_at: 完成（resolved）时间
    """

    __tablename__ = "diagnostic_dispatches"

    id: Mapped[str] = mapped_column(
        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
    )
    conversation_id: Mapped[str] = mapped_column(
        String(36), nullable=False, comment="关联会话ID"
    )
    employee_id: Mapped[str] = mapped_column(
        String(64), nullable=False, comment="员工ID"
    )
    template_ids: Mapped[list] = mapped_column(
        JSON, nullable=False, default=list, comment="诊断模板ID列表"
    )
    script_content: Mapped[Optional[str]] = mapped_column(
        Text, nullable=True, comment="生成的脚本内容"
    )
    script_hash: Mapped[Optional[str]] = mapped_column(
        String(64), nullable=True, comment="脚本SHA256哈希"
    )
    upload_token: Mapped[Optional[str]] = mapped_column(
        String(128), nullable=True, comment="一次性上传token"
    )
    status: Mapped[str] = mapped_column(
        String(20), nullable=False, default="dispatched",
        comment="状态: dispatched/executed/analyzed/resolved"
    )
    report_id: Mapped[Optional[str]] = mapped_column(
        String(36), nullable=True, comment="关联诊断报告ID"
    )
    fix_dispatched: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=False, comment="是否已下发修复包"
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=datetime.now,
        comment="下发时间"
    )
    completed_at: Mapped[Optional[datetime]] = mapped_column(
        DateTime(timezone=True), nullable=True, comment="完成时间"
    )

    __table_args__ = (
        Index("idx_diag_dispatch_conv", "conversation_id"),
        Index("idx_diag_dispatch_employee", "employee_id"),
        Index("idx_diag_dispatch_status", "status"),
    )

    def __repr__(self) -> str:
        return f"<DiagnosticDispatch(id={self.id}, conv={self.conversation_id}, status={self.status})>"


class DiagnosticReport(Base):
    """诊断报告 — 存储采集到的检查结果和AI分析结论。

    Attributes:
        id: 报告ID
        dispatch_id: 关联的下发记录ID
        conversation_id: 关联的会话ID
        employee_id: 员工ID
        template_ids: 涉及的诊断模板ID列表
        report_data: 检查结果JSON（[{name, status, detail, raw_output}]）
        ai_analysis: AI分析结论JSON（{root_cause, confidence, suggested_actions}）
        status: 报告状态（pending/analyzed/resolved）
        created_at: 报告上传时间
    """

    __tablename__ = "diagnostic_reports"

    id: Mapped[str] = mapped_column(
        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
    )
    dispatch_id: Mapped[Optional[str]] = mapped_column(
        String(36), nullable=True, comment="关联下发记录ID"
    )
    conversation_id: Mapped[str] = mapped_column(
        String(36), nullable=False, comment="关联会话ID"
    )
    employee_id: Mapped[str] = mapped_column(
        String(64), nullable=False, comment="员工ID"
    )
    template_ids: Mapped[list] = mapped_column(
        JSON, nullable=False, default=list, comment="涉及诊断模板ID列表"
    )
    report_data: Mapped[Dict[str, Any]] = mapped_column(
        JSON, nullable=False, default=dict,
        comment="检查结果: [{name, status(pass/fail/warn/pending), detail, raw_output}]"
    )
    ai_analysis: Mapped[Optional[Dict[str, Any]]] = mapped_column(
        JSON, nullable=True,
        comment="AI分析: {root_cause, confidence, severity, suggested_actions}"
    )
    status: Mapped[str] = mapped_column(
        String(20), nullable=False, default="pending",
        comment="报告状态: pending/analyzed/resolved"
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=datetime.now,
        comment="报告上传时间"
    )

    __table_args__ = (
        Index("idx_diag_report_conv", "conversation_id"),
        Index("idx_diag_report_employee", "employee_id"),
        Index("idx_diag_report_status", "status"),
    )

    def __repr__(self) -> str:
        return f"<DiagnosticReport(id={self.id}, conv={self.conversation_id}, status={self.status})>"