TanWenyan/leaudit-platform-frontend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
22ef99754cba6f5a45e06aa3ed7cf015c6ca3fb9
leaudit-platform-frontend/auth_doc/系统概览.md
T
LiangShiyong 3d1dbb3f97 all in
2025-12-05 00:09:32 +08:00

30 KiB
Raw Blame History

统计分析模块 API 文档

路径: app/routes/v3/statistics.py 路由前缀: /admin/v3/statistics 功能: 提供评查数据的统计分析，包括高频错误评查点和高风险用户排行

📋 目录

1. 模块概述

统计分析模块专为管理员提供数据洞察能力，帮助发现系统中的薄弱环节和高风险用户，从而进行针对性的培训和改进。

核心特性

  1. 仅管理员访问

    • 所有统计接口仅限 admin 角色访问
    • 普通用户无法查看统计数据

  2. 地市隔离

    • 管理员只能查看自己地市的统计数据
    • 从 JWT Token 中提取 area 字段进行数据过滤
    • 确保数据安全和隐私

  3. 时间范围过滤

    • 支持按时间范围统计（start_date - end_date）
    • 不提供时间参数时统计全部历史数据

  4. Top N 排行

    • 支持自定义返回数量（limit 参数）
    • 高频错误评查点：最多返回 50 条
    • 高风险用户：最多返回 20 条

应用场景

统计类型 应用场景 改进措施
高频错误评查点 发现常见合规问题 加强培训、优化模板
高风险用户 识别需要培训的用户 针对性培训、质量把控

2. 接口列表

序号 HTTP方法 路径 功能 认证要求 权限要求
1 GET /top-error-points 获取高频错误评查点 Top N JWT必需 admin
2 GET /top-risk-users 获取高风险用户 Top N JWT必需 admin

3. 详细接口说明

3.1 获取高频错误评查点

统计最常出错的评查点，按累计出错用户数排序。

接口信息

  • HTTP方法: GET
  • 路径: /admin/v3/statistics/top-error-points
  • 认证: JWT Token（必需）
  • 权限: admin（管理员）

请求参数

Headers:

Authorization: Bearer <JWT_TOKEN>

Query 参数:

参数 类型 必填 默认值 范围 说明
limit int 否 10 1-50 返回Top N条记录
start_date str 否 null - 开始时间（格式：YYYY-MM-DD）
end_date str 否 null - 结束时间（格式：YYYY-MM-DD）

请求示例

示例 1: 获取 Top 10 高频错误评查点（全部历史）

cURL:

curl -X GET "http://localhost:8000/admin/v3/statistics/top-error-points?limit=10" \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

Python:

import requests

url = "http://localhost:8000/admin/v3/statistics/top-error-points"
headers = {
    "Authorization": f"Bearer {jwt_token}"
}
params = {
    "limit": 10
}

response = requests.get(url, headers=headers, params=params)
data = response.json()

print(f"高频错误评查点 Top {data['total']}:")
for item in data['items']:
    print(f"  {item['rank']}. {item['point_name']} - {item['error_user_count']} 人出错")

JavaScript:

const params = new URLSearchParams({
  limit: '10'
});

const response = await fetch(
  `http://localhost:8000/admin/v3/statistics/top-error-points?${params}`,
  {
    headers: {
      'Authorization': `Bearer ${jwtToken}`
    }
  }
);
const data = await response.json();

console.log(`高频错误评查点 Top ${data.total}:`);
data.items.forEach(item => {
  console.log(`  ${item.rank}. ${item.point_name} - ${item.error_user_count} 人出错`);
});

示例 2: 获取 2025年11月的 Top 20 高频错误评查点

cURL:

curl -X GET "http://localhost:8000/admin/v3/statistics/top-error-points?limit=20&start_date=2025-11-01&end_date=2025-11-30" \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

Python:

import requests

url = "http://localhost:8000/admin/v3/statistics/top-error-points"
headers = {
    "Authorization": f"Bearer {jwt_token}"
}
params = {
    "limit": 20,
    "start_date": "2025-11-01",
    "end_date": "2025-11-30"
}

response = requests.get(url, headers=headers, params=params)
data = response.json()

print(f"2025年11月高频错误评查点 Top {data['total']}:")
for item in data['items']:
    print(f"  {item['rank']}. {item['point_name']} (ID: {item['evaluation_point_id']})")
    print(f"      累计出错用户数: {item['error_user_count']}")

JavaScript:

const params = new URLSearchParams({
  limit: '20',
  start_date: '2025-11-01',
  end_date: '2025-11-30'
});

const response = await fetch(
  `http://localhost:8000/admin/v3/statistics/top-error-points?${params}`,
  {
    headers: {
      'Authorization': `Bearer ${jwtToken}`
    }
  }
);
const data = await response.json();
console.log(`2025年11月高频错误评查点 Top ${data.total}`);

响应示例

成功响应 (200 OK):

{
  "total": 10,
  "items": [
    {
      "rank": 1,
      "evaluation_point_id": 45,
      "point_name": "管辖地点明确性",
      "error_user_count": 38
    },
    {
      "rank": 2,
      "evaluation_point_id": 67,
      "point_name": "履约保证金比例合规性",
      "error_user_count": 25
    },
    {
      "rank": 3,
      "evaluation_point_id": 23,
      "point_name": "付款条款完整性",
      "error_user_count": 19
    },
    {
      "rank": 4,
      "evaluation_point_id": 89,
      "point_name": "违约责任条款明确性",
      "error_user_count": 15
    },
    {
      "rank": 5,
      "evaluation_point_id": 12,
      "point_name": "工程质量标准规范性",
      "error_user_count": 12
    }
  ]
}

失败响应 - 权限不足 (403 Forbidden):

{
  "detail": "此接口仅限管理员访问"
}

失败响应 - 缺少地市信息 (400 Bad Request):

{
  "detail": "用户缺少地市信息，无法进行统计查询"
}

失败响应 - 时间格式错误 (400 Bad Request):

{
  "detail": "开始时间格式错误，应为 YYYY-MM-DD，实际为: 2025/11/01"
}

业务逻辑

1. 验证JWT Token
2. 提取用户信息（user_id, user_role, area）
3. ✅ 权限检查
   ├─ 验证user_role == "admin"
   ├─ 验证current_user.area存在
   └─ 不满足则返回403或400
4. 📅 解析时间参数
   ├─ start_date: 转换为datetime对象
   ├─ end_date: 转换为datetime对象，并调整到23:59:59
   └─ 格式错误则返回400
5. 🗂️ 调用服务层
   ├─ StatisticsService.get_top_error_evaluation_points()
   ├─ 传入参数: area, limit, start_date, end_date
   └─ 查询evaluation_results表
6. 📊 统计逻辑
   ├─ 筛选条件:
   │   ├─ result = 'false' (评查失败)
   │   ├─ documents.ou_id = current_user.area (地市隔离)
   │   └─ created_at BETWEEN start_date AND end_date (时间范围)
   ├─ 分组统计:
   │   ├─ GROUP BY evaluation_point_id
   │   └─ COUNT(DISTINCT user_id) AS error_user_count (去重用户)
   └─ 排序: ORDER BY error_user_count DESC
7. 📤 构建响应
   ├─ 添加排名（rank）
   └─ 返回TopErrorPointsResponse

统计说明

去重规则:

  • 同一用户在多个文档中触发同一评查点错误，只计数一次
  • 统计的是 "有多少人犯了这个错误"，而不是 "这个错误发生了多少次"

示例:

用户A的文档1: 管辖地点明确性 - 失败
用户A的文档2: 管辖地点明确性 - 失败
用户B的文档1: 管辖地点明确性 - 失败

结果: 管辖地点明确性的error_user_count = 2（只计数A和B两个用户）

3.2 获取高风险用户

统计最容易出错的用户，按累计出错数排序。

接口信息

  • HTTP方法: GET
  • 路径: /admin/v3/statistics/top-risk-users
  • 认证: JWT Token（必需）
  • 权限: admin（管理员）

请求参数

Headers:

Authorization: Bearer <JWT_TOKEN>

Query 参数:

参数 类型 必填 默认值 范围 说明
limit int 否 5 1-20 返回Top N条记录
start_date str 否 null - 开始时间（格式：YYYY-MM-DD）
end_date str 否 null - 结束时间（格式：YYYY-MM-DD）

请求示例

示例 1: 获取 Top 5 高风险用户（全部历史）

cURL:

curl -X GET "http://localhost:8000/admin/v3/statistics/top-risk-users?limit=5" \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

Python:

import requests

url = "http://localhost:8000/admin/v3/statistics/top-risk-users"
headers = {
    "Authorization": f"Bearer {jwt_token}"
}
params = {
    "limit": 5
}

response = requests.get(url, headers=headers, params=params)
data = response.json()

print(f"高风险用户 Top {data['total']}:")
for item in data['items']:
    print(f"  {item['rank']}. {item['user_name']} ({item['department']})")
    print(f"      累计出错: {item['total_errors']} 次")
    print(f"      单文档平均出错: {item['avg_errors_per_doc']:.2f} 次")

JavaScript:

const params = new URLSearchParams({
  limit: '5'
});

const response = await fetch(
  `http://localhost:8000/admin/v3/statistics/top-risk-users?${params}`,
  {
    headers: {
      'Authorization': `Bearer ${jwtToken}`
    }
  }
);
const data = await response.json();

console.log(`高风险用户 Top ${data.total}:`);
data.items.forEach(item => {
  console.log(`  ${item.rank}. ${item.user_name} (${item.department})`);
  console.log(`      累计出错: ${item.total_errors} 次`);
  console.log(`      单文档平均出错: ${item.avg_errors_per_doc.toFixed(2)} 次`);
});

示例 2: 获取 2025年11月的 Top 10 高风险用户

cURL:

curl -X GET "http://localhost:8000/admin/v3/statistics/top-risk-users?limit=10&start_date=2025-11-01&end_date=2025-11-30" \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

Python:

import requests

url = "http://localhost:8000/admin/v3/statistics/top-risk-users"
headers = {
    "Authorization": f"Bearer {jwt_token}"
}
params = {
    "limit": 10,
    "start_date": "2025-11-01",
    "end_date": "2025-11-30"
}

response = requests.get(url, headers=headers, params=params)
data = response.json()

print(f"2025年11月高风险用户 Top {data['total']}:")
for item in data['items']:
    print(f"  {item['rank']}. {item['user_name']} (ID: {item['user_id']})")
    print(f"      部门: {item['department']}")
    print(f"      累计出错: {item['total_errors']} 次")
    print(f"      单文档平均出错: {item['avg_errors_per_doc']:.2f} 次")

响应示例

成功响应 (200 OK):

{
  "total": 5,
  "items": [
    {
      "rank": 1,
      "user_id": 123,
      "user_name": "张三",
      "department": "工程部",
      "total_errors": 45,
      "avg_errors_per_doc": 3.75
    },
    {
      "rank": 2,
      "user_id": 456,
      "user_name": "李四",
      "department": "采购部",
      "total_errors": 38,
      "avg_errors_per_doc": 3.17
    },
    {
      "rank": 3,
      "user_id": 789,
      "user_name": "王五",
      "department": "法务部",
      "total_errors": 32,
      "avg_errors_per_doc": 2.67
    },
    {
      "rank": 4,
      "user_id": 234,
      "user_name": "赵六",
      "department": "工程部",
      "total_errors": 28,
      "avg_errors_per_doc": 2.33
    },
    {
      "rank": 5,
      "user_id": 567,
      "user_name": "孙七",
      "department": "采购部",
      "total_errors": 25,
      "avg_errors_per_doc": 2.08
    }
  ]
}

失败响应 - 权限不足 (403 Forbidden):

{
  "detail": "此接口仅限管理员访问"
}

业务逻辑

1. 验证JWT Token
2. 提取用户信息（user_id, user_role, area）
3. ✅ 权限检查
   ├─ 验证user_role == "admin"
   ├─ 验证current_user.area存在
   └─ 不满足则返回403或400
4. 📅 解析时间参数（同高频错误评查点）
5. 🗂️ 调用服务层
   ├─ StatisticsService.get_top_risk_users()
   ├─ 传入参数: area, limit, start_date, end_date
   └─ 查询evaluation_results + documents + users表
6. 📊 统计逻辑
   ├─ 筛选条件:
   │   ├─ result = 'false' (评查失败)
   │   ├─ documents.ou_id = current_user.area (地市隔离)
   │   └─ created_at BETWEEN start_date AND end_date (时间范围)
   ├─ 分组统计:
   │   ├─ GROUP BY documents.user_id
   │   ├─ COUNT(*) AS total_errors (累计出错数)
   │   └─ COUNT(DISTINCT document_id) AS doc_count (文档总数)
   ├─ 计算平均值:
   │   └─ avg_errors_per_doc = total_errors / doc_count
   └─ 排序: ORDER BY total_errors DESC
7. 📤 构建响应
   ├─ 添加排名（rank）
   ├─ 联表查询用户名和部门
   └─ 返回TopRiskUsersResponse

统计说明

计算公式:

累计出错数 = 该用户所有文档的错误评查点总数
单文档平均出错数 = 累计出错数 / 文档总数

示例:

用户A:
  - 文档1: 3个错误评查点
  - 文档2: 5个错误评查点
  - 文档3: 2个错误评查点

结果:
  - total_errors = 10
  - doc_count = 3
  - avg_errors_per_doc = 10 / 3 = 3.33

4. 权限控制

4.1 权限验证函数

代码位置: app/routes/v3/statistics.py:24-55

def require_admin(current_user: User = Depends(verify_token)) -> User:
    """
    权限验证：要求管理员角色

    Args:
        current_user: 当前登录用户

    Returns:
        User: 验证通过的管理员用户

    Raises:
        HTTPException: 403 如果用户不是管理员
    """
    if current_user.user_role != "admin":
        logger.warning(
            f"用户 {current_user.username} (角色:{current_user.user_role}) "
            f"尝试访问管理员专属统计接口"
        )
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="此接口仅限管理员访问"
        )

    # 检查用户是否有地市信息
    if not hasattr(current_user, 'area') or not current_user.area:
        logger.error(f"管理员用户 {current_user.username} 缺少地市(area)信息")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户缺少地市信息，无法进行统计查询"
        )

    return current_user

4.2 权限检查流程

1. JWT Token 验证
   ├─ 验证Token签名
   ├─ 验证Token有效期
   └─ 提取用户信息（user_id, username, user_role, area）
2. 角色验证
   ├─ 检查user_role == "admin"
   └─ 不是管理员则返回403
3. 地市信息验证
   ├─ 检查current_user.area存在
   └─ 不存在则返回400

4.3 权限级别

用户角色 访问权限 可见数据范围
admin ✅ 允许访问 只能查看自己地市的统计数据
provincial_admin ❌ 禁止访问 -
user ❌ 禁止访问 -
guest ❌ 禁止访问 -

注意: 即使是省级管理员（provincial_admin）也无法访问统计接口，因为统计接口专为地市级管理员设计。

5. 地市隔离机制

5.1 地市隔离原理

数据过滤逻辑:

-- 高频错误评查点统计
SELECT
  ep.id AS evaluation_point_id,
  ep.name AS point_name,
  COUNT(DISTINCT d.user_id) AS error_user_count
FROM evaluation_results er
JOIN evaluation_points ep ON er.evaluation_point_id = ep.id
JOIN documents d ON er.document_id = d.id
WHERE er.evaluated_results->>'result' = 'false'
  AND d.ou_id = ${current_user.area}  -- 地市隔离
  AND er.created_at BETWEEN ${start_date} AND ${end_date}
GROUP BY ep.id, ep.name
ORDER BY error_user_count DESC
LIMIT ${limit};

关键字段:

  • documents.ou_id: 文档所属地市（如: "梅州", "云浮", "揭阳", "潮州"）
  • current_user.area: 当前管理员所属地市

5.2 地市映射

地市代码 地市名称 说明
梅州 梅州市 地市级
云浮 云浮市 地市级
揭阳 揭阳市 地市级
潮州 潮州市 地市级
省级 广东省 省级（暂不支持统计接口）

5.3 地市隔离示例

场景:

  • 梅州管理员登录
  • JWT Token中 area = "梅州"

查询结果:

  • ✅ 包含：梅州市用户的所有文档评查数据
  • ❌ 不包含：云浮、揭阳、潮州的文档评查数据

效果:

梅州管理员查询高频错误评查点:
  1. 管辖地点明确性 - 38人出错（都是梅州市用户）
  2. 履约保证金比例 - 25人出错（都是梅州市用户）
  ...

云浮管理员查询高频错误评查点:
  1. 付款条款完整性 - 42人出错（都是云浮市用户）
  2. 工程质量标准 - 30人出错（都是云浮市用户）
  ...

6. 数据结构

6.1 高频错误评查点

interface TopErrorPointItem {
  rank: number;                    // 排名（1, 2, 3...）
  evaluation_point_id: number;     // 评查点ID
  point_name: string;              // 评查点名称
  error_user_count: number;        // 累计出错用户数（去重）
}

interface TopErrorPointsResponse {
  total: number;                          // 返回记录数
  items: TopErrorPointItem[];             // 评查点列表
}

6.2 高风险用户

interface TopRiskUserItem {
  rank: number;                    // 排名（1, 2, 3...）
  user_id: number;                 // 用户ID
  user_name: string;               // 用户真实姓名
  department: string;              // 所在部门
  total_errors: number;            // 累计出错数
  avg_errors_per_doc: number;      // 单文档平均出错数（保留2位小数）
}

interface TopRiskUsersResponse {
  total: number;                          // 返回记录数
  items: TopRiskUserItem[];               // 用户列表
}

7. 使用示例

7.1 完整的统计看板示例

Vue 3 + Element Plus + ECharts:

<template>
  <div class="statistics-dashboard">
    <el-row :gutter="20">
      <!-- 时间范围选择器 -->
      <el-col :span="24">
        <el-card>
          <el-date-picker
            v-model="dateRange"
            type="daterange"
            range-separator="至"
            start-placeholder="开始日期"
            end-placeholder="结束日期"
            value-format="YYYY-MM-DD"
            @change="fetchAllStatistics"
          />
          <el-button type="primary" style="margin-left: 10px;" @click="resetDateRange">
            查看全部历史
          </el-button>
        </el-card>
      </el-col>

      <!-- 高频错误评查点 -->
      <el-col :span="12">
        <el-card>
          <template #header>
            <div class="card-header">
              <span>高频错误评查点 Top 10</span>
              <el-tag type="danger">重点关注</el-tag>
            </div>
          </template>

          <div ref="errorPointsChart" style="height: 400px;"></div>

          <el-table :data="errorPoints" style="margin-top: 20px;">
            <el-table-column prop="rank" label="排名" width="80" />
            <el-table-column prop="point_name" label="评查点名称" />
            <el-table-column prop="error_user_count" label="出错人数" width="120">
              <template #default="{ row }">
                <el-tag type="danger">{{ row.error_user_count }} 人</el-tag>
              </template>
            </el-table-column>
          </el-table>
        </el-card>
      </el-col>

      <!-- 高风险用户 -->
      <el-col :span="12">
        <el-card>
          <template #header>
            <div class="card-header">
              <span>高风险用户 Top 10</span>
              <el-tag type="warning">需要培训</el-tag>
            </div>
          </template>

          <div ref="riskUsersChart" style="height: 400px;"></div>

          <el-table :data="riskUsers" style="margin-top: 20px;">
            <el-table-column prop="rank" label="排名" width="80" />
            <el-table-column prop="user_name" label="用户" />
            <el-table-column prop="department" label="部门" />
            <el-table-column prop="total_errors" label="累计出错" width="120">
              <template #default="{ row }">
                <el-tag type="warning">{{ row.total_errors }} 次</el-tag>
              </template>
            </el-table-column>
            <el-table-column prop="avg_errors_per_doc" label="平均出错" width="120">
              <template #default="{ row }">
                {{ row.avg_errors_per_doc.toFixed(2) }}
              </template>
            </el-table-column>
          </el-table>
        </el-card>
      </el-col>
    </el-row>
  </div>
</template>

<script setup>
import { ref, onMounted, nextTick } from 'vue';
import axios from 'axios';
import * as echarts from 'echarts';
import { ElMessage } from 'element-plus';

const dateRange = ref(null);
const errorPoints = ref([]);
const riskUsers = ref([]);
const errorPointsChart = ref(null);
const riskUsersChart = ref(null);

let errorPointsChartInstance = null;
let riskUsersChartInstance = null;

async function fetchErrorPoints() {
  try {
    const params = {
      limit: 10
    };

    if (dateRange.value && dateRange.value.length === 2) {
      params.start_date = dateRange.value[0];
      params.end_date = dateRange.value[1];
    }

    const { data } = await axios.get('/admin/v3/statistics/top-error-points', { params });
    errorPoints.value = data.items;

    // 更新图表
    renderErrorPointsChart(data.items);
  } catch (error) {
    ElMessage.error('获取高频错误评查点失败');
    console.error(error);
  }
}

async function fetchRiskUsers() {
  try {
    const params = {
      limit: 10
    };

    if (dateRange.value && dateRange.value.length === 2) {
      params.start_date = dateRange.value[0];
      params.end_date = dateRange.value[1];
    }

    const { data } = await axios.get('/admin/v3/statistics/top-risk-users', { params });
    riskUsers.value = data.items;

    // 更新图表
    renderRiskUsersChart(data.items);
  } catch (error) {
    ElMessage.error('获取高风险用户失败');
    console.error(error);
  }
}

async function fetchAllStatistics() {
  await Promise.all([
    fetchErrorPoints(),
    fetchRiskUsers()
  ]);
}

function resetDateRange() {
  dateRange.value = null;
  fetchAllStatistics();
}

function renderErrorPointsChart(items) {
  if (!errorPointsChartInstance) {
    errorPointsChartInstance = echarts.init(errorPointsChart.value);
  }

  const option = {
    title: {
      text: '高频错误评查点统计'
    },
    tooltip: {
      trigger: 'axis',
      axisPointer: {
        type: 'shadow'
      }
    },
    xAxis: {
      type: 'value',
      name: '出错人数'
    },
    yAxis: {
      type: 'category',
      data: items.map(item => item.point_name).reverse(),
      axisLabel: {
        formatter: (value) => {
          return value.length > 10 ? value.substring(0, 10) + '...' : value;
        }
      }
    },
    series: [{
      name: '出错人数',
      type: 'bar',
      data: items.map(item => item.error_user_count).reverse(),
      itemStyle: {
        color: '#f56c6c'
      },
      label: {
        show: true,
        position: 'right'
      }
    }]
  };

  errorPointsChartInstance.setOption(option);
}

function renderRiskUsersChart(items) {
  if (!riskUsersChartInstance) {
    riskUsersChartInstance = echarts.init(riskUsersChart.value);
  }

  const option = {
    title: {
      text: '高风险用户统计'
    },
    tooltip: {
      trigger: 'axis',
      axisPointer: {
        type: 'shadow'
      }
    },
    legend: {
      data: ['累计出错数', '平均出错数']
    },
    xAxis: {
      type: 'category',
      data: items.map(item => item.user_name)
    },
    yAxis: [
      {
        type: 'value',
        name: '累计出错数'
      },
      {
        type: 'value',
        name: '平均出错数'
      }
    ],
    series: [
      {
        name: '累计出错数',
        type: 'bar',
        data: items.map(item => item.total_errors),
        itemStyle: {
          color: '#e6a23c'
        }
      },
      {
        name: '平均出错数',
        type: 'line',
        yAxisIndex: 1,
        data: items.map(item => item.avg_errors_per_doc),
        itemStyle: {
          color: '#f56c6c'
        }
      }
    ]
  };

  riskUsersChartInstance.setOption(option);
}

onMounted(() => {
  nextTick(() => {
    fetchAllStatistics();
  });

  // 响应式调整图表大小
  window.addEventListener('resize', () => {
    errorPointsChartInstance?.resize();
    riskUsersChartInstance?.resize();
  });
});
</script>

<style scoped>
.card-header {
  display: flex;
  justify-content: space-between;
  align-items: center;
}
</style>

8. 最佳实践

8.1 定期统计分析

建议频率:

  • 周报: 每周一查看上周统计数据
  • 月报: 每月初查看上月统计数据
  • 季度报: 每季度末查看季度统计数据

Python 定时任务示例:

import requests
from datetime import datetime, timedelta
import schedule
import time

def weekly_statistics_report(jwt_token: str):
    """
    每周统计报告生成
    """
    # 计算上周的日期范围
    today = datetime.now()
    last_monday = today - timedelta(days=today.weekday() + 7)
    last_sunday = last_monday + timedelta(days=6)

    start_date = last_monday.strftime('%Y-%m-%d')
    end_date = last_sunday.strftime('%Y-%m-%d')

    print(f"\n=== 上周统计报告 ({start_date} ~ {end_date}) ===\n")

    # 获取高频错误评查点
    error_points_resp = requests.get(
        "http://localhost:8000/admin/v3/statistics/top-error-points",
        headers={"Authorization": f"Bearer {jwt_token}"},
        params={
            "limit": 10,
            "start_date": start_date,
            "end_date": end_date
        }
    )

    if error_points_resp.status_code == 200:
        error_points = error_points_resp.json()['items']
        print("高频错误评查点 Top 10:")
        for item in error_points:
            print(f"  {item['rank']}. {item['point_name']} - {item['error_user_count']} 人")

    # 获取高风险用户
    risk_users_resp = requests.get(
        "http://localhost:8000/admin/v3/statistics/top-risk-users",
        headers={"Authorization": f"Bearer {jwt_token}"},
        params={
            "limit": 5,
            "start_date": start_date,
            "end_date": end_date
        }
    )

    if risk_users_resp.status_code == 200:
        risk_users = risk_users_resp.json()['items']
        print("\n高风险用户 Top 5:")
        for item in risk_users:
            print(f"  {item['rank']}. {item['user_name']} ({item['department']}) - {item['total_errors']} 次")

    print("\n==========================================\n")


# 设置定时任务：每周一早上9点执行
schedule.every().monday.at("09:00").do(weekly_statistics_report, jwt_token="your_jwt_token")

while True:
    schedule.run_pending()
    time.sleep(60)

8.2 改进措施建议

基于高频错误评查点:

出错人数 改进措施 优先级
> 30人 全员培训、修订模板、优化流程 🔴 高
20-30人 针对性培训、发布指导文档 🟡 中
10-20人 内部分享、案例学习 🟢 低
< 10人 个别指导 -

基于高风险用户:

累计出错数 改进措施 优先级
> 40次 一对一培训、质量把控 🔴 高
30-40次 小组培训、定期复查 🟡 中
20-30次 案例分享、自我学习 🟢 低
< 20次 日常指导 -

8.3 统计数据导出

Python导出Excel示例:

import requests
import pandas as pd
from datetime import datetime

def export_statistics_to_excel(jwt_token: str, output_file: str):
    """
    导出统计数据到Excel
    """
    # 获取高频错误评查点
    error_points_resp = requests.get(
        "http://localhost:8000/admin/v3/statistics/top-error-points",
        headers={"Authorization": f"Bearer {jwt_token}"},
        params={"limit": 50}
    )
    error_points = error_points_resp.json()['items']

    # 获取高风险用户
    risk_users_resp = requests.get(
        "http://localhost:8000/admin/v3/statistics/top-risk-users",
        headers={"Authorization": f"Bearer {jwt_token}"},
        params={"limit": 20}
    )
    risk_users = risk_users_resp.json()['items']

    # 创建Excel文件
    with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
        # 高频错误评查点
        df_error_points = pd.DataFrame(error_points)
        df_error_points.to_excel(
            writer,
            sheet_name='高频错误评查点',
            index=False
        )

        # 高风险用户
        df_risk_users = pd.DataFrame(risk_users)
        df_risk_users.to_excel(
            writer,
            sheet_name='高风险用户',
            index=False
        )

    print(f"统计数据已导出到: {output_file}")


# 使用示例
export_statistics_to_excel(
    jwt_token="your_jwt_token",
    output_file=f"统计报告_{datetime.now().strftime('%Y%m%d')}.xlsx"
)

相关文档

文档版本: v1.0 最后更新: 2025-11-20 维护者: DocAuditAI 开发团队

Reference in New Issue View Git Blame Copy Permalink