AI 驱动安全运营：智能 SOC、自动化响应与漏洞分析 Agent

import json
from dataclasses import dataclass
from enum import Enum
from openai import OpenAI


class AlertSeverity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"


class Verdict(Enum):
    TRUE_POSITIVE = "true_positive"
    FALSE_POSITIVE = "false_positive"
    NEEDS_INVESTIGATION = "needs_investigation"


@dataclass
class AlertAnalysis:
    verdict: Verdict
    severity: AlertSeverity
    confidence: float
    summary: str
    recommended_actions: list[str]
    false_positive_reason: str | None


ALERT_ANALYSIS_SYSTEM_PROMPT = """你是一位资深安全运营分析师，专注于告警分类与优先级评估。

你的任务是分析 SIEM 告警，判断其真实性并评估风险等级。分析时需要综合考虑：
1. 告警规则的触发条件和误报历史
2. 涉及资产的业务重要性
3. 告警上下文（时间、频率、关联事件）
4. 攻击链的可能性

输出严格使用以下 JSON 格式：
{
    "verdict": "true_positive|false_positive|needs_investigation",
    "severity": "critical|high|medium|low|info",
    "confidence": 0.0-1.0,
    "summary": "一句话摘要",
    "recommended_actions": ["建议操作1", "建议操作2"],
    "false_positive_reason": "误报原因（仅当 verdict=false_positive 时填写）"
}

重要原则：
- 宁可误报升级，不可漏报降级
- 涉及核心资产的告警应默认提高优先级
- 对于无法确定的情况，选择 needs_investigation"""


class AlertAnalyzer:
    def __init__(self, client: OpenAI, model: str = "gpt-4o"):
        self.client = client
        self.model = model
        self.alert_history: list[dict] = []
        self.fp_patterns: list[str] = []

    def enrich_context(self, alert: dict, asset_info: dict) -> str:
        history = self._get_related_history(alert, asset_id=alert.get("source_ip"))
        fp_context = self._match_fp_patterns(alert)

        context = f"""
## 告警信息
- 规则名称: {alert['rule_name']}
- 规则ID: {alert['rule_id']}
- 触发时间: {alert['timestamp']}
- 源IP: {alert['source_ip']}
- 目标IP: {alert['dest_ip']}
- 告警描述: {alert['description']}
- 原始日志摘要: {alert.get('raw_log_snippet', 'N/A')}

## 资产信息
- 资产名称: {asset_info.get('name', 'unknown')}
- 业务重要性: {asset_info.get('criticality', 'unknown')}
- 资产类型: {asset_info.get('type', 'unknown')}
- 所属部门: {asset_info.get('department', 'unknown')}

## 关联历史
- 该告警规则最近24h触发次数: {alert.get('rule_trigger_count_24h', 'N/A')}
- 该源IP历史误报次数: {history.get('fp_count', 0)}
- 该源IP历史确认攻击次数: {history.get('tp_count', 0)}

## 已知误报模式
{fp_context if fp_context else '无匹配的已知误报模式'}
"""
        return context

    def analyze(self, alert: dict, asset_info: dict) -> AlertAnalysis:
        context = self.enrich_context(alert, asset_info)
        user_prompt = f"请分析以下告警:\n{context}"

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": ALERT_ANALYSIS_SYSTEM_PROMPT},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0.1,
            response_format={"type": "json_object"},
        )

        result = json.loads(response.choices[0].message.content)

        return AlertAnalysis(
            verdict=Verdict(result["verdict"]),
            severity=AlertSeverity(result["severity"]),
            confidence=result["confidence"],
            summary=result["summary"],
            recommended_actions=result["recommended_actions"],
            false_positive_reason=result.get("false_positive_reason"),
        )

    def batch_analyze(self, alerts: list[dict], asset_db: dict) -> list[AlertAnalysis]:
        results = []
        for alert in alerts:
            asset_info = asset_db.get(alert["dest_ip"], {})
            analysis = self.analyze(alert, asset_info)
            self.alert_history.append({
                "alert": alert,
                "analysis": analysis,
            })
            results.append(analysis)
        return results

    def _get_related_history(self, alert: dict, asset_id: str) -> dict:
        related = [
            h for h in self.alert_history
            if h["alert"].get("source_ip") == asset_id
            or h["alert"].get("rule_id") == alert.get("rule_id")
        ]
        return {
            "fp_count": sum(
                1 for h in related
                if h["analysis"].verdict == Verdict.FALSE_POSITIVE
            ),
            "tp_count": sum(
                1 for h in related
                if h["analysis"].verdict == Verdict.TRUE_POSITIVE
            ),
        }

    def _match_fp_patterns(self, alert: dict) -> str:
        matched = [
            p for p in self.fp_patterns
            if p.lower() in alert.get("description", "").lower()
        ]
        return "\n".join(f"- {p}" for p in matched) if matched else ""

from dataclasses import dataclass
from datetime import datetime, timedelta


@dataclass
class InvestigationStep:
    name: str
    data_source: str
    query_template: str
    purpose: str


HORIZONTAL_MOVEMENT_PLAYBOOK = [
    InvestigationStep(
        name="初始告警分析",
        data_source="SIEM",
        query_template="""
        SELECT timestamp, src_ip, dest_ip, dest_port, rule_name, 
               rule_severity, description
        FROM siem_alerts 
        WHERE rule_id = '{rule_id}' 
          AND src_ip = '{src_ip}'
          AND dest_ip = '{dest_ip}'
          AND timestamp BETWEEN '{start_time}' AND '{end_time}'
        ORDER BY timestamp
        """,
        purpose="获取原始告警详情和上下文",
    ),
    InvestigationStep(
        name="端点进程行为",
        data_source="EDR",
        query_template="""
        SELECT timestamp, hostname, process_name, process_path,
               parent_process, command_line, user, action
        FROM edr_events 
        WHERE hostname = '{hostname}' 
          AND user = '{user}'
          AND timestamp BETWEEN '{start_time}' AND '{end_time}'
          AND action IN ('process_create', 'remote_thread', 'named_pipe')
        ORDER BY timestamp
        """,
        purpose="检查端点上是否有可疑进程创建、远程线程注入或命名管道通信",
    ),
    InvestigationStep(
        name="网络连接分析",
        data_source="NDR",
        query_template="""
        SELECT timestamp, src_ip, dst_ip, dst_port, protocol,
               bytes_sent, bytes_received, connection_duration,
               ja3_hash, tls_version
        FROM ndr_flows 
        WHERE (src_ip = '{src_ip}' OR dst_ip = '{src_ip}')
          AND dst_port IN (445, 5985, 5986, 3389, 22)
          AND timestamp BETWEEN '{start_time}' AND '{end_time}'
        ORDER BY timestamp
        """,
        purpose="检查SMB、WinRM、RDP、SSH等横向移动常用协议的连接",
    ),
    InvestigationStep(
        name="身份认证事件",
        data_source="IAM",
        query_template="""
        SELECT timestamp, user, source_ip, target_system,
               auth_method, auth_result, mfa_used
        FROM auth_events 
        WHERE user = '{user}'
          AND timestamp BETWEEN '{start_time}' AND '{end_time}'
          AND auth_result = 'failure'
        ORDER BY timestamp
        """,
        purpose="检查是否存在暴力破解或异常认证尝试",
    ),
    InvestigationStep(
        name="文件传输痕迹",
        data_source="EDR",
        query_template="""
        SELECT timestamp, hostname, file_name, file_path, file_hash,
               file_size, action, user
        FROM edr_file_events
        WHERE hostname = '{hostname}'
          AND timestamp BETWEEN '{start_time}' AND '{end_time}')
          AND (file_path LIKE '%\\\\%\\admin$%'
               OR file_path LIKE '%\\\\%\\C$%'
               OR file_path LIKE '%Temp%\\%')
        ORDER BY timestamp
        """,
        purpose="检查是否有通过管理共享或临时目录传输的文件",
    ),
]


class InvestigationAgent:
    def __init__(self, llm_client, tool_registry):
        self.llm = llm_client
        self.tools = tool_registry
        self.playbooks = {
            "lateral_movement": HORIZONTAL_MOVEMENT_PLAYBOOK,
        }

    def execute_investigation(
        self, alert: dict, playbook_name: str
    ) -> dict:
        playbook = self.playbooks[playbook_name]
        investigation_results = []
        timeline_events = []

        for step in playbook:
            query = self._render_query(step, alert)
            raw_results = self.tools.execute_query(
                step.data_source, query
            )
            investigation_results.append({
                "step": step.name,
                "source": step.data_source,
                "purpose": step.purpose,
                "results": raw_results,
            })
            timeline_events.extend(
                self._parse_timeline_events(raw_results, step.data_source)
            )

        timeline_events.sort(key=lambda e: e["timestamp"])
        unique_events = self._deduplicate_events(timeline_events)

        analysis_prompt = self._build_analysis_prompt(
            alert, investigation_results, unique_events
        )

        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "你是高级安全事件调查分析师。基于收集到的多源"
                        "证据，构建完整的攻击时间线，分析攻击者的TTPs（战术、"
                        "技术、过程），判断事件严重程度，并给出根因分析和"
                        "遏制建议。使用 MITRE ATT&CK 框架标注攻击技术。"
                    ),
                },
                {"role": "user", "content": analysis_prompt},
            ],
            temperature=0.1,
        )

        return {
            "alert": alert,
            "playbook": playbook_name,
            "evidence": investigation_results,
            "timeline": unique_events,
            "analysis": response.choices[0].message.content,
        }

    def _render_query(self, step: InvestigationStep, alert: dict) -> str:
        context = {
            "rule_id": alert["rule_id"],
            "src_ip": alert["source_ip"],
            "dest_ip": alert["dest_ip"],
            "hostname": alert.get("hostname", ""),
            "user": alert.get("user", ""),
            "start_time": (
                datetime.fromisoformat(alert["timestamp"])
                - timedelta(hours=1)
            ).isoformat(),
            "end_time": (
                datetime.fromisoformat(alert["timestamp"])
                + timedelta(hours=1)
            ).isoformat(),
        }
        return step.query_template.format(**context)

    def _parse_timeline_events(
        self, results: list[dict], source: str
    ) -> list[dict]:
        return [
            {
                "timestamp": r.get("timestamp", ""),
                "source": source,
                "event_type": r.get("action", r.get("auth_method", "")),
                "detail": r,
            }
            for r in results
        ]

    def _deduplicate_events(self, events: list[dict]) -> list[dict]:
        seen = set()
        unique = []
        for e in events:
            key = (e["timestamp"], e["source"], str(e["detail"]))
            if key not in seen:
                seen.add(key)
                unique.append(e)
        return unique

    def _build_analysis_prompt(
        self, alert, results, timeline
    ) -> str:
        timeline_text = "\n".join(
            f"- [{e['timestamp']}] ({e['source']}) {e['event_type']}"
            for e in timeline
        )
        evidence_text = "\n\n".join(
            f"### {r['step']} ({r['source']})\n"
            f"目的: {r['purpose']}\n"
            f"结果数量: {len(r['results'])}\n"
            f"关键数据: {json.dumps(r['results'][:5], ensure_ascii=False, default=str)}"
            for r in results
        )
        return f"""
## 原始告警
{json.dumps(alert, ensure_ascii=False, default=str)}

## 证据汇总
{evidence_text}

## 时间线
{timeline_text}

请基于以上多源数据进行综合分析，输出：
1. 攻击时间线（关键节点标注）
2. MITRE ATT&CK 技术映射
3. 根因分析
4. 影响范围评估
5. 遏制与修复建议
"""

HUNTING_QUERY_GENERATION_PROMPT = """你是一位威胁狩猎专家，精通各大 SIEM 平台的查询语言。

根据给定的狩猎假设，生成针对目标 SIEM 平台的查询语句。

支持的查询语言：
- SPL（Splunk）
- KQL（Microsoft Sentinel / Defender）
- EQL（Elastic Security）

要求：
1. 查询必须高效，优先使用索引字段
2. 查询时间范围默认为最近 7 天
3. 输出应包含查询语句和对查询逻辑的简要说明
4. 如果假设需要多个查询才能完整覆盖，列出所有查询

输出格式（JSON）：
{
    "queries": [
        {
            "platform": "splunk",
            "query": "...",
            "description": "查询逻辑说明"
        }
    ],
    "coverage_analysis": "该查询覆盖了假设的哪些方面，是否有遗漏",
    "tuning_suggestions": "降低误报率的调优建议"
}"""


class ThreatHuntingEngine:
    def __init__(self, llm_client, siem_connector):
        self.llm = llm_client
        self.siem = siem_connector
        self.hunting_history = []

    def generate_hypothesis(
        self, threat_intel: list[dict], org_assets: dict
    ) -> list[dict]:
        intel_summary = "\n".join(
            f"- {i.get('title', '')}: {i.get('description', '')}"
            for i in threat_intel
        )
        asset_summary = "\n".join(
            f"- {name}: {info.get('type', '')} "
            f"(业务重要性: {info.get('criticality', '')})"
            for name, info in org_assets.items()
        )

        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "你是高级威胁狩猎分析师。基于最新威胁情报和组织"
                        "资产画像，生成针对性的狩猎假设。每个假设应可验证、"
                        "可操作，并关联具体的 ATT&CK 技术。"
                    ),
                },
                {
                    "role": "user",
                    "content": (
                        f"## 最新威胁情报\n{intel_summary}\n\n"
                        f"## 组织资产画像\n{asset_summary}\n\n"
                        "请生成 3-5 个最相关的狩猎假设。"
                    ),
                },
            ],
            temperature=0.3,
        )
        return json.loads(response.choices[0].message.content)

    def generate_detection_queries(
        self, hypothesis: dict, target_platform: str = "splunk"
    ) -> dict:
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": HUNTING_QUERY_GENERATION_PROMPT,
                },
                {
                    "role": "user",
                    "content": (
                        f"## 狩猎假设\n"
                        f"{json.dumps(hypothesis, ensure_ascii=False)}\n\n"
                        f"目标平台: {target_platform}"
                    ),
                },
            ],
            temperature=0.1,
            response_format={"type": "json_object"},
        )
        return json.loads(response.choices[0].message.content)

    def execute_and_analyze(
        self, queries: dict, hypothesis: dict
    ) -> dict:
        all_findings = []
        for q in queries.get("queries", []):
            results = self.siem.execute_query(q["query"])
            if results:
                analysis = self._analyze_findings(results, hypothesis)
                all_findings.append({
                    "query": q,
                    "result_count": len(results),
                    "analysis": analysis,
                })

        return {
            "hypothesis": hypothesis,
            "findings": all_findings,
            "overall_assessment": self._generate_assessment(all_findings),
        }

    def _analyze_findings(
        self, results: list[dict], hypothesis: dict
    ) -> str:
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "分析狩猎查询结果，判断是否与狩猎假设相关，"
                        "是否发现了真实的威胁活动。区分真正的恶意行为"
                        "和合法的管理操作。"
                    ),
                },
                {
                    "role": "user",
                    "content": (
                        f"假设: {hypothesis.get('description', '')}\n\n"
                        f"查询结果（前10条）:\n"
                        f"{json.dumps(results[:10], ensure_ascii=False, default=str)}"
                    ),
                },
            ],
            temperature=0.1,
        )
        return response.choices[0].message.content

    def _generate_assessment(self, findings: list[dict]) -> str:
        summary = "\n".join(
            f"- {f['query']['description']}: "
            f"命中 {f['result_count']} 条"
            for f in findings
        )
        return f"狩猎执行摘要:\n{summary}"

import asyncio
from datetime import datetime, timedelta
from enum import Enum


class ResponseAction(Enum):
    ISOLATE_HOST = "isolate_host"
    BLOCK_IP = "block_ip"
    DISABLE_USER = "disable_user"
    REVOKE_SESSION = "revoke_session"
    ADD_FIREWALL_RULE = "add_firewall_rule"
    QUARANTINE_FILE = "quarantine_file"
    ESCALATE_TO_ANALYST = "escalate_to_analyst"


class RiskLevel(Enum):
    AUTO_EXECUTE = "auto_execute"
    REQUIRES_APPROVAL = "requires_approval"
    MANUAL_ONLY = "manual_only"


RISK_MATRIX = {
    ResponseAction.BLOCK_IP: RiskLevel.REQUIRES_APPROVAL,
    ResponseAction.REVOKE_SESSION: RiskLevel.AUTO_EXECUTE,
    ResponseAction.DISABLE_USER: RiskLevel.REQUIRES_APPROVAL,
    ResponseAction.ISOLATE_HOST: RiskLevel.REQUIRES_APPROVAL,
    ResponseAction.QUARANTINE_FILE: RiskLevel.AUTO_EXECUTE,
    ResponseAction.ADD_FIREWALL_RULE: RiskLevel.MANUAL_ONLY,
    ResponseAction.ESCALATE_TO_ANALYST: RiskLevel.AUTO_EXECUTE,
}

RESPONSE_PLAN_PROMPT = """你是 SOC 自动化响应编排系统。根据事件分析结果，
生成结构化的应急响应计划。

要求：
1. 按优先级排序响应动作
2. 评估每个动作的风险等级
3. 考虑动作的副作用和前置条件
4. 确保动作顺序的逻辑正确性（如：先隔离再取证）

输出 JSON 格式：
{
    "response_plan": [
        {
            "action": "动作类型",
            "parameters": {"目标": "参数"},
            "risk_level": "auto_execute|requires_approval|manual_only",
            "rationale": "执行理由",
            "prerequisites": ["前置条件"],
            "rollback": "回滚方案"
        }
    ],
    "estimated_containment_time": "预计遏制时间",
    "impact_assessment": "对业务的预期影响"
}"""


class AutomatedResponder:
    def __init__(self, llm_client, orchestration_engine, approval_queue):
        self.llm = llm_client
        self.orchestrator = orchestration_engine
        self.approval_queue = approval_queue
        self.executed_actions: list[dict] = []

    async def handle_event(self, event_analysis: dict) -> dict:
        plan = await self._generate_response_plan(event_analysis)
        execution_results = []

        for step in plan.get("response_plan", []):
            action = ResponseAction(step["action"])
            risk_level = RISK_MATRIX.get(action, RiskLevel.MANUAL_ONLY)

            if risk_level == RiskLevel.AUTO_EXECUTE:
                result = await self._execute_action(action, step["parameters"])
                execution_results.append(result)

            elif risk_level == RiskLevel.REQUIRES_APPROVAL:
                approval_id = await self.approval_queue.request_approval(
                    action=action,
                    parameters=step["parameters"],
                    rationale=step["rationale"],
                    rollback=step["rollback"],
                    timeout=timedelta(minutes=15),
                )
                approved = await self.approval_queue.wait_for_decision(
                    approval_id
                )
                if approved:
                    result = await self._execute_action(
                        action, step["parameters"]
                    )
                else:
                    result = {
                        "action": action.value,
                        "status": "skipped",
                        "reason": "approval_denied",
                    }
                execution_results.append(result)

            else:
                execution_results.append({
                    "action": action.value,
                    "status": "pending_manual",
                    "assigned_to": "escalation_team",
                })

        verification = await self._verify_containment(event_analysis)

        return {
            "event_id": event_analysis.get("event_id"),
            "plan": plan,
            "executions": execution_results,
            "verification": verification,
            "timestamp": datetime.now().isoformat(),
        }

    async def _generate_response_plan(
        self, event_analysis: dict
    ) -> dict:
        response = await self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": RESPONSE_PLAN_PROMPT},
                {
                    "role": "user",
                    "content": json.dumps(
                        event_analysis, ensure_ascii=False, default=str
                    ),
                },
            ],
            temperature=0.1,
            response_format={"type": "json_object"},
        )
        return json.loads(response.choices[0].message.content)

    async def _execute_action(
        self, action: ResponseAction, parameters: dict
    ) -> dict:
        result = await self.orchestrator.execute(action, parameters)
        self.executed_actions.append({
            "action": action,
            "parameters": parameters,
            "result": result,
            "timestamp": datetime.now().isoformat(),
        })
        return result

    async def _verify_containment(self, event_analysis: dict) -> dict:
        response = await self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "验证应急响应措施是否已生效。检查：\n"
                        "1. 被隔离的主机是否已与网络断开\n"
                        "2. 被封禁的IP是否仍在产生连接\n"
                        "3. 被吊销的会话是否仍然活跃"
                    ),
                },
                {
                    "role": "user",
                    "content": (
                        f"已执行的操作:\n"
                        f"{json.dumps(self.executed_actions[-5:], ensure_ascii=False, default=str)}"
                    ),
                },
            ],
            temperature=0.1,
        )
        return {
            "verification_result": response.choices[0].message.content,
            "all_contained": True,
        }

from dataclasses import dataclass, field


@dataclass
class VulnAssessment:
    cve_id: str
    cvss_score: float
    epss_percentile: float
    exploit_available: bool
    affected_assets: list[str]
    business_impact: str
    priority: str
    remediation_steps: list[str]
    estimated_remediation_time: str


VULN_ANALYSIS_PROMPT = """你是企业漏洞管理专家。基于收集到的多维度
漏洞信息，进行综合影响评估并生成修复建议。

评估维度：
1. 技术影响：CVSS 评分、漏洞类型、攻击向量
2. 威胁情报：EPSS 利用概率、是否有公开 Exploit
3. 业务影响：受影响资产的业务重要性、暴露面
4. 修复方案：官方补丁可用性、修复复杂度、回滚方案

输出 JSON 格式：
{
    "priority": "P0|P1|P2|P3|P4",
    "business_impact": "业务影响描述",
    "risk_score": 0-100,
    "remediation_steps": ["步骤1", "步骤2"],
    "workaround": "临时缓解方案（如无补丁）",
    "estimated_remediation_time": "预估修复时间",
    "monitoring_suggestions": "修复前的监控建议"
}"""


class VulnAnalysisAgent:
    def __init__(self, llm_client, tool_registry, rag_store):
        self.llm = llm_client
        self.tools = tool_registry
        self.rag = rag_store

    async def analyze(self, cve_id: str) -> VulnAssessment:
        nvd_data = await self.tools.query_nvd(cve_id)
        exploit_data = await self.tools.query_exploit_db(cve_id)
        epss_data = await self.tools.query_epss(cve_id)
        affected_assets = await self.tools.query_cmdb(
            cve_id=cve_id,
            cpe=nvd_data.get("affected_products", []),
        )
        remediation_context = self.rag.search(
            query=f"{cve_id} {nvd_data.get('description', '')}",
            top_k=5,
        )

        analysis_input = {
            "cve_id": cve_id,
            "nvd": nvd_data,
            "exploit": exploit_data,
            "epss": epss_data,
            "affected_assets": affected_assets,
            "remediation_context": remediation_context,
        }

        response = await self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": VULN_ANALYSIS_PROMPT},
                {
                    "role": "user",
                    "content": json.dumps(
                        analysis_input, ensure_ascii=False, default=str
                    ),
                },
            ],
            temperature=0.1,
            response_format={"type": "json_object"},
        )

        result = json.loads(response.choices[0].message.content)

        return VulnAssessment(
            cve_id=cve_id,
            cvss_score=nvd_data.get("cvss_v3_score", 0.0),
            epss_percentile=epss_data.get("percentile", 0.0),
            exploit_available=exploit_data.get("available", False),
            affected_assets=[a["hostname"] for a in affected_assets],
            business_impact=result["business_impact"],
            priority=result["priority"],
            remediation_steps=result["remediation_steps"],
            estimated_remediation_time=result["estimated_remediation_time"],
        )

    async def batch_analyze(
        self, cve_list: list[str]
    ) -> list[VulnAssessment]:
        assessments = []
        for cve_id in cve_list:
            assessment = await self.analyze(cve_id)
            assessments.append(assessment)
        assessments.sort(
            key=lambda a: (
                {"P0": 0, "P1": 1, "P2": 2, "P3": 3, "P4": 4}[
                    a.priority
                ]
            )
        )
        return assessments

@dataclass
class SOCMetrics:
    mttd_hours: float
    mttr_hours: float
    alert_throughput_per_day: int
    false_positive_rate: float
    false_negative_rate: float
    human_intervention_rate: float
    first_call_resolution_rate: float
    cost_per_incident: float

    def calculate_ai_roi(self, ai_tool_cost: float, period_months: int) -> dict:
        baseline_cost_per_incident = 1200.0
        ai_cost_per_incident = self.cost_per_incident
        incidents_per_month = self.alert_throughput_per_day * 30

        baseline_total = baseline_cost_per_incident * incidents_per_month * period_months
        ai_total = (
            ai_cost_per_incident * incidents_per_month * period_months
            + ai_tool_cost
        )
        savings = baseline_total - ai_total

        return {
            "baseline_total_cost": baseline_total,
            "ai_total_cost": ai_total,
            "net_savings": savings,
            "roi_percentage": (savings / ai_tool_cost) * 100 if ai_tool_cost > 0 else 0,
            "breakeven_months": (
                ai_tool_cost / (baseline_cost_per_incident - ai_cost_per_incident) / incidents_per_month
                if incidents_per_month > 0 and baseline_cost_per_incident > ai_cost_per_incident
                else float("inf")
            ),
        }

    def generate_report(self) -> str:
        return f"""
## 安全运营效果报告

### 效率指标
- 平均检测时间（MTTD）: {self.mttd_hours:.1f} 小时
- 平均响应时间（MTTR）: {self.mttr_hours:.1f} 小时
- 告警处理吞吐量: {self.alert_throughput_per_day} 条/天

### 质量指标
- 误报率: {self.false_positive_rate:.1%}
- 漏报率: {self.false_negative_rate:.1%}
- 人工介入率: {self.human_intervention_rate:.1%}
- 一次正确率: {self.first_call_resolution_rate:.1%}

### 成本指标
- 单事件处理成本: ¥{self.cost_per_incident:,.2f}
"""