┌─────────────────────────────────────────────────────────────┐
│                    🎨 报告展示层                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  实时仪表盘 │  │  详细报告   │  │  趋势分析   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    📊 数据处理层                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  数据聚合   │  │  统计计算   │  │  报告生成   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    🔧 数据采集层                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│  │  执行数据   │  │  性能数据   │  │  错误日志   │        │
│  └─────────────┘  └─────────────┘  └─────────────┘        │
├─────────────────────────────────────────────────────────────┤
│                    💾 数据存储层                            │
│        MySQL + Redis + InfluxDB(时序数据)                  │
└─────────────────────────────────────────────────────────────┘

技术选型：工具链的艺术

前端可视化：

ECharts：百度出品，图表类型丰富，性能优秀
Ant Design Charts：基于G2的React图表库，与Ant Design完美融合
D3.js：自定义图表的终极武器（复杂场景使用）

后端数据处理：

Pandas：数据处理的瑞士军刀
NumPy：数值计算的基石
SQLAlchemy：复杂查询的好帮手

报告生成：

Allure：专业的测试报告框架
ReportLab：PDF报告生成
Jinja2：模板引擎，灵活定制

📊 数据模型设计

报告数据模型：统计的基础

app/models/report.py：

from peewee import *
from .base import BaseModel
from .project import Project
from .testcase import TestCase
import json
from datetime import datetime, timedelta

class ReportTemplate(BaseModel):
    """报告模板模型"""
    
    name = CharField(max_length=100, verbose_name="模板名称")
    description = TextField(null=True, verbose_name="模板描述")
    template_type = CharField(
        max_length=20,
        choices=[
            ('dashboard', '仪表盘'),
            ('detailed', '详细报告'),
            ('summary', '汇总报告'),
            ('trend', '趋势报告')
        ],
        verbose_name="模板类型"
    )
    
    # 模板配置 - JSON格式存储图表配置
    config = TextField(verbose_name="模板配置")
    
    # 适用范围
    projects = CharField(max_length=500, null=True, verbose_name="适用项目")  # 逗号分隔的项目ID
    
    is_default = BooleanField(default=False, verbose_name="是否默认模板")
    
    class Meta:
        table_name = 'report_templates'
    
    def get_config_dict(self):
        """获取配置字典"""
        try:
            return json.loads(self.config)
        except (json.JSONDecodeError, TypeError):
            return {}

class TestStatistics(BaseModel):
    """测试统计数据模型 - 按天聚合的统计数据"""
    
    project = ForeignKeyField(Project, backref='statistics', verbose_name="所属项目")
    stat_date = DateField(verbose_name="统计日期")
    
    # 执行统计
    total_executions = IntegerField(default=0, verbose_name="总执行次数")
    success_executions = IntegerField(default=0, verbose_name="成功执行次数")
    failed_executions = IntegerField(default=0, verbose_name="失败执行次数")
    
    # 用例统计
    total_cases = IntegerField(default=0, verbose_name="总用例数")
    active_cases = IntegerField(default=0, verbose_name="活跃用例数")
    
    # 性能统计
    avg_response_time = FloatField(null=True, verbose_name="平均响应时间")
    max_response_time = FloatField(null=True, verbose_name="最大响应时间")
    min_response_time = FloatField(null=True, verbose_name="最小响应时间")
    
    # 错误统计
    error_count = IntegerField(default=0, verbose_name="错误数量")
    error_types = TextField(null=True, verbose_name="错误类型统计")  # JSON格式
    
    class Meta:
        table_name = 'test_statistics'
        indexes = (
            (('project', 'stat_date'), True),  # 项目和日期组合唯一
        )
    
    @property
    def success_rate(self):
        """成功率"""
        if self.total_executions == 0:
            return 0
        return round(self.success_executions / self.total_executions * 100, 2)
    
    def get_error_types_dict(self):
        """获取错误类型字典"""
        try:
            return json.loads(self.error_types) if self.error_types else {}
        except (json.JSONDecodeError, TypeError):
            return {}

class ReportSnapshot(BaseModel):
    """报告快照模型 - 保存生成的报告"""
    
    project = ForeignKeyField(Project, backref='report_snapshots', verbose_name="所属项目")
    template = ForeignKeyField(ReportTemplate, backref='snapshots', verbose_name="使用模板")
    
    name = CharField(max_length=200, verbose_name="报告名称")
    report_type = CharField(max_length=20, verbose_name="报告类型")
    
    # 报告内容
    content = TextField(verbose_name="报告内容")  # JSON格式存储图表数据
    file_path = CharField(max_length=500, null=True, verbose_name="文件路径")  # PDF等文件路径
    
    # 时间范围
    start_date = DateField(verbose_name="开始日期")
    end_date = DateField(verbose_name="结束日期")
    
    # 生成信息
    generated_by = CharField(max_length=50, verbose_name="生成人")
    generated_at = DateTimeField(default=datetime.now, verbose_name="生成时间")
    
    class Meta:
        table_name = 'report_snapshots'

📈 数据聚合服务

统计数据计算：让数字有意义

app/services/statistics_service.py：

from datetime import datetime, timedelta
from typing import Dict, List, Any
from app.models.report import TestStatistics
from app.models.execution import ExecutionRecord, ExecutionStep
from app.models.project import Project
from app.models.testcase import TestCase
from peewee import fn, JOIN
import json

class StatisticsService:
    """统计服务类"""
    
    @staticmethod
    def calculate_daily_statistics(project_id: int, date: datetime.date):
        """计算指定日期的统计数据"""
        
        # 查询当天的执行记录
        executions = ExecutionRecord.select().where(
            (ExecutionRecord.testcase.in_(
                TestCase.select().where(TestCase.project == project_id)
            )) &
            (fn.DATE(ExecutionRecord.created_at) == date)
        )
        
        # 基础统计
        total_executions = executions.count()
        success_executions = executions.where(ExecutionRecord.status == 'success').count()
        failed_executions = executions.where(ExecutionRecord.status == 'failed').count()
        
        # 用例统计
        total_cases = TestCase.select().where(TestCase.project == project_id).count()
        active_cases = TestCase.select().where(
            (TestCase.project == project_id) & 
            (TestCase.status == 'active')
        ).count()
        
        # 性能统计
        performance_stats = executions.where(
            ExecutionRecord.duration.is_null(False)
        ).aggregate(
            avg_duration=fn.AVG(ExecutionRecord.duration),
            max_duration=fn.MAX(ExecutionRecord.duration),
            min_duration=fn.MIN(ExecutionRecord.duration)
        )
        
        # 错误统计
        error_executions = executions.where(ExecutionRecord.status == 'failed')
        error_types = {}
        for execution in error_executions:
            if execution.error_message:
                # 简单的错误分类逻辑
                error_type = StatisticsService._classify_error(execution.error_message)
                error_types[error_type] = error_types.get(error_type, 0) + 1
        
        # 保存或更新统计数据
        stat, created = TestStatistics.get_or_create(
            project=project_id,
            stat_date=date,
            defaults={
                'total_executions': total_executions,
                'success_executions': success_executions,
                'failed_executions': failed_executions,
                'total_cases': total_cases,
                'active_cases': active_cases,
                'avg_response_time': performance_stats.get('avg_duration'),
                'max_response_time': performance_stats.get('max_duration'),
                'min_response_time': performance_stats.get('min_duration'),
                'error_count': len(error_types),
                'error_types': json.dumps(error_types, ensure_ascii=False)
            }
        )
        
        if not created:
            # 更新现有记录
            stat.total_executions = total_executions
            stat.success_executions = success_executions
            stat.failed_executions = failed_executions
            stat.total_cases = total_cases
            stat.active_cases = active_cases
            stat.avg_response_time = performance_stats.get('avg_duration')
            stat.max_response_time = performance_stats.get('max_duration')
            stat.min_response_time = performance_stats.get('min_duration')
            stat.error_count = len(error_types)
            stat.error_types = json.dumps(error_types, ensure_ascii=False)
            stat.save()
        
        return stat
    
    @staticmethod
    def _classify_error(error_message: str) -> str:
        """错误分类逻辑"""
        error_message = error_message.lower()
        
        if 'timeout' in error_message or 'time out' in error_message:
            return '超时错误'
        elif 'connection' in error_message:
            return '连接错误'
        elif 'assertion' in error_message or 'assert' in error_message:
            return '断言错误'
        elif '404' in error_message:
            return '资源不存在'
        elif '500' in error_message:
            return '服务器错误'
        elif 'json' in error_message:
            return 'JSON解析错误'
        else:
            return '其他错误'
    
    @staticmethod
    def get_trend_data(project_id: int, days: int = 30) -> Dict[str, Any]:
        """获取趋势数据"""
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=days-1)
        
        # 查询统计数据
        stats = TestStatistics.select().where(
            (TestStatistics.project == project_id) &
            (TestStatistics.stat_date >= start_date) &
            (TestStatistics.stat_date <= end_date)
        ).order_by(TestStatistics.stat_date)
        
        # 构建趋势数据
        dates = []
        success_rates = []
        execution_counts = []
        avg_response_times = []
        
        for stat in stats:
            dates.append(stat.stat_date.strftime('%Y-%m-%d'))
            success_rates.append(stat.success_rate)
            execution_counts.append(stat.total_executions)
            avg_response_times.append(stat.avg_response_time or 0)
        
        return {
            'dates': dates,
            'success_rates': success_rates,
            'execution_counts': execution_counts,
            'avg_response_times': avg_response_times
        }
    
    @staticmethod
    def get_project_summary(project_id: int) -> Dict[str, Any]:
        """获取项目汇总数据"""
        # 最近30天的统计
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=30)
        
        stats = TestStatistics.select().where(
            (TestStatistics.project == project_id) &
            (TestStatistics.stat_date >= start_date) &
            (TestStatistics.stat_date <= end_date)
        )
        
        # 汇总计算
        total_executions = sum(stat.total_executions for stat in stats)
        total_success = sum(stat.success_executions for stat in stats)
        total_failed = sum(stat.failed_executions for stat in stats)
        
        # 最新数据
        latest_stat = stats.order_by(TestStatistics.stat_date.desc()).first()
        
        # 错误类型统计
        error_types_summary = {}
        for stat in stats:
            error_types = stat.get_error_types_dict()
            for error_type, count in error_types.items():
                error_types_summary[error_type] = error_types_summary.get(error_type, 0) + count
        
        return {
            'summary': {
                'total_executions': total_executions,
                'success_rate': round(total_success / total_executions * 100, 2) if total_executions > 0 else 0,
                'total_cases': latest_stat.total_cases if latest_stat else 0,
                'active_cases': latest_stat.active_cases if latest_stat else 0,
                'avg_response_time': latest_stat.avg_response_time if latest_stat else 0
            },
            'error_distribution': error_types_summary,
            'trend_data': StatisticsService.get_trend_data(project_id, 30)
        }

🎨 前端可视化组件

仪表盘组件：数据的"颜值担当"

src/components/Charts/Dashboard.tsx：

import React, { useState, useEffect } from 'react';
import { Card, Row, Col, Statistic, Spin } from 'antd';
import {
  ArrowUpOutlined,
  ArrowDownOutlined,
  CheckCircleOutlined,
  CloseCircleOutlined,
  ClockCircleOutlined,
  FileTextOutlined
} from '@ant-design/icons';
import { Line, Pie, Column } from '@ant-design/plots';
import { reportApi } from '../../services/report';
import './Dashboard.less';

interface DashboardProps {
  projectId: number;
  timeRange?: number; // 天数
}

interface SummaryData {
  summary: {
    total_executions: number;
    success_rate: number;
    total_cases: number;
    active_cases: number;
    avg_response_time: number;
  };
  error_distribution: Record<string, number>;
  trend_data: {
    dates: string[];
    success_rates: number[];
    execution_counts: number[];
    avg_response_times: number[];
  };
}

const Dashboard: React.FC<DashboardProps> = ({ projectId, timeRange = 30 }) => {
  const [loading, setLoading] = useState(false);
  const [data, setData] = useState<SummaryData | null>(null);

  // 获取仪表盘数据
  const fetchDashboardData = async () => {
    setLoading(true);
    try {
      const response = await reportApi.getProjectSummary(projectId);
      setData(response.data.data);
    } catch (error) {
      console.error('获取仪表盘数据失败:', error);
    } finally {
      setLoading(false);
    }
  };

  useEffect(() => {
    if (projectId) {
      fetchDashboardData();
    }
  }, [projectId, timeRange]);

  if (loading) {
    return (
      <div className="dashboard-loading">
        <Spin size="large" />
      </div>
    );
  }

  if (!data) {
    return <div>暂无数据</div>;
  }

  const { summary, error_distribution, trend_data } = data;

  // 成功率趋势图配置
  const successRateConfig = {
    data: trend_data.dates.map((date, index) => ({
      date,
      value: trend_data.success_rates[index],
    })),
    xField: 'date',
    yField: 'value',
    smooth: true,
    color: '#1890ff',
    point: {
      size: 3,
      shape: 'circle',
    },
    yAxis: {
      min: 0,
      max: 100,
    },
    annotations: [
      {
        type: 'line',
        start: ['min', 95],
        end: ['max', 95],
        style: {
          stroke: '#52c41a',
          lineDash: [4, 4],
        },
      },
    ],
  };

  // 执行次数柱状图配置
  const executionCountConfig = {
    data: trend_data.dates.map((date, index) => ({
      date,
      value: trend_data.execution_counts[index],
    })),
    xField: 'date',
    yField: 'value',
    color: '#52c41a',
    columnWidthRatio: 0.6,
  };

  // 错误分布饼图配置
  const errorDistributionConfig = {
    data: Object.entries(error_distribution).map(([type, count]) => ({
      type,
      value: count,
    })),
    angleField: 'value',
    colorField: 'type',
    radius: 0.8,
    label: {
      type: 'outer',
      content: '{name} {percentage}',
    },
    interactions: [
      {
        type: 'element-active',
      },
    ],
  };

  return (
    <div className="dashboard">
      {/* 核心指标卡片 */}
      <Row gutter={16} style={{ marginBottom: 16 }}>
        <Col span={6}>
          <Card>
            <Statistic
              title="总执行次数"
              value={summary.total_executions}
              prefix={<FileTextOutlined />}
              valueStyle={{ color: '#1890ff' }}
            />
          </Card>
        </Col>
        <Col span={6}>
          <Card>
            <Statistic
              title="成功率"
              value={summary.success_rate}
              precision={2}
              suffix="%"
              prefix={
                summary.success_rate >= 95 ? (
                  <ArrowUpOutlined style={{ color: '#3f8600' }} />
                ) : (
                  <ArrowDownOutlined style={{ color: '#cf1322' }} />
                )
              }
              valueStyle={{
                color: summary.success_rate >= 95 ? '#3f8600' : '#cf1322',
              }}
            />
          </Card>
        </Col>
        <Col span={6}>
          <Card>
            <Statistic
              title="活跃用例"
              value={summary.active_cases}
              suffix={`/ ${summary.total_cases}`}
              prefix={<CheckCircleOutlined />}
              valueStyle={{ color: '#52c41a' }}
            />
          </Card>
        </Col>
        <Col span={6}>
          <Card>
            <Statistic
              title="平均响应时间"
              value={summary.avg_response_time}
              precision={0}
              suffix="ms"
              prefix={<ClockCircleOutlined />}
              valueStyle={{
                color: summary.avg_response_time > 2000 ? '#cf1322' : '#3f8600',
              }}
            />
          </Card>
        </Col>
      </Row>

      {/* 趋势图表 */}
      <Row gutter={16} style={{ marginBottom: 16 }}>
        <Col span={12}>
          <Card title="成功率趋势" size="small">
            <Line {...successRateConfig} height={300} />
          </Card>
        </Col>
        <Col span={12}>
          <Card title="执行次数趋势" size="small">
            <Column {...executionCountConfig} height={300} />
          </Card>
        </Col>
      </Row>

      {/* 错误分析 */}
      <Row gutter={16}>
        <Col span={12}>
          <Card title="错误类型分布" size="small">
            {Object.keys(error_distribution).length > 0 ? (
              <Pie {...errorDistributionConfig} height={300} />
            ) : (
              <div style={{ textAlign: 'center', padding: '50px 0' }}>
                <CheckCircleOutlined style={{ fontSize: 48, color: '#52c41a' }} />
                <p style={{ marginTop: 16 }}>暂无错误，测试质量良好！</p>
              </div>
            )}
          </Card>
        </Col>
        <Col span={12}>
          <Card title="质量评估" size="small">
            <div className="quality-assessment">
              <div className="quality-item">
                <span className="label">整体质量:</span>
                <span className={`value ${summary.success_rate >= 95 ? 'excellent' : summary.success_rate >= 85 ? 'good' : 'poor'}`}>
                  {summary.success_rate >= 95 ? '优秀' : summary.success_rate >= 85 ? '良好' : '需改进'}
                </span>
              </div>
              <div className="quality-item">
                <span className="label">性能表现:</span>
                <span className={`value ${summary.avg_response_time <= 1000 ? 'excellent' : summary.avg_response_time <= 2000 ? 'good' : 'poor'}`}>
                  {summary.avg_response_time <= 1000 ? '优秀' : summary.avg_response_time <= 2000 ? '良好' : '需优化'}
                </span>
              </div>
              <div className="quality-item">
                <span className="label">用例覆盖:</span>
                <span className={`value ${(summary.active_cases / summary.total_cases) >= 0.8 ? 'excellent' : 'good'}`}>
                  {Math.round((summary.active_cases / summary.total_cases) * 100)}%
                </span>
              </div>
            </div>
          </Card>
        </Col>
      </Row>
    </div>
  );
};

export default Dashboard;

🎯 下一步预告

今天我们打造了一个强大的测试报告与数据可视化系统，实现了：

完整的数据统计模型
智能的数据聚合服务
美观的可视化组件
实时的仪表盘展示

下一篇我们将学习定时任务与异步处理机制，让测试平台能够自动化运行，真正实现"无人值守"的测试执行。毕竟，自动化的最高境界就是让机器自己跑测试！

💡 数据可视化小贴士：好的图表就像好的故事，要有起承转合。记住，图表的目的是让数据说话，不是炫技。简洁明了比花里胡哨更重要！