#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称：performance_service.py
模块描述：性能测试服务层，负责 CRUD、执行调度、快照采集与 WebSocket 推送

作者：czj
创建日期：2026-08-12
最后修改：2026-08-12
"""

import asyncio
import json
import logging
import time
from datetime import datetime
from types import SimpleNamespace
from typing import Any, Callable, Dict, List, Optional, Tuple

from sqlalchemy import select, func, desc
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.performance import (
    PerformanceTask,
    PerformanceSnapshot,
    PerformanceExecution,
    PerformanceProject,
    PerformanceBatchExecution,
    PerformanceProjectReport,
)
from app.models.performance_output import PerformanceTaskOutput
from app.models.api_preset import ApiPreset
from app.schemas.performance import (
    PerformanceTaskCreate,
    PerformanceProjectCreate,
    PerformanceProjectUpdate,
)
from app.executors.performance_executor import PerformanceExecutor
from app.websocket.manager import manager
from app.utils.id_generator import generate_id
from app.config import settings

logger = logging.getLogger(__name__)

# 默认的账号配置（与 playwright_executor.LOGIN_CONFIG / login_template_service 一致）
_DEFAULT_ACCOUNTS = {
    "superadmin": {"username": "admin@xty", "password": "Ubains@13579"},
    "admin": {"username": "admin@xty", "password": "Ubains@13579"},
    "user": {"username": "admin@xty", "password": "Ubains@13579"},
    "captcha": "csba",
}


class PerformanceService:
    """
    性能测试服务

    提供性能测试任务的 CRUD 操作、执行调度、实时监控和报告生成。
    执行引擎在后台线程池中运行，避免阻塞 FastAPI 事件循环。
    """

    # 正在运行的任务缓存：{task_id: PerformanceExecutor}
    _running_executors: Dict[str, PerformanceExecutor] = {}
    # 后台任务集合，防止 GC 回收
    _background_tasks: set = set()

    def __init__(self, db: AsyncSession):
        """
        初始化服务

        Args:
            db: 数据库会话
        """
        self.db = db

    # ==================== CRUD ====================

    async def get_task(self, task_id: str) -> Optional[PerformanceTask]:
        """
        获取任务详情

        Args:
            task_id: 任务ID

        Returns:
            PerformanceTask: 任务对象，不存在返回 None
        """
        return await self.db.get(PerformanceTask, task_id)

    async def list_tasks(
        self,
        page: int = 1,
        page_size: int = 20,
        status: Optional[str] = None,
        project_id: Optional[str] = None,
    ) -> Tuple[List[PerformanceTask], int]:
        """
        获取任务列表（分页）

        Args:
            page: 页码
            page_size: 每页数量
            status: 按状态过滤
            project_id: 按项目过滤；空字符串查未归属，None 查全部

        Returns:
            tuple: (任务列表, 总数)
        """
        query = select(PerformanceTask)
        count_query = select(func.count()).select_from(PerformanceTask)

        if status:
            query = query.where(PerformanceTask.status == status)
            count_query = count_query.where(PerformanceTask.status == status)

        if project_id is not None:
            if project_id == "":
                # 查询未归属项目的任务
                query = query.where(PerformanceTask.project_id.is_(None))
                count_query = count_query.where(PerformanceTask.project_id.is_(None))
            else:
                query = query.where(PerformanceTask.project_id == project_id)
                count_query = count_query.where(PerformanceTask.project_id == project_id)

        total = (await self.db.execute(count_query)).scalar() or 0

        query = (
            query.order_by(desc(PerformanceTask.created_at))
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
        result = await self.db.execute(query)
        items = list(result.scalars().all())

        return items, total

    async def create_task(self, data: PerformanceTaskCreate) -> PerformanceTask:
        """
        创建性能测试任务

        如果传入了 preset_id，从接口预设继承接口配置字段
        （method/target_url/headers/body/body_template/capture_rules/
        auth_required/sign_request/account_key），
        任务显式传入的字段覆盖预设值。

        Args:
            data: 任务创建 Schema（含 preset_id 时只继承未显式传入的接口字段）

        Returns:
            PerformanceTask: 创建的任务对象
        """
        # 用户显式传入的字段（排除 schema 默认值，避免默认值覆盖预设）
        explicit = data.model_dump(exclude_unset=True)

        # 从接口预设继承字段
        preset_id = explicit.get("preset_id")
        if preset_id:
            preset = await self.db.get(ApiPreset, preset_id)
            if not preset:
                raise ValueError(f"接口预设不存在: {preset_id}")
            explicit = self._inherit_from_preset(explicit, preset)
            if not explicit.get("target_url"):
                raise ValueError("目标接口 URL 不能为空")

        # 用 schema 默认值补齐其余压测参数
        payload = PerformanceTaskCreate(**explicit).model_dump()

        # 补齐 Optional 压测参数的默认值（前端可能传 null）
        if payload.get("qps_target") is None:
            payload["qps_target"] = 100
        if payload.get("ramp_up") is None:
            payload["ramp_up"] = 10
        if payload.get("step_concurrency") is None:
            payload["step_concurrency"] = []
        if payload.get("step_duration") is None:
            payload["step_duration"] = 60
        if payload.get("description") is None:
            payload["description"] = ""

        assertions = payload.pop("assertions", [])
        task = PerformanceTask(
            id=generate_id("perf"),
            **payload,
        )
        task.assertions = assertions
        self.db.add(task)
        await self.db.commit()
        await self.db.refresh(task)
        logger.info(f"创建性能测试任务: {task.id} - {task.name}")
        return task

    @staticmethod
    def _inherit_from_preset(data: Dict[str, Any], preset: ApiPreset) -> Dict[str, Any]:
        """
        从接口预设继承字段（任务显式传入的值覆盖预设默认值）

        Args:
            data: 任务创建数据（exclude_unset 后的显式字段，未传入字段以默认值补齐）
            preset: 接口预设对象

        Returns:
            dict: 合并后的任务数据
        """
        # 接口配置字段：任务显式传入的优先，否则用预设值
        if not data.get("method"):
            data["method"] = preset.method
        if not data.get("target_url"):
            data["target_url"] = preset.target_url
        if not data.get("headers"):
            data["headers"] = preset.headers or {}
        if data.get("body") is None:
            data["body"] = preset.body
        if not data.get("body_template"):
            data["body_template"] = preset.body_template
        if data.get("capture_rules") is None:
            data["capture_rules"] = preset.capture_rules or []
        if data.get("auth_required") is None:
            data["auth_required"] = preset.auth_required
        if data.get("sign_request") is None:
            data["sign_request"] = preset.sign_request
        if not data.get("account_key"):
            data["account_key"] = preset.account_key
        # 登录凭据：任务显式传入的优先，否则用预设值（预设留空则不覆盖）
        if not data.get("login_username") and preset.login_username:
            data["login_username"] = preset.login_username
        if not data.get("login_password") and preset.login_password:
            data["login_password"] = preset.login_password
        return data

    async def update_task(
        self, task_id: str, data: Dict[str, Any]
    ) -> Optional[PerformanceTask]:
        """
        更新性能测试任务

        Args:
            task_id: 任务ID
            data: 更新数据

        Returns:
            PerformanceTask: 更新后的任务，不存在返回 None
        """
        task = await self.get_task(task_id)
        if not task:
            return None

        # 不允许更新正在运行的任务
        if task.status == "running":
            raise ValueError("正在运行的任务不能修改")

        assertions = data.pop("assertions", None)
        for key, value in data.items():
            if value is not None and hasattr(task, key):
                setattr(task, key, value)
        if assertions is not None:
            task.assertions = assertions

        await self.db.commit()
        await self.db.refresh(task)
        return task

    async def delete_task(self, task_id: str) -> bool:
        """
        删除性能测试任务

        Args:
            task_id: 任务ID

        Returns:
            bool: 是否成功删除
        """
        task = await self.get_task(task_id)
        if not task:
            return False

        if task.status == "running":
            raise ValueError("正在运行的任务不能删除")

        await self.db.delete(task)
        await self.db.commit()
        return True

    # ==================== 执行调度 ====================

    async def run_task(
        self,
        task_id: str,
        config_override: Optional[Dict[str, Any]] = None,
        progress_callback: Optional[Callable] = None,
        triggered_by: str = "manual",
        batch_id: Optional[str] = None,
        execution_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        执行性能测试任务

        在后台线程池中运行，避免阻塞 FastAPI 事件循环。
        执行完成后自动更新任务状态和统计结果。

        Args:
            task_id: 任务ID
            config_override: 可覆盖的配置项（如临时修改并发数）
            progress_callback: 进度回调
            triggered_by: 触发方式（manual/batch/legacy）
            batch_id: 关联批次ID（批量执行时传入）
            execution_id: 复用的执行记录ID（run_task_async 已创建记录时传入，
                          避免重复创建导致僵尸 running 记录）

        Returns:
            dict: 任务数据（含执行结果）
        """
        task = await self.get_task(task_id)
        if not task:
            raise ValueError(f"任务不存在: {task_id}")

        if task.status == "running":
            raise ValueError("任务正在执行中，不能重复启动")

        # 应用配置覆盖
        if config_override:
            for key, value in config_override.items():
                if hasattr(task, key):
                    setattr(task, key, value)

        # 创建/复用执行记录
        if execution_id:
            # 复用 run_task_async 已创建的记录（避免双条：一条 running 卡死 + 一条真实执行）
            execution = await self.db.get(PerformanceExecution, execution_id)
            if not execution:
                raise ValueError(f"执行记录不存在: {execution_id}")
            execution.status = "running"
            execution.start_time = datetime.utcnow()
        else:
            execution_id = generate_id("exec")
            execution = PerformanceExecution(
                id=execution_id,
                task_id=task_id,
                task_name=task.name,
                project_id=task.project_id,
                status="running",
                start_time=datetime.utcnow(),
                triggered_by=triggered_by,
                batch_id=batch_id,
                created_at=datetime.utcnow(),
            )
            self.db.add(execution)
        await self.db.flush()

        # 更新状态为运行中
        task.status = "running"
        task.start_time = datetime.utcnow()
        task.error_message = None
        await self.db.flush()

        # 预先把 ORM 对象转成纯字典，避免跨线程 SQLAlchemy 脱钩问题
        task_dict = task.to_dict()

        # 快照收集缓冲区（跨线程收集，执行完成后批量写入）
        snapshot_buffer: List[Dict[str, Any]] = []
        ws_group = f"perf_{task_id}"

        def snapshot_callback(snapshot: dict) -> None:
            """每秒快照回调：收集快照 + WebSocket 推送"""
            snapshot["task_id"] = task_id
            snapshot["execution_id"] = execution_id
            snapshot["timestamp"] = time.time()
            snapshot_buffer.append(dict(snapshot))

            # WebSocket 推送（在回调线程中执行）
            try:
                ws_msg = {
                    "type": "perf_snapshot",
                    "task_id": task_id,
                    "execution_id": execution_id,
                    "data": snapshot,
                }
                # 使用 asyncio.run_coroutine_threadsafe 在事件循环中广播
                loop = asyncio.get_event_loop()
                if loop.is_running():
                    asyncio.run_coroutine_threadsafe(
                        manager.broadcast(ws_group, ws_msg), loop
                    )
            except Exception as e:
                logger.warning(f"WebSocket 推送快照失败: {e}")

        try:
            # 在线程池中执行
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(
                None,
                self._run_sync,
                task_dict,
                snapshot_callback,
            )

            if result.get("error"):
                task.status = "failed"
                task.error_message = result.get("error_message", "执行异常")
                logger.error(f"性能测试执行失败 [{task_id}]: {result.get('error_message')}")
            else:
                task.status = "completed"
                # 更新统计字段
                self._apply_result(task, result)
                logger.info(f"性能测试执行完成 [{task_id}]: "
                            f"请求={result.get('total_requests', 0)}, "
                            f"TPS={result.get('actual_tps', 0)}")

        except Exception as e:
            task.status = "failed"
            task.error_message = str(e)
            logger.exception(f"性能测试执行异常 [{task_id}]: {e}")

        finally:
            task.end_time = datetime.utcnow()
            if task.start_time:
                task.duration_actual = (task.end_time - task.start_time).total_seconds()
            else:
                task.duration_actual = 0.0
            await self.db.flush()

            # 更新执行记录
            execution.status = task.status
            execution.error_message = task.error_message
            execution.end_time = task.end_time
            execution.duration_actual = task.duration_actual
            if result and not result.get("error"):
                self._copy_result_to_execution(execution, result)
            execution.snapshot_count = len(snapshot_buffer)
            await self.db.flush()

            # 批量保存快照（关联 execution_id）
            await self._save_snapshots(task_id, snapshot_buffer, execution_id)

            # 保存响应捕获输出（供其他任务 REF 引用）
            await self._save_outputs(task_id, result.get("captured") if result else None)

            await self.db.commit()

            # 清理运行缓存
            self._running_executors.pop(task_id, None)

            # 发送完成通知（含 execution_id）
            try:
                asyncio.ensure_future(
                    manager.broadcast(ws_group, {
                        "type": "perf_complete",
                        "task_id": task_id,
                        "execution_id": execution_id,
                        "status": task.status,
                        "data": task.to_dict(),
                    })
                )
            except Exception:
                pass

        return task.to_dict()

    def _run_sync(
        self,
        task_dict: Dict[str, Any],
        snapshot_callback: Callable,
    ) -> Dict[str, Any]:
        """
        同步执行包装器（在后台线程运行）

        创建独立事件循环运行 asyncio 引擎，避免与 FastAPI 事件循环冲突。

        Args:
            task_dict: 任务数据字典
            snapshot_callback: 快照回调

        Returns:
            dict: 执行结果汇总
        """
        # 设置独立事件循环（参考 PlaywrightExecutor / SecurityExecutor 模式）
        asyncio.set_event_loop(None)
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            # 构建 http_client_config
            client_config = self._build_client_config(task_dict)

            # 创建 SimpleNamespace 以支持属性访问（executor 使用 task.mode 等）
            task_obj = SimpleNamespace(**task_dict)

            executor = PerformanceExecutor(client_config, ref_provider=self._build_ref_provider())
            self._running_executors[task_dict.get("id", "")] = executor

            result = loop.run_until_complete(
                executor.execute(task_obj, snapshot_callback=snapshot_callback)
            )
            return result
        except Exception as e:
            logger.exception(f"_run_sync 异常: {e}")
            return {"error": True, "error_message": str(e)}
        finally:
            loop.close()
            # 清理运行缓存
            tid = task_dict.get("id", "")
            self._running_executors.pop(tid, None)

    async def stop_task(self, task_id: str) -> bool:
        """
        停止正在运行的性能测试任务

        Args:
            task_id: 任务ID

        Returns:
            bool: 是否成功停止
        """
        executor = self._running_executors.get(task_id)
        if not executor:
            logger.warning(f"未找到正在运行的任务: {task_id}")
            return False

        executor.stop()
        logger.info(f"已发送停止信号: {task_id}")

        # 更新数据库状态
        task = await self.get_task(task_id)
        if task and task.status == "running":
            task.status = "cancelled"
            task.end_time = datetime.utcnow()
            if task.start_time:
                task.duration_actual = (task.end_time - task.start_time).total_seconds()
            await self.db.commit()

        return True

    # ==================== 快照与报告 ====================

    async def get_snapshots(
        self,
        task_id: str,
        page: int = 1,
        page_size: int = 500,
    ) -> Tuple[List[PerformanceSnapshot], int]:
        """
        获取任务快照数据

        Args:
            task_id: 任务ID
            page: 页码
            page_size: 每页数量

        Returns:
            tuple: (快照列表, 总数)
        """
        query = (
            select(PerformanceSnapshot)
            .where(PerformanceSnapshot.task_id == task_id)
            .order_by(PerformanceSnapshot.elapsed.asc())
        )
        count_query = (
            select(func.count())
            .select_from(PerformanceSnapshot)
            .where(PerformanceSnapshot.task_id == task_id)
        )

        total = (await self.db.execute(count_query)).scalar() or 0
        query = query.offset((page - 1) * page_size).limit(page_size)
        result = await self.db.execute(query)
        snapshots = list(result.scalars().all())

        return snapshots, total

    async def get_report(
        self,
        task_id: Optional[str] = None,
        execution_id: Optional[str] = None,
    ) -> Optional[Dict[str, Any]]:
        """
        获取性能测试报告

        支持两种模式：
        - execution_id 优先：读取指定执行记录
        - task_id 回退：查询最新执行记录，无执行记录时从任务本身构建（兼容存量数据）

        Args:
            task_id: 任务ID（execution_id 为空时使用）
            execution_id: 执行记录ID（优先）

        Returns:
            dict: 报告数据（含摘要、指标、快照），不存在返回 None
        """
        # 模式1：指定执行记录
        if execution_id:
            return await self._build_report_from_execution(execution_id)

        if not task_id:
            return None

        # 模式2：查最新执行记录
        exec_result = await self.db.execute(
            select(PerformanceExecution)
            .where(PerformanceExecution.task_id == task_id)
            .order_by(PerformanceExecution.start_time.desc())
            .limit(1)
        )
        latest_execution = exec_result.scalar_one_or_none()

        if latest_execution:
            return await self._build_report_from_execution(latest_execution.id)

        # 模式3：存量数据 — 从任务本身构建报告
        task = await self.get_task(task_id)
        if not task:
            return None

        # 获取所有快照（按 task_id 查，兼容旧数据）
        result = await self.db.execute(
            select(PerformanceSnapshot)
            .where(PerformanceSnapshot.task_id == task_id)
            .order_by(PerformanceSnapshot.elapsed.asc())
        )
        snapshots = list(result.scalars().all())

        # 汇总状态码分布
        status_2xx = sum(s.status_2xx for s in snapshots)
        status_3xx = sum(s.status_3xx for s in snapshots)
        status_4xx = sum(s.status_4xx for s in snapshots)
        status_5xx = sum(s.status_5xx for s in snapshots)

        return {
            "summary": {
                "task_id": task.id,
                "task_name": task.name,
                "target_url": task.target_url,
                "method": task.method,
                "mode": task.mode,
                "scenario_type": task.scenario_type,
                "concurrency": task.concurrency,
                "duration_setting": task.duration,
                "duration_actual": task.duration_actual,
                "start_time": task.start_time.isoformat() if task.start_time else None,
                "end_time": task.end_time.isoformat() if task.end_time else None,
                "account_key": task.account_key,
            },
            "metrics": {
                "total_requests": task.total_requests,
                "success_count": task.success_count,
                "fail_count": task.fail_count,
                "error_rate": task.error_rate,
                "actual_tps": task.actual_tps,
                "peak_tps": task.peak_tps,
                "min_response_time": task.min_response_time,
                "max_response_time": task.max_response_time,
                "avg_response_time": task.avg_response_time,
                "p50_response_time": task.p50_response_time,
                "p90_response_time": task.p90_response_time,
                "p95_response_time": task.p95_response_time,
                "p99_response_time": task.p99_response_time,
                "std_dev": task.std_dev,
                "latency_avg": task.latency_avg,
                "latency_min": task.latency_min,
                "latency_max": task.latency_max,
                "connect_time_avg": task.connect_time_avg,
                "connect_time_max": task.connect_time_max,
                "apdex": task.apdex,
                "total_sent_bytes": task.total_sent_bytes,
                "total_received_bytes": task.total_received_bytes,
                "sent_kb_per_sec": round(task.total_sent_bytes / 1024 / task.duration_actual, 2) if task.duration_actual > 0 else 0.0,
                "received_kb_per_sec": round(task.total_received_bytes / 1024 / task.duration_actual, 2) if task.duration_actual > 0 else 0.0,
                "status_2xx": status_2xx,
                "status_3xx": status_3xx,
                "status_4xx": status_4xx,
                "status_5xx": status_5xx,
                "error_type_timeout": task.error_type_timeout,
                "error_type_connect_error": task.error_type_connect_error,
                "error_type_client_error": task.error_type_client_error,
                "error_type_http_4xx": task.error_type_http_4xx,
                "error_type_http_5xx": task.error_type_http_5xx,
                "error_type_assertion": task.error_type_assertion,
            },
            "snapshots": [s.to_dict() for s in snapshots],
            "resource_summary": task.resource_summary,
            "target_resource_summary": task.target_resource_summary,
            "transaction_summary": task.transaction_summary,
            "api_summary": task.api_summary,
        }

    async def _build_report_from_execution(
        self, execution_id: str
    ) -> Optional[Dict[str, Any]]:
        """
        从执行记录构建报告数据

        Args:
            execution_id: 执行记录ID

        Returns:
            dict: 报告数据，不存在返回 None
        """
        execution = await self.db.get(PerformanceExecution, execution_id)
        if not execution:
            return None

        # 获取该执行记录关联的快照
        result = await self.db.execute(
            select(PerformanceSnapshot)
            .where(PerformanceSnapshot.execution_id == execution_id)
            .order_by(PerformanceSnapshot.elapsed.asc())
        )
        snapshots = list(result.scalars().all())

        # 汇总状态码分布
        status_2xx = sum(s.status_2xx for s in snapshots)
        status_3xx = sum(s.status_3xx for s in snapshots)
        status_4xx = sum(s.status_4xx for s in snapshots)
        status_5xx = sum(s.status_5xx for s in snapshots)

        return {
            "summary": {
                "task_id": execution.task_id,
                "task_name": execution.task_name,
                "target_url": "",
                "method": "",
                "mode": "",
                "scenario_type": None,
                "concurrency": 0,
                "duration_setting": 0,
                "duration_actual": execution.duration_actual,
                "start_time": execution.start_time.isoformat() if execution.start_time else None,
                "end_time": execution.end_time.isoformat() if execution.end_time else None,
                "account_key": "",
            },
            "metrics": {
                "total_requests": execution.total_requests,
                "success_count": execution.success_count,
                "fail_count": execution.fail_count,
                "error_rate": execution.error_rate,
                "actual_tps": execution.actual_tps,
                "peak_tps": execution.peak_tps,
                "min_response_time": execution.min_response_time,
                "max_response_time": execution.max_response_time,
                "avg_response_time": execution.avg_response_time,
                "p50_response_time": execution.p50_response_time,
                "p90_response_time": execution.p90_response_time,
                "p95_response_time": execution.p95_response_time,
                "p99_response_time": execution.p99_response_time,
                "std_dev": execution.std_dev,
                "latency_avg": execution.latency_avg,
                "latency_min": execution.latency_min,
                "latency_max": execution.latency_max,
                "connect_time_avg": execution.connect_time_avg,
                "connect_time_max": execution.connect_time_max,
                "apdex": execution.apdex,
                "total_sent_bytes": execution.total_sent_bytes,
                "total_received_bytes": execution.total_received_bytes,
                "sent_kb_per_sec": round(execution.total_sent_bytes / 1024 / execution.duration_actual, 2) if execution.duration_actual > 0 else 0.0,
                "received_kb_per_sec": round(execution.total_received_bytes / 1024 / execution.duration_actual, 2) if execution.duration_actual > 0 else 0.0,
                "status_2xx": status_2xx,
                "status_3xx": status_3xx,
                "status_4xx": status_4xx,
                "status_5xx": status_5xx,
                "error_type_timeout": execution.error_type_timeout,
                "error_type_connect_error": execution.error_type_connect_error,
                "error_type_client_error": execution.error_type_client_error,
                "error_type_http_4xx": execution.error_type_http_4xx,
                "error_type_http_5xx": execution.error_type_http_5xx,
                "error_type_assertion": execution.error_type_assertion,
            },
            "snapshots": [s.to_dict() for s in snapshots],
            "resource_summary": execution.resource_summary,
            "target_resource_summary": execution.target_resource_summary,
            "transaction_summary": execution.transaction_summary,
            "api_summary": execution.api_summary,
        }

    # ==================== 执行历史 CRUD ====================

    async def get_executions(
        self,
        task_id: Optional[str] = None,
        project_id: Optional[str] = None,
        page: int = 1,
        page_size: int = 20,
    ) -> Dict[str, Any]:
        """
        获取执行历史列表（按时间倒序）

        Args:
            task_id: 按任务筛选（可选）
            project_id: 按项目筛选（可选）
            page: 页码
            page_size: 每页数量

        Returns:
            dict: {total, items}
        """
        query = select(PerformanceExecution)
        count_query = select(func.count()).select_from(PerformanceExecution)

        if task_id:
            query = query.where(PerformanceExecution.task_id == task_id)
            count_query = count_query.where(PerformanceExecution.task_id == task_id)
        if project_id:
            query = query.where(PerformanceExecution.project_id == project_id)
            count_query = count_query.where(PerformanceExecution.project_id == project_id)

        total = (await self.db.execute(count_query)).scalar() or 0

        result = await self.db.execute(
            query
            .order_by(PerformanceExecution.start_time.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
        executions = list(result.scalars().all())

        return {
            "total": total,
            "items": [{
                "id": e.id,
                "task_id": e.task_id,
                "task_name": e.task_name,
                "project_id": e.project_id,
                "status": e.status,
                "start_time": e.start_time.isoformat() if e.start_time else None,
                "end_time": e.end_time.isoformat() if e.end_time else None,
                "duration_actual": e.duration_actual or 0.0,
                "total_requests": e.total_requests or 0,
                "actual_tps": e.actual_tps or 0.0,
                "avg_response_time": e.avg_response_time or 0.0,
                "p95_response_time": e.p95_response_time or 0.0,
                "error_rate": e.error_rate or 0.0,
                "triggered_by": e.triggered_by or "manual",
                "snapshot_count": e.snapshot_count or 0,
            } for e in executions],
        }

    async def get_execution_report(self, execution_id: str) -> Optional[Dict[str, Any]]:
        """
        获取指定执行记录的报告（同 get_report 但指定 execution_id）

        Args:
            execution_id: 执行记录ID

        Returns:
            dict: 报告数据，不存在返回 None
        """
        return await self._build_report_from_execution(execution_id)

    async def get_execution_snapshots(
        self,
        execution_id: str,
        page: int = 1,
        page_size: int = 500,
    ) -> Tuple[List[PerformanceSnapshot], int]:
        """
        获取指定执行记录的快照数据

        Args:
            execution_id: 执行记录ID
            page: 页码
            page_size: 每页数量

        Returns:
            tuple: (快照列表, 总数)
        """
        query = (
            select(PerformanceSnapshot)
            .where(PerformanceSnapshot.execution_id == execution_id)
            .order_by(PerformanceSnapshot.elapsed.asc())
        )
        count_query = (
            select(func.count())
            .select_from(PerformanceSnapshot)
            .where(PerformanceSnapshot.execution_id == execution_id)
        )

        total = (await self.db.execute(count_query)).scalar() or 0
        query = query.offset((page - 1) * page_size).limit(page_size)
        result = await self.db.execute(query)
        snapshots = list(result.scalars().all())

        return snapshots, total

    async def delete_execution(self, execution_id: str) -> bool:
        """
        删除执行记录（级联删除关联快照）

        Args:
            execution_id: 执行记录ID

        Returns:
            bool: 是否删除成功
        """
        execution = await self.db.get(PerformanceExecution, execution_id)
        if not execution:
            return False

        await self.db.delete(execution)
        await self.db.commit()
        logger.info(f"删除执行记录: {execution_id}")
        return True

    async def _migrate_legacy_executions(self) -> int:
        """
        存量数据回填：将现有任务的最新执行结果转为执行记录

        幂等设计：仅对 triggered_by='legacy' 的记录执行，已存在 legacy 记录则跳过。

        Returns:
            int: 回填的执行记录数
        """
        # 检查是否已有 legacy 回填记录
        existing = await self.db.execute(
            select(func.count()).select_from(PerformanceExecution)
            .where(PerformanceExecution.triggered_by == "legacy")
        )
        if existing.scalar() and existing.scalar() > 0:
            logger.info("存量回填已执行过，跳过")
            return 0

        # 查询所有已完成的性能任务
        result = await self.db.execute(
            select(PerformanceTask)
            .where(PerformanceTask.status.in_(["completed", "failed"]))
        )
        tasks = list(result.scalars().all())
        migrated = 0

        for task in tasks:
            try:
                # 创建执行记录
                execution = PerformanceExecution(
                    id=generate_id("exec"),
                    task_id=task.id,
                    task_name=task.name,
                    project_id=task.project_id,
                    status=task.status,
                    error_message=task.error_message,
                    start_time=task.start_time,
                    end_time=task.end_time,
                    duration_actual=task.duration_actual or 0.0,
                    total_requests=task.total_requests or 0,
                    success_count=task.success_count or 0,
                    fail_count=task.fail_count or 0,
                    error_rate=task.error_rate or 0.0,
                    actual_tps=task.actual_tps or 0.0,
                    peak_tps=task.peak_tps or 0.0,
                    min_response_time=task.min_response_time or 0.0,
                    max_response_time=task.max_response_time or 0.0,
                    avg_response_time=task.avg_response_time or 0.0,
                    p50_response_time=task.p50_response_time or 0.0,
                    p90_response_time=task.p90_response_time or 0.0,
                    p95_response_time=task.p95_response_time or 0.0,
                    p99_response_time=task.p99_response_time or 0.0,
                    std_dev=task.std_dev or 0.0,
                    latency_avg=task.latency_avg or 0.0,
                    latency_min=task.latency_min or 0.0,
                    latency_max=task.latency_max or 0.0,
                    connect_time_avg=task.connect_time_avg or 0.0,
                    connect_time_max=task.connect_time_max or 0.0,
                    apdex=task.apdex or 1.0,
                    total_sent_bytes=task.total_sent_bytes or 0,
                    total_received_bytes=task.total_received_bytes or 0,
                    error_type_timeout=task.error_type_timeout or 0,
                    error_type_connect_error=task.error_type_connect_error or 0,
                    error_type_client_error=task.error_type_client_error or 0,
                    error_type_http_4xx=task.error_type_http_4xx or 0,
                    error_type_http_5xx=task.error_type_http_5xx or 0,
                    error_type_assertion=task.error_type_assertion or 0,
                    status_2xx=task.status_2xx or 0,
                    status_3xx=task.status_3xx or 0,
                    status_4xx=task.status_4xx or 0,
                    status_5xx=task.status_5xx or 0,
                    resource_summary=task.resource_summary,
                    target_resource_summary=task.target_resource_summary,
                    transaction_summary=task.transaction_summary,
                    api_summary=task.api_summary,
                    snapshot_count=0,
                    triggered_by="legacy",
                    created_at=task.created_at or datetime.utcnow(),
                )
                self.db.add(execution)

                # 回填存量快照的 execution_id
                snap_result = await self.db.execute(
                    select(PerformanceSnapshot)
                    .where(PerformanceSnapshot.task_id == task.id)
                    .where(PerformanceSnapshot.execution_id.is_(None))
                )
                old_snapshots = list(snap_result.scalars().all())
                for snap in old_snapshots:
                    snap.execution_id = execution.id
                execution.snapshot_count = len(old_snapshots)

                migrated += 1
                logger.info(f"存量回填: task={task.id} -> execution={execution.id}, 快照={len(old_snapshots)}条")

            except Exception as e:
                logger.error(f"存量回填失败 task={task.id}: {e}")
                continue

        if migrated > 0:
            await self.db.flush()
            logger.info(f"存量回填完成: 共 {migrated} 条")

        return migrated

    # ==================== 异步非阻塞执行 ====================

    async def run_task_async(
        self,
        task_id: str,
        config_override: Optional[Dict[str, Any]] = None,
        triggered_by: str = "manual",
        batch_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        异步非阻塞执行性能测试任务

        创建执行记录后立即返回，在后台异步执行。
        前台无需等待任务完成即可返回 execution_id，前端据此跳转监控页。

        Args:
            task_id: 任务ID
            config_override: 可覆盖的配置项
            triggered_by: 触发方式
            batch_id: 关联批次ID

        Returns:
            dict: {message, task_id, execution_id, status}
        """
        task = await self.get_task(task_id)
        if not task:
            raise ValueError(f"任务不存在: {task_id}")

        if task.status == "running":
            raise ValueError("任务正在执行中，不能重复启动")

        # 创建执行记录（running 状态）
        execution_id = generate_id("exec")
        execution = PerformanceExecution(
            id=execution_id,
            task_id=task_id,
            task_name=task.name,
            project_id=task.project_id,
            status="running",
            start_time=datetime.utcnow(),
            triggered_by=triggered_by,
            batch_id=batch_id,
            created_at=datetime.utcnow(),
        )
        self.db.add(execution)
        await self.db.commit()

        # 在后台异步执行
        bg_task = asyncio.create_task(
            self._run_single_background(
                task_id=task_id,
                execution_id=execution_id,
                config_override=config_override,
                triggered_by=triggered_by,
            )
        )
        self._background_tasks.add(bg_task)
        bg_task.add_done_callback(self._background_tasks.discard)

        return {
            "message": "任务已开始执行",
            "task_id": task_id,
            "execution_id": execution_id,
            "status": "running",
        }

    async def _run_single_background(
        self,
        task_id: str,
        execution_id: str,
        config_override: Optional[Dict[str, Any]] = None,
        triggered_by: str = "manual",
    ) -> None:
        """
        后台单任务执行（使用独立数据库会话）

        Args:
            task_id: 任务ID
            execution_id: 执行记录ID
            config_override: 可覆盖的配置项
            triggered_by: 触发方式
        """
        from app.database import async_session_maker

        async with async_session_maker() as db:
            try:
                service = PerformanceService(db)
                await service.run_task(
                    task_id=task_id,
                    config_override=config_override,
                    triggered_by=triggered_by,
                    execution_id=execution_id,
                )
            except Exception as e:
                logger.error(f"后台单任务执行失败 [{task_id}]: {e}")
                try:
                    exec_record = await db.get(PerformanceExecution, execution_id)
                    if exec_record:
                        exec_record.status = "failed"
                        exec_record.error_message = str(e)
                        await db.commit()
                except Exception:
                    pass

    @staticmethod
    def _report_to_dict(report: PerformanceProjectReport) -> Dict[str, Any]:
        """将合并报告 ORM 对象转为 API 响应字典"""
        task_summaries = []
        try:
            task_summaries = json.loads(report.task_summaries or "[]")
        except (json.JSONDecodeError, TypeError):
            pass

        return {
            "id": report.id,
            "project_id": report.project_id,
            "batch_id": report.batch_id,
            "created_at": report.created_at.isoformat() if report.created_at else None,
            "summary": {
                "total_requests": report.total_requests,
                "success_count": report.success_count,
                "fail_count": report.fail_count,
                "error_rate": report.error_rate,
                "actual_tps": report.actual_tps,
                "avg_response_time": report.avg_response_time,
                "min_response_time": report.min_response_time,
                "max_response_time": report.max_response_time,
                "p50_response_time": report.p50_response_time,
                "p90_response_time": report.p90_response_time,
                "p95_response_time": report.p95_response_time,
                "p99_response_time": report.p99_response_time,
                "std_dev": report.std_dev,
                "apdex": report.apdex,
                "status_2xx": report.status_2xx,
                "status_3xx": report.status_3xx,
                "status_4xx": report.status_4xx,
                "status_5xx": report.status_5xx,
                "total_sent_bytes": report.total_sent_bytes,
                "total_received_bytes": report.total_received_bytes,
                "peak_tps": report.peak_tps,
                "duration_actual": report.duration_actual,
            },
            "task_summaries": task_summaries,
        }

    async def get_project_report(self, project_id: str) -> Optional[Dict[str, Any]]:
        """
        获取项目最新合并报告

        Args:
            project_id: 项目ID

        Returns:
            dict: 合并报告数据，无则返回 None
        """
        result = await self.db.execute(
            select(PerformanceProjectReport)
            .where(PerformanceProjectReport.project_id == project_id)
            .order_by(PerformanceProjectReport.created_at.desc())
            .limit(1)
        )
        report = result.scalar_one_or_none()
        if not report:
            return None
        return self._report_to_dict(report)

    async def get_batch_report(self, batch_id: str) -> Optional[Dict[str, Any]]:
        """
        按批次ID获取合并报告

        Args:
            batch_id: 批量执行ID

        Returns:
            dict: 合并报告数据，无则返回 None
        """
        result = await self.db.execute(
            select(PerformanceProjectReport).where(
                PerformanceProjectReport.batch_id == batch_id
            )
        )
        report = result.scalar_one_or_none()
        if not report:
            return None
        return self._report_to_dict(report)

    async def list_project_reports(
        self, project_id: str, page: int = 1, page_size: int = 20
    ) -> Dict[str, Any]:
        """
        获取项目历史合并报告列表

        Args:
            project_id: 项目ID
            page: 页码（从 1 开始）
            page_size: 每页数量

        Returns:
            dict: {total, items}
        """
        base_query = select(PerformanceProjectReport).where(
            PerformanceProjectReport.project_id == project_id
        )

        # 总数
        count_result = await self.db.execute(
            select(func.count()).select_from(PerformanceProjectReport).where(
                PerformanceProjectReport.project_id == project_id
            )
        )
        total = count_result.scalar() or 0

        # 分页列表
        result = await self.db.execute(
            base_query
            .order_by(PerformanceProjectReport.created_at.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
        reports = list(result.scalars().all())

        return {
            "total": total,
            "items": [self._report_to_dict(r) for r in reports],
        }

    # ==================== 内部方法 ====================

    @staticmethod
    def _build_client_config(task_dict: Dict[str, Any]) -> Dict[str, Any]:
        """
        构建 HttpClient 配置

        被测系统 base_url 与认证路径（登录/验证码）均从系统配置
        （settings，即系统管理-系统配置保存的值）运行时动态读取，
        不再依赖任务中填写的 target_url——切换被测环境后无需改任务。

        Args:
            task_dict: 任务数据字典

        Returns:
            dict: HttpClient 配置字典
        """
        # 构建 accounts 字典（优先使用自定义凭据）
        accounts = dict(_DEFAULT_ACCOUNTS)
        if task_dict.get("login_username") and task_dict.get("login_password"):
            key = task_dict.get("account_key", "custom")
            accounts[key] = {
                "username": task_dict["login_username"],
                "password": task_dict["login_password"],
            }

        return {
            "target": {
                "base_url": settings.TARGET_URL or "",
                "verify_ssl": False,
                "timeout": 30,
            },
            "accounts": accounts,
            "auth": {
                "login_path": settings.AUTH_LOGIN_PATH or "/platform/api/auth/login",
                "captcha_path": settings.AUTH_CAPTCHA_PATH or "/platform/api/code",
            },
        }

    @staticmethod
    def _apply_result(task: PerformanceTask, result: Dict[str, Any]) -> None:
        """
        将执行结果写入任务统计字段

        Args:
            task: 任务对象
            result: 执行结果字典
        """
        task.total_requests = result.get("total_requests", 0)
        task.success_count = result.get("success_count", 0)
        task.fail_count = result.get("fail_count", 0)
        task.min_response_time = result.get("min_response_time", 0.0)
        task.max_response_time = result.get("max_response_time", 0.0)
        task.avg_response_time = result.get("avg_response_time", 0.0)
        task.p50_response_time = result.get("p50_response_time", 0.0)
        task.p90_response_time = result.get("p90_response_time", 0.0)
        task.p95_response_time = result.get("p95_response_time", 0.0)
        task.p99_response_time = result.get("p99_response_time", 0.0)
        task.std_dev = result.get("std_dev", 0.0)
        task.latency_avg = result.get("latency_avg", 0.0)
        task.latency_min = result.get("latency_min", 0.0)
        task.latency_max = result.get("latency_max", 0.0)
        task.connect_time_avg = result.get("connect_time_avg", 0.0)
        task.connect_time_max = result.get("connect_time_max", 0.0)
        task.peak_tps = result.get("peak_tps", 0.0)
        task.total_sent_bytes = result.get("total_sent_bytes", 0)
        task.total_received_bytes = result.get("total_received_bytes", 0)
        task.apdex = result.get("apdex", 1.0)
        task.error_type_timeout = result.get("error_type_timeout", 0)
        task.error_type_connect_error = result.get("error_type_connect_error", 0)
        task.error_type_client_error = result.get("error_type_client_error", 0)
        task.error_type_http_4xx = result.get("error_type_http_4xx", 0)
        task.error_type_http_5xx = result.get("error_type_http_5xx", 0)
        task.error_type_assertion = result.get("error_type_assertion", 0)
        task.status_2xx = result.get("status_2xx", 0)
        task.status_3xx = result.get("status_3xx", 0)
        task.status_4xx = result.get("status_4xx", 0)
        task.status_5xx = result.get("status_5xx", 0)
        task.actual_tps = result.get("actual_tps", 0.0)
        task.error_rate = result.get("error_rate", 0.0)
        # 执行机资源监控汇总（资源监控功能）
        task.resource_summary = result.get("resource_summary")
        # 目标机资源监控汇总（SSH 方式采集被测系统资源）
        task.target_resource_summary = result.get("target_resource_summary")
        # 事务处理时间汇总（事务功能）
        task.transaction_summary = result.get("transaction_summary")
        # 接口维度指标汇总（长稳压测多接口场景）
        task.api_summary = result.get("api_summary")

    @staticmethod
    def _copy_result_to_execution(
        execution: PerformanceExecution, result: Dict[str, Any]
    ) -> None:
        """
        将执行结果写入执行记录统计字段（与 _apply_result 的字段一一对应）

        Args:
            execution: 执行记录对象
            result: 执行结果字典
        """
        execution.total_requests = result.get("total_requests", 0)
        execution.success_count = result.get("success_count", 0)
        execution.fail_count = result.get("fail_count", 0)
        execution.min_response_time = result.get("min_response_time", 0.0)
        execution.max_response_time = result.get("max_response_time", 0.0)
        execution.avg_response_time = result.get("avg_response_time", 0.0)
        execution.p50_response_time = result.get("p50_response_time", 0.0)
        execution.p90_response_time = result.get("p90_response_time", 0.0)
        execution.p95_response_time = result.get("p95_response_time", 0.0)
        execution.p99_response_time = result.get("p99_response_time", 0.0)
        execution.std_dev = result.get("std_dev", 0.0)
        execution.latency_avg = result.get("latency_avg", 0.0)
        execution.latency_min = result.get("latency_min", 0.0)
        execution.latency_max = result.get("latency_max", 0.0)
        execution.connect_time_avg = result.get("connect_time_avg", 0.0)
        execution.connect_time_max = result.get("connect_time_max", 0.0)
        execution.peak_tps = result.get("peak_tps", 0.0)
        execution.total_sent_bytes = result.get("total_sent_bytes", 0)
        execution.total_received_bytes = result.get("total_received_bytes", 0)
        execution.apdex = result.get("apdex", 1.0)
        execution.error_type_timeout = result.get("error_type_timeout", 0)
        execution.error_type_connect_error = result.get("error_type_connect_error", 0)
        execution.error_type_client_error = result.get("error_type_client_error", 0)
        execution.error_type_http_4xx = result.get("error_type_http_4xx", 0)
        execution.error_type_http_5xx = result.get("error_type_http_5xx", 0)
        execution.error_type_assertion = result.get("error_type_assertion", 0)
        execution.status_2xx = result.get("status_2xx", 0)
        execution.status_3xx = result.get("status_3xx", 0)
        execution.status_4xx = result.get("status_4xx", 0)
        execution.status_5xx = result.get("status_5xx", 0)
        execution.actual_tps = result.get("actual_tps", 0.0)
        execution.error_rate = result.get("error_rate", 0.0)
        # 执行机资源监控汇总（资源监控功能）
        execution.resource_summary = result.get("resource_summary")
        # 目标机资源监控汇总（SSH 方式采集被测系统资源）
        execution.target_resource_summary = result.get("target_resource_summary")
        # 事务处理时间汇总（事务功能）
        execution.transaction_summary = result.get("transaction_summary")
        # 接口维度指标汇总（长稳压测多接口场景）
        execution.api_summary = result.get("api_summary")

    async def _save_snapshots(
        self,
        task_id: str,
        snapshots: List[Dict[str, Any]],
        execution_id: Optional[str] = None,
    ) -> int:
        """
        批量保存快照到数据库

        Args:
            task_id: 任务ID
            snapshots: 快照数据列表
            execution_id: 关联执行记录ID（新执行写入；None 用于存量回填）

        Returns:
            int: 保存的快照数量
        """
        if not snapshots:
            return 0

        batch = []
        for s in snapshots:
            batch.append(PerformanceSnapshot(
                task_id=task_id,
                execution_id=execution_id,
                timestamp=s.get("timestamp", time.time()),
                elapsed=s.get("elapsed", 0.0),
                concurrent_users=s.get("concurrent_users", 0),
                tps=s.get("tps", 0.0),
                avg_response_time=s.get("avg_response_time", 0.0),
                min_response_time=s.get("min_response_time", 0.0),
                max_response_time=s.get("max_response_time", 0.0),
                p50_response_time=s.get("p50", 0.0),
                p90_response_time=s.get("p90", 0.0),
                p95_response_time=s.get("p95", 0.0),
                p99_response_time=s.get("p99", 0.0),
                std_dev=s.get("std_dev", 0.0),
                latency_avg=s.get("latency_avg", 0.0),
                connect_time_avg=s.get("connect_time_avg", 0.0),
                sent_bytes=s.get("sent_bytes", 0),
                received_bytes=s.get("received_bytes", 0),
                peak_tps=s.get("peak_tps", 0.0),
                success_count=s.get("success_count", 0),
                fail_count=s.get("fail_count", 0),
                status_2xx=s.get("status_2xx", 0),
                status_3xx=s.get("status_3xx", 0),
                status_4xx=s.get("status_4xx", 0),
                status_5xx=s.get("status_5xx", 0),
            ))

        self.db.add_all(batch)
        await self.db.flush()
        logger.info(f"批量保存快照: {len(batch)} 条 (task_id={task_id}, execution_id={execution_id})")
        return len(batch)

    async def _save_outputs(
        self, task_id: str, captured: Optional[Dict[str, List[Any]]]
    ) -> int:
        """
        保存响应捕获输出到数据库

        使用 INSERT ... ON CONFLICT 语义（先查后写），
        每个 task_id+key 唯一，已存在时追加值。

        Args:
            task_id: 任务ID
            captured: 捕获数据 {key: [values]}

        Returns:
            int: 保存/更新的输出条目数
        """
        if not captured:
            return 0

        saved = 0
        for key, values in captured.items():
            if not values:
                continue

            # 查重：同一 task_id+key 已存在则追加
            result = await self.db.execute(
                select(PerformanceTaskOutput)
                .where(
                    PerformanceTaskOutput.task_id == task_id,
                    PerformanceTaskOutput.key == key,
                )
            )
            existing = result.scalar_one_or_none()

            if existing:
                # 合并去重
                existing_values = list(existing.value) if existing.value else []
                seen = set(str(v) for v in existing_values)
                for v in values:
                    key_str = str(v)
                    if key_str not in seen:
                        existing_values.append(v)
                        seen.add(key_str)
                existing.value = existing_values
            else:
                self.db.add(PerformanceTaskOutput(
                    task_id=task_id,
                    key=key,
                    value=list(values),
                ))
            saved += 1

        await self.db.flush()
        logger.info(f"保存响应捕获输出: {saved} 条 (task_id={task_id})")
        return saved

    def _build_ref_provider(self) -> Callable[[str, str], list]:
        """
        构建跨任务引用取值回调

        供 TemplateResolver 的 REF 占位符使用，
        用 asyncio.run_coroutine_threadsafe 在事件循环中查询数据库。

        Returns:
            Callable: (task_id, key) -> list
        """
        db = self.db

        def _ref_provider(task_id: str, key: str) -> list:
            """同步回调，在 executor 工作线程中调用"""
            try:
                loop = asyncio.get_event_loop()
                if loop.is_running():
                    future = asyncio.run_coroutine_threadsafe(
                        self._query_output(task_id, key), loop
                    )
                    return future.result(timeout=5) or []
                return []
            except Exception as e:
                logger.warning(f"REF 查询失败 task_id={task_id}, key={key}: {e}")
                return []

        return _ref_provider

    async def _query_output(self, task_id: str, key: str) -> list:
        """
        查询某个任务的输出值

        Args:
            task_id: 任务ID
            key: 输出键名

        Returns:
            list: 输出值列表
        """
        result = await self.db.execute(
            select(PerformanceTaskOutput)
            .where(
                PerformanceTaskOutput.task_id == task_id,
                PerformanceTaskOutput.key == key,
            )
        )
        output = result.scalar_one_or_none()
        return list(output.value) if output and output.value else []

    # ==================== 项目 CRUD ====================

    async def list_projects(
        self,
        page: int = 1,
        page_size: int = 100,
        project_type: Optional[str] = None,
    ) -> Tuple[List[PerformanceProject], int]:
        """
        获取项目列表

        Args:
            page: 页码
            page_size: 每页数量
            project_type: 按类型筛选 (standard/custom)

        Returns:
            tuple: (项目列表, 总数)
        """
        query = select(PerformanceProject)
        count_query = select(func.count()).select_from(PerformanceProject)

        if project_type:
            query = query.where(PerformanceProject.project_type == project_type)
            count_query = count_query.where(PerformanceProject.project_type == project_type)

        total = (await self.db.execute(count_query)).scalar() or 0

        query = (
            query.order_by(PerformanceProject.sort_order.asc(), desc(PerformanceProject.created_at))
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
        result = await self.db.execute(query)
        projects = list(result.scalars().all())

        return projects, total

    async def create_project(self, data: PerformanceProjectCreate) -> PerformanceProject:
        """
        创建项目

        Args:
            data: 项目创建数据

        Returns:
            PerformanceProject: 创建的项目
        """
        payload = data.model_dump()
        project = PerformanceProject(
            id=generate_id("proj"),
            **payload,
        )
        self.db.add(project)
        await self.db.commit()
        await self.db.refresh(project)
        logger.info(f"创建性能测试项目: {project.id} - {project.name}")
        return project

    async def get_project(self, project_id: str) -> Optional[PerformanceProject]:
        """
        获取项目详情

        Args:
            project_id: 项目ID

        Returns:
            PerformanceProject: 项目对象，不存在返回 None
        """
        return await self.db.get(PerformanceProject, project_id)

    async def update_project(
        self, project_id: str, data: Dict[str, Any]
    ) -> Optional[PerformanceProject]:
        """
        更新项目

        Args:
            project_id: 项目ID
            data: 更新数据

        Returns:
            PerformanceProject: 更新后的项目，不存在返回 None
        """
        project = await self.get_project(project_id)
        if not project:
            return None

        for key, value in data.items():
            if value is not None and hasattr(project, key):
                setattr(project, key, value)

        await self.db.commit()
        await self.db.refresh(project)
        return project

    async def delete_project(self, project_id: str) -> bool:
        """
        删除项目（不级联删除任务，仅置空 project_id）

        Args:
            project_id: 项目ID

        Returns:
            bool: 是否成功删除
        """
        project = await self.get_project(project_id)
        if not project:
            return False

        # 解除关联：将该项目下所有任务的 project_id 置为 NULL
        stmt = (
            select(PerformanceTask)
            .where(PerformanceTask.project_id == project_id)
        )
        result = await self.db.execute(stmt)
        tasks = list(result.scalars().all())
        for task in tasks:
            task.project_id = None

        await self.db.delete(project)
        await self.db.commit()
        logger.info(f"删除性能测试项目: {project_id}")
        return True

    async def get_project_stats(self, project_id: str) -> Dict[str, Any]:
        """
        获取项目统计信息

        Args:
            project_id: 项目ID

        Returns:
            dict: 统计信息（任务数、状态分布等）
        """
        # 查询该项目下的所有任务
        result = await self.db.execute(
            select(PerformanceTask)
            .where(PerformanceTask.project_id == project_id)
        )
        tasks = list(result.scalars().all())

        task_count = len(tasks)
        status_distribution = {}
        last_run_status = None
        last_run_time = None

        for task in tasks:
            status_distribution[task.status] = status_distribution.get(task.status, 0) + 1
            # 最近执行的任务（start_time 非空且最近）
            if task.start_time and (
                last_run_time is None or task.start_time > last_run_time
            ):
                last_run_time = task.start_time
                last_run_status = task.status

        return {
            "project_id": project_id,
            "task_count": task_count,
            "status_distribution": status_distribution,
            "last_run_status": last_run_status,
            "last_run_time": last_run_time.isoformat() if last_run_time else None,
        }

    # ==================== 批量执行 ====================

    @staticmethod
    def _weighted_percentile(
        percentiles: List[float], counts: List[int], p: float
    ) -> float:
        """
        加权近似分位数合并算法

        各任务的 P 分位数独立计算，合并时无法精确还原全局分位数，
        采用加权近似法：按分位数值排序，累计请求数到达目标位置时取值。

        Args:
            percentiles: 各任务的指定分位数列表
            counts: 各任务请求数列表
            p: 目标分位数 (50/90/95/99)

        Returns:
            float: 加权近似分位数
        """
        if not percentiles:
            return 0.0
        # 过滤请求数为 0 的任务（无效权重）
        pairs = [(v, c) for v, c in zip(percentiles, counts) if c > 0]
        if not pairs:
            return 0.0
        pairs.sort(key=lambda x: x[0])
        total = sum(c for _, c in pairs)
        if total <= 0:
            return 0.0
        target = total * p / 100.0
        cumulative = 0
        for val, cnt in pairs:
            cumulative += cnt
            if cumulative >= target:
                return val
        return pairs[-1][0]

    async def _aggregate_task_results(self, task_ids: List[str]) -> Dict[str, Any]:
        """
        聚合多个任务的执行结果为合并报告指标

        聚合算法（参考 PRD 3.3）：
        - 求和：total_requests / success_count / fail_count / 状态码 / 字节数 / duration_actual
        - 重算：error_rate = fail / total，actual_tps = total_requests / duration_actual
        - 加权平均：avg_response_time / apdex（权重为各任务请求数）
        - 取极值：min_response_time / max_response_time / peak_tps
        - 加权近似：p50 / p90 / p95 / p99（_weighted_percentile）
        - 标准差：基于聚合均值和方差重算（√(Σ((σ_i² + μ_i²) × n_i) / N - 总体均值²)）

        Args:
            task_ids: 任务ID列表

        Returns:
            dict: 合并报告聚合指标 + 任务明细
        """
        if not task_ids:
            return {}

        result = await self.db.execute(
            select(PerformanceTask).where(PerformanceTask.id.in_(task_ids))
        )
        tasks = list(result.scalars().all())
        if not tasks:
            return {}

        # ---- 基础求和 ----
        total_requests = sum(t.total_requests for t in tasks)
        success_count = sum(t.success_count for t in tasks)
        fail_count = sum(t.fail_count for t in tasks)
        status_2xx = sum(t.status_2xx for t in tasks)
        status_3xx = sum(t.status_3xx for t in tasks)
        status_4xx = sum(t.status_4xx for t in tasks)
        status_5xx = sum(t.status_5xx for t in tasks)
        total_sent_bytes = sum(t.total_sent_bytes for t in tasks)
        total_received_bytes = sum(t.total_received_bytes for t in tasks)
        duration_actual = sum(t.duration_actual or 0 for t in tasks)

        # ---- 重算比率 ----
        error_rate = (fail_count / total_requests) if total_requests > 0 else 0.0
        actual_tps = (total_requests / duration_actual) if duration_actual > 0 else 0.0

        # ---- 加权平均 ----
        avg_response_time = (
            sum((t.avg_response_time or 0) * t.total_requests for t in tasks) / total_requests
            if total_requests > 0 else 0.0
        )
        apdex = (
            sum((t.apdex if t.apdex is not None else 1.0) * t.total_requests for t in tasks) / total_requests
            if total_requests > 0 else 0.0
        )

        # ---- 极值 ----
        min_response_time = min((t.min_response_time or 0) for t in tasks)
        max_response_time = max((t.max_response_time or 0) for t in tasks)
        peak_tps = max((t.peak_tps or 0) for t in tasks)

        # ---- 加权近似分位数 ----
        counts = [t.total_requests for t in tasks]
        p50 = self._weighted_percentile([t.p50_response_time or 0 for t in tasks], counts, 50)
        p90 = self._weighted_percentile([t.p90_response_time or 0 for t in tasks], counts, 90)
        p95 = self._weighted_percentile([t.p95_response_time or 0 for t in tasks], counts, 95)
        p99 = self._weighted_percentile([t.p99_response_time or 0 for t in tasks], counts, 99)

        # ---- 标准差（合并方差公式） ----
        # σ_total² = Σ((σ_i² + μ_i²) × n_i) / N - μ_total²
        if total_requests > 0:
            pooled_e2 = sum(
                ((t.std_dev or 0) ** 2 + (t.avg_response_time or 0) ** 2) * t.total_requests
                for t in tasks
            ) / total_requests
            std_dev = max(0.0, (pooled_e2 - avg_response_time ** 2) ** 0.5)
        else:
            std_dev = 0.0

        # ---- 任务明细 ----
        task_summaries = []
        for t in tasks:
            t_error_rate = (t.fail_count / t.total_requests) if t.total_requests > 0 else 0.0
            task_summaries.append({
                "task_id": t.id,
                "task_name": t.name,
                "status": t.status,
                "total_requests": t.total_requests,
                "success_count": t.success_count,
                "fail_count": t.fail_count,
                "avg_response_time": t.avg_response_time,
                "actual_tps": t.actual_tps,
                "error_rate": t_error_rate,
            })

        return {
            "total_requests": total_requests,
            "success_count": success_count,
            "fail_count": fail_count,
            "error_rate": round(error_rate, 6),
            "actual_tps": round(actual_tps, 2),
            "avg_response_time": round(avg_response_time, 2),
            "min_response_time": round(min_response_time, 2),
            "max_response_time": round(max_response_time, 2),
            "p50_response_time": round(p50, 2),
            "p90_response_time": round(p90, 2),
            "p95_response_time": round(p95, 2),
            "p99_response_time": round(p99, 2),
            "std_dev": round(std_dev, 2),
            "apdex": round(apdex, 4),
            "status_2xx": status_2xx,
            "status_3xx": status_3xx,
            "status_4xx": status_4xx,
            "status_5xx": status_5xx,
            "total_sent_bytes": total_sent_bytes,
            "total_received_bytes": total_received_bytes,
            "peak_tps": round(peak_tps, 2),
            "duration_actual": round(duration_actual, 2),
            "task_summaries": task_summaries,
        }

    async def batch_run_tasks(
        self,
        task_ids: List[str],
        project_id: Optional[str] = None,
        progress_callback: Optional[Callable] = None,
        batch_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        批量执行多个任务（串行，一个完成后再执行下一个）

        Args:
            task_ids: 任务ID列表
            project_id: 关联项目ID（可选）
            progress_callback: 进度回调
            batch_id: 已存在的批次ID（由 create_batch_and_start 创建）；不传则新建

        Returns:
            dict: 批量执行结果汇总
        """
        if not batch_id:
            batch_id = generate_id("batch")
            # 持久化批量执行记录（running）
            batch_record = PerformanceBatchExecution(
                id=batch_id,
                project_id=project_id,
                task_ids=json.dumps(task_ids, ensure_ascii=False),
                status="running",
                total_count=len(task_ids),
                success_count=0,
                fail_count=0,
                started_at=datetime.utcnow(),
                created_at=datetime.utcnow(),
            )
            self.db.add(batch_record)
            await self.db.commit()
        else:
            # 复用已创建的批次记录
            batch_record = await self.db.get(PerformanceBatchExecution, batch_id)
            if not batch_record:
                batch_record = PerformanceBatchExecution(
                    id=batch_id,
                    project_id=project_id,
                    task_ids=json.dumps(task_ids, ensure_ascii=False),
                    status="running",
                    total_count=len(task_ids),
                    success_count=0,
                    fail_count=0,
                    started_at=datetime.utcnow(),
                    created_at=datetime.utcnow(),
                )
                self.db.add(batch_record)
                await self.db.commit()

        total = len(task_ids)
        success_count = 0
        fail_count = 0
        results = []

        # 广播分组：全局分组（BatchMonitor 兼容）+ 批次专属分组（/ws/batch/{batch_id}）
        ws_group = "perf_batch"
        batch_group = f"perf_batch_{batch_id}"

        async def _broadcast_progress(payload: Dict[str, Any]) -> None:
            """同时广播到全局分组与批次专属分组"""
            for group in (ws_group, batch_group):
                try:
                    await manager.broadcast(group, payload)
                except Exception as e:
                    logger.warning(f"批量进度广播失败 [{group}]: {e}")

        # 广播批量执行开始
        await _broadcast_progress({
            "type": "perf_batch_progress",
            "batch_id": batch_id,
            "project_id": project_id,
            "current_index": 0,
            "total_count": total,
            "current_task_id": None,
            "current_task_name": None,
            "current_status": "running",
            "success_count": 0,
            "fail_count": 0,
        })

        for idx, task_id in enumerate(task_ids):
            current_index = idx + 1
            task = None
            try:
                # 获取任务名称用于进度展示
                task = await self.get_task(task_id)
                task_name = task.name if task else task_id

                # 广播当前任务开始
                await _broadcast_progress({
                    "type": "perf_batch_progress",
                    "batch_id": batch_id,
                    "project_id": project_id,
                    "current_index": current_index,
                    "total_count": total,
                    "current_task_id": task_id,
                    "current_task_name": task_name,
                    "current_status": "running",
                    "success_count": success_count,
                    "fail_count": fail_count,
                })

                # 执行单个任务（串行，等待完成）
                result = await self.run_task(task_id, triggered_by="batch", batch_id=batch_id)

                if result.get("status") == "completed":
                    success_count += 1
                else:
                    fail_count += 1

                results.append({
                    "task_id": task_id,
                    "status": result.get("status", "failed"),
                })

            except Exception as e:
                fail_count += 1
                logger.error(f"批量执行任务失败 [{task_id}]: {e}")
                results.append({
                    "task_id": task_id,
                    "status": "failed",
                    "error": str(e),
                })

            # 广播当前任务完成
            await _broadcast_progress({
                "type": "perf_batch_progress",
                "batch_id": batch_id,
                "project_id": project_id,
                "current_index": current_index,
                "total_count": total,
                "current_task_id": task_id,
                "current_task_name": task.name if task else task_id,
                "current_status": "completed",
                "success_count": success_count,
                "fail_count": fail_count,
            })

        # 广播批量执行完成
        final_status = "completed" if fail_count == 0 else "partially_failed"
        await _broadcast_progress({
            "type": "perf_batch_progress",
            "batch_id": batch_id,
            "project_id": project_id,
            "current_index": total,
            "total_count": total,
            "current_task_id": None,
            "current_task_name": None,
            "current_status": final_status,
            "success_count": success_count,
            "fail_count": fail_count,
        })

        # ---- 生成合并报告 ----
        report_id = None
        try:
            if project_id:
                aggregated = await self._aggregate_task_results(task_ids)
                if aggregated:
                    report_id = generate_id("rpt")
                    report = PerformanceProjectReport(
                        id=report_id,
                        project_id=project_id,
                        batch_id=batch_id,
                        task_ids=json.dumps(task_ids, ensure_ascii=False),
                        total_requests=aggregated.get("total_requests", 0),
                        success_count=aggregated.get("success_count", 0),
                        fail_count=aggregated.get("fail_count", 0),
                        error_rate=aggregated.get("error_rate", 0.0),
                        actual_tps=aggregated.get("actual_tps", 0.0),
                        avg_response_time=aggregated.get("avg_response_time", 0.0),
                        min_response_time=aggregated.get("min_response_time", 0.0),
                        max_response_time=aggregated.get("max_response_time", 0.0),
                        p50_response_time=aggregated.get("p50_response_time", 0.0),
                        p90_response_time=aggregated.get("p90_response_time", 0.0),
                        p95_response_time=aggregated.get("p95_response_time", 0.0),
                        p99_response_time=aggregated.get("p99_response_time", 0.0),
                        std_dev=aggregated.get("std_dev", 0.0),
                        apdex=aggregated.get("apdex", 1.0),
                        status_2xx=aggregated.get("status_2xx", 0),
                        status_3xx=aggregated.get("status_3xx", 0),
                        status_4xx=aggregated.get("status_4xx", 0),
                        status_5xx=aggregated.get("status_5xx", 0),
                        total_sent_bytes=aggregated.get("total_sent_bytes", 0),
                        total_received_bytes=aggregated.get("total_received_bytes", 0),
                        peak_tps=aggregated.get("peak_tps", 0.0),
                        duration_actual=aggregated.get("duration_actual", 0.0),
                        task_summaries=json.dumps(
                            aggregated.get("task_summaries", []), ensure_ascii=False
                        ),
                        created_at=datetime.utcnow(),
                    )
                    self.db.add(report)
        except Exception as e:
            logger.error(f"生成合并报告失败: {e}")
            report_id = None

        # 更新批量执行记录状态
        try:
            batch_record.status = final_status
            batch_record.success_count = success_count
            batch_record.fail_count = fail_count
            batch_record.completed_at = datetime.utcnow()
            await self.db.commit()
        except Exception as e:
            logger.error(f"更新批量执行记录失败: {e}")
            await self.db.rollback()

        return {
            "batch_id": batch_id,
            "total": total,
            "success_count": success_count,
            "fail_count": fail_count,
            "status": final_status,
            "results": results,
            "report_id": report_id,
        }

    async def run_project_all(
        self,
        project_id: str,
    ) -> Dict[str, Any]:
        """
        执行项目下所有任务

        Args:
            project_id: 项目ID

        Returns:
            dict: 批量执行结果
        """
        # 查询项目下所有可执行状态的任务
        result = await self.db.execute(
            select(PerformanceTask)
            .where(
                PerformanceTask.project_id == project_id,
                PerformanceTask.status.in_(["pending", "completed", "failed", "cancelled"]),
            )
            .order_by(PerformanceTask.created_at.asc())
        )
        tasks = list(result.scalars().all())

        if not tasks:
            raise ValueError("项目下没有可执行的任务")

        task_ids = [t.id for t in tasks]
        return await self.batch_run_tasks(task_ids, project_id=project_id)

    async def get_project_task_ids(self, project_id: str) -> List[str]:
        """
        获取项目下所有可执行状态的任务 ID 列表

        Args:
            project_id: 项目ID

        Returns:
            list: 任务ID列表
        """
        result = await self.db.execute(
            select(PerformanceTask)
            .where(
                PerformanceTask.project_id == project_id,
                PerformanceTask.status.in_(["pending", "completed", "failed", "cancelled"]),
            )
            .order_by(PerformanceTask.created_at.asc())
        )
        tasks = list(result.scalars().all())
        return [t.id for t in tasks]

    async def create_batch_and_start(
        self,
        task_ids: List[str],
        project_id: Optional[str] = None,
    ) -> Tuple[str, int]:
        """
        创建批量执行记录并在后台启动异步执行

        Args:
            task_ids: 任务ID列表
            project_id: 关联项目ID

        Returns:
            tuple: (batch_id, task_count)
        """
        batch_id = generate_id("batch")
        total = len(task_ids)

        # 持久化批量执行记录
        batch_record = PerformanceBatchExecution(
            id=batch_id,
            project_id=project_id,
            task_ids=json.dumps(task_ids, ensure_ascii=False),
            status="running",
            total_count=total,
            success_count=0,
            fail_count=0,
            started_at=datetime.utcnow(),
            created_at=datetime.utcnow(),
        )
        self.db.add(batch_record)

        # 将批次内所有任务状态重置为 pending（避免读到上一次执行的残留 completed/failed 状态）
        for task_id in task_ids:
            task = await self.db.get(PerformanceTask, task_id)
            if task and task.status in ("completed", "failed", "cancelled", "pending"):
                task.status = "pending"

        await self.db.commit()

        # 在后台异步执行
        task = asyncio.create_task(
            self._run_batch_background(batch_id, task_ids, project_id)
        )
        self._background_tasks.add(task)
        task.add_done_callback(self._background_tasks.discard)

        return batch_id, total

    async def _run_batch_background(
        self,
        batch_id: str,
        task_ids: List[str],
        project_id: Optional[str] = None,
    ) -> None:
        """
        后台批量执行（使用独立数据库会话）

        Args:
            batch_id: 批次ID
            task_ids: 任务ID列表
            project_id: 关联项目ID
        """
        from app.database import async_session_maker

        async with async_session_maker() as db:
            try:
                service = PerformanceService(db)
                await service.batch_run_tasks(
                    task_ids=task_ids,
                    project_id=project_id,
                    batch_id=batch_id,
                )
            except Exception as e:
                logger.error(f"后台批量执行失败 [{batch_id}]: {e}")
                try:
                    batch = await db.get(PerformanceBatchExecution, batch_id)
                    if batch:
                        batch.status = "failed"
                        await db.commit()
                except Exception:
                    pass

    async def get_batch_details(self, batch_id: str) -> Optional[Dict[str, Any]]:
        """
        获取批量执行详情

        Args:
            batch_id: 批次ID

        Returns:
            dict: 批量执行详情（含任务状态列表）
        """
        batch = await self.db.get(PerformanceBatchExecution, batch_id)
        if not batch:
            return None

        # 解析任务ID列表
        task_ids = json.loads(batch.task_ids) if batch.task_ids else []

        # 查询各任务状态
        tasks = []
        for task_id in task_ids:
            task = await self.db.get(PerformanceTask, task_id)
            if task:
                tasks.append({
                    "task_id": task.id,
                    "name": task.name,
                    "status": task.status,
                    "actual_tps": task.actual_tps,
                    "avg_response_time": task.avg_response_time,
                    "error_rate": task.error_rate,
                    "actual_concurrency": task.concurrency,
                    "duration_actual": task.duration_actual,
                })
            else:
                tasks.append({
                    "task_id": task_id,
                    "name": task_id,
                    "status": "unknown",
                    "actual_tps": None,
                    "avg_response_time": None,
                    "error_rate": None,
                    "actual_concurrency": None,
                    "duration_actual": None,
                })

        # 查询项目名称
        project_name = ""
        if batch.project_id:
            project = await self.db.get(PerformanceProject, batch.project_id)
            if project:
                project_name = project.name

        return {
            "batch_id": batch.id,
            "project_id": batch.project_id,
            "project_name": project_name,
            "status": batch.status,
            "total_count": batch.total_count,
            "success_count": batch.success_count,
            "fail_count": batch.fail_count,
            "started_at": batch.started_at,
            "completed_at": batch.completed_at,
            "tasks": tasks,
        }