企业级软件研发团队绩效考核系统开发（持续更新 Day 3）
·
作者：呱牛
发布日期：2026年3月26日
标签：FastAPI、绩效考核、指标计算、规则引擎、Python
🔥 今日亮点
2026年3月26日 - 指标得分计算引擎开发完成
-
✅ 新增指标得分计算引擎：实现基于规则的自动分值计算
-
✅ 支持多种计算单位：灵活支持"个"、"天"等多种计量单位
-
✅ 动态规则配置：所有规则从数据库读取，无需修改代码
-
✅ 批量计算优化：支持按考核期次和数据时点批量计算
-
✅ 完成7个指标实现：C101、C102、C103、C105、C106、C107已上线
📋 文章目录
🎯 指标规则设计
1.1 规则设计
绩效考核系统的核心在于规则的可配置性和计算的准确性。我们采用"配置驱动"的设计理念：
┌─────────────────────────────────────────────────────────────┐
│ 规则配置层 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 指标编码规则 │ │ 判断条件规则 │ │ 分值计算规则 │ │
│ │ (C101-C401) │ │ (已完成/归档) │ │ (个/天/次) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 规则执行层 │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 指标得分计算引擎 │ │
│ │ - 动态匹配规则 │ │
│ │ - 灵活计算分值 │ │
│ │ - 批量更新结果 │ │
│ └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘1.2 核心配置表设计
1.2.1 指标分值配置表
CREATE TABLE *_pa_perf_indicator_score (
id INT PRIMARY KEY AUTO_INCREMENT,
indicator_category VARCHAR(50) COMMENT '指标分类：需求工作/项目工作/运维工作',
indicator_level1_code VARCHAR(20) COMMENT '指标一级编码：C101、C102等',
indicator_level1_name VARCHAR(100) COMMENT '指标名称',
indicator_condition_code VARCHAR(20) COMMENT '判断条件编码（唯一）',
judgment_condition VARCHAR(100) COMMENT '判断条件：已完成、已归档、需求分析等',
indicator_score DECIMAL(5,2) COMMENT '指标分值',
unit VARCHAR(20) COMMENT '计量单位：个、天、次等',
sub_indicator_rule TEXT COMMENT '评分标准规则说明',
indicator_status TINYINT DEFAULT 1 COMMENT '状态：0-禁用 1-启用',
-- 审计字段...
UNIQUE KEY uk_condition_code (indicator_condition_code)
);1.2.2 配置示例数据
|
指标编码 |
判断条件 |
分值 |
单位 |
说明 |
|---|---|---|---|---|
|
C101 |
已提交流程 |
20 |
个 |
需求流程已提交 |
|
C101 |
协同配合 |
5 |
个 |
协同配合完成 |
|
C102 |
需求分析 |
10 |
天 |
按工作天数计算 |
|
C102 |
详细设计 |
10 |
天 |
按工作天数计算 |
|
C105 |
已归档 |
5 |
个 |
文档已归档 |
|
C106 |
已完成 |
2 |
个 |
面试已完成 |
|
C107 |
已完成 |
1 |
个 |
信息维护已完成 |
1.3 指标分类体系
绩效考核指标体系
├── 需求工作类 (C1XX)
│ ├── C101: 需求流程管理
│ ├── C102: 需求分析工作
│ ├── C103: 需求评审管理
│ ├── C105: 文档归档管理
│ ├── C106: 外包面试管理
│ └── C107: 外包信息维护
├── 项目工作类 (C2XX)
│ ├── C201-C214: 项目管理相关
├── 运维工作类 (C3XX)
│ └── C301: 运维值班管理
└── 管理工作类 (C4XX)
├── C401: 管理指标1
└── C402: 管理指标2🏗️ 计算引擎架构
2.1 整体计算流程
┌─────────────────────────────────────────────────────────────────────┐
│ 指标得分计算流程 │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 步骤1: 触发条件 │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ Excel文件上传并解析成功 │ │
│ │ - 考核期次 (assessment_period) │ │
│ │ - 数据时点 (data_date) │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 步骤2: 数据筛选 │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 根据考核期次和数据时点筛选指标明细记录 │ │
│ │ SELECT * FROM *_pa_kpi_cXXX │ │
│ │ WHERE assessment_period = ? AND data_date = ? │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 步骤3: 规则匹配 │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 遍历每条记录，获取judgment_condition │ │
│ │ 查询评分规则表： │ │
│ │ SELECT * FROM *_pa_perf_indicator_score │ │
│ │ WHERE indicator_level1_code = 'CXXX' │ │
│ │ AND judgment_condition = ? │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 步骤4: 分值计算 │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 根据unit类型计算： │ │
│ │ IF unit == '个': │ │
│ │ score = indicator_score × 1 │ │
│ │ ELIF unit == '天': │ │
│ │ score = indicator_score × work_days │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 步骤5: 结果更新 │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ UPDATE *_pa_kpi_cXXX │ │
│ │ SET indicator_score = ? │ │
│ │ WHERE id = ? │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘2.2 服务层架构
# 统一的服务层设计模式
class *PaKpiCXXXService:
"""
指标XXX服务层
"""
@classmethod
async def calculate_scores_yurdmc_pa_kpi_cXXX_service(
cls,
auth: AuthSchema,
assessment_period: str = None,
data_date = None
) -> dict:
"""
计算CXXX指标得分
参数:
- auth: 认证信息
- assessment_period: 考核期次（用于筛选数据）
- data_date: 数据时点（用于筛选数据）
返回:
- success_count: 成功更新记录数
- total_count: 总记录数
"""
pass⚙️ 核心算法详解
3.1 分值计算算法
async def calculate_scores_service(cls, auth, assessment_period, data_date):
"""核心计算逻辑"""
# 1. 构建搜索条件
search_dict = {}
if assessment_period:
search_dict['assessment_period'] = assessment_period
if data_date:
search_dict['data_date'] = data_date
# 2. 获取指标记录
records = await CRUD(auth).list_crud(search=search_dict)
# 3. 遍历计算
for record in records:
judgment_condition = record.judgment_condition
if not judgment_condition:
continue
# 4. 查询评分规则
score_rules = await ScoreCRUD(auth).list_crud(
search={
'indicator_level1_code': 'CXXX',
'judgment_condition': judgment_condition
}
)
if score_rules:
rule = score_rules[0]
unit = rule.unit or '个'
# 5. 计算分值
if unit == '个':
score = Decimal(str(rule.indicator_score)) * Decimal('1')
elif unit == '天':
work_days = getattr(record, 'work_days', 0)
score = Decimal(str(rule.indicator_score)) * Decimal(str(work_days))
# 6. 更新结果
await CRUD(auth).update_crud(
id=record.id,
data={'indicator_score': float(score)}
)3.2 计算示例
示例1：C101（按"个"计算）
场景：需求流程已提交
数据库配置：
- indicator_level1_code: C101
- judgment_condition: 已提交流程
- indicator_score: 20
- unit: 个
计算过程：
indicator_score = 20 × 1 = 20
结果：该条记录得分为20分示例2：C102（按"天"计算）
场景：需求分析工作
数据库配置：
- indicator_level1_code: C102
- judgment_condition: 需求分析
- indicator_score: 10
- unit: 天
业务数据：
- work_days: 0.5
计算过程：
indicator_score = 10 × 0.5 = 5
结果：该条记录得分为5分示例3：C105（按"个"计算）
场景：文档已归档
数据库配置：
- indicator_level1_code: C105
- judgment_condition: 已归档
- indicator_score: 5
- unit: 个
计算过程：
indicator_score = 5 × 1 = 5
结果：该条记录得分为5分🔧 开发指南
4.1 新增指标计算步骤
步骤1：添加数据库字段
ALTER TABLE `*_pa_kpi_cXXX`
ADD COLUMN `indicator_score` DECIMAL(5,2) NULL COMMENT '指标得分'
AFTER `judgment_condition`;步骤2：更新模型层 (model.py)
# *_pa_kpi_cXXX/model.py
indicator_score: Mapped[float | None] = mapped_column(
Numeric(5, 2),
nullable=True,
comment='指标得分'
)步骤3：更新Schema层 (schema.py)
# *_pa_kpi_cXXX/schema.py
class *PaKpiCXXXCreateSchema(BaseModel):
# ... 其他字段
indicator_score: float | None = Field(default=None, description='指标得分')步骤4：实现计算服务 (service.py)
# yurdmc_pa_kpi_cXXX/service.py
from decimal import Decimal
from ..yurdmc_pa_perf_indicator_score.crud import YurdmcPaPerfIndicatorScoreCRUD
class YurdmcPaKpiCXXXService:
@classmethod
async def calculate_scores_yurdmc_pa_kpi_cXXX_service(
cls, auth, assessment_period=None, data_date=None
):
# 实现计算逻辑
pass步骤5：集成到主流程
# *_pa_excel_upload_record/service.py
from ..yurdmc_pa_kpi_cXXX.service import YurdmcPaKpiCXXXService
cXXX_result = await *PaKpiCXXXService.calculate_scores_yurdmc_pa_kpi_cXXX_service(
auth=auth,
assessment_period=record.assessment_period,
data_date=record.data_date
)4.2 界面展示
📊 统一架构图
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 绩效考核系统统一架构图 │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 前端层 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 文件上传 │ │ 数据展示 │ │ 配置管理 │ │ 进度监控 │ │
│ │ (上传Excel) │ │ (指标明细) │ │ (分值规则) │ │ (计算状态) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
↓ HTTP/REST API
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 控制器层 │
│ ┌─────────────────────────────────────────────────────────────────────────────┐ │
│ │ Excel上传记录控制器 │ │
│ │ POST /upload - 文件上传解析 │ │
│ │ POST /parse/{id} - 触发解析流程 │ │
│ │ POST /calculate/{id} - 计算考核得分 │ │
│ └─────────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 服务层 │
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐ │
│ │ 动态Excel解析器 (DynamicExcelParser) │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ 映射配置解析 │ │ 数据校验转换 │ │ 批量插入数据 │ │ │
│ │ │ (读取配置表) │ │ (类型检查) │ │ (事务处理) │ │ │
│ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐ │
│ │ 数据清洗工具 (DataCleaningUtil) │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ 空值检测过滤 │ │ 业务字段检查 │ │ 清洗日志记录 │ │ │
│ │ │ (必填字段) │ │ (非系统字段) │ │ (追踪分析) │ │ │
│ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐ │
│ │ 指标得分计算引擎 (ScoreCalculator) │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ 规则动态匹配 │ │ 分值灵活计算 │ │ 批量结果更新 │ │ │
│ │ │ (judgment_) │ │ (个/天/次) │ │ (异步处理) │ │ │
│ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 数据层 │
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐ │
│ │ 配置表 (Configuration) │ │
│ │ ┌─────────────────────────┐ ┌─────────────────────────────────────────┐ │ │
│ │ │ *_pa_indicator_ │ │ *_pa_perf_indicator_score │ │ │
│ │ │ mapping │ │ (指标分值配置表) │ │ │
│ │ │ - Excel列映射 │ │ - indicator_level1_code (C101-C401) │ │ │
│ │ │ - 数据库字段映射 │ │ - judgment_condition (判断条件) │ │ │
│ │ │ - 数据类型定义 │ │ - indicator_score (分值) │ │ │
│ │ │ - 必填字段标识 │ │ - unit (计量单位) │ │ │
│ │ └─────────────────────────┘ └─────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐ │
│ │ 业务表 (Business) │ │
│ │ ┌──────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ *_pa_excel_upload_record (上传记录表) │ │ │
│ │ │ - assessment_period (考核期次) │ │ │
│ │ │ - data_date (数据时点) │ │ │
│ │ │ - parse_status (解析状态) │ │ │
│ │ │ - calculation_status (计算状态) │ │ │
│ │ └──────────────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────────────────┐ │
│ │ 指标明细表 (Indicator Detail) │ │
│ │ │ │
│ │ 需求工作类 项目工作类 运维工作类 │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │
│ │ │ C101 需求流程 │ │ C201 项目启动 │ │ C301 运维值班 │ │ │
│ │ │ C102 需求分析 │ │ C202 项目计划 │ └──────────────┘ │ │
│ │ │ C103 需求评审 │ │ C203 项目执行 │ │ │
│ │ │ C105 文档归档 │ │ ... │ │ │
│ │ │ C106 外包面试 │ │ C214 项目收尾 │ │ │
│ │ │ C107 信息维护 │ └──────────────┘ │ │
│ │ └──────────────┘ │ │
│ │ │ │
│ │ 管理工作类 │ │
│ │ ┌──────────────┐ │ │
│ │ │ C401 管理指标1│ │ │
│ │ │ C402 管理指标2│ │ │
│ │ └──────────────┘ │ │
│ │ │ │
│ │ 每个表结构： │ │
│ │ - staff_no (员工工号) │ │
│ │ - assessment_period (考核期次) │ │
│ │ - data_date (数据时点) │ │
│ │ - judgment_condition (判断条件) │ │
│ │ - indicator_score (指标得分) ← 计算结果 │ │
│ └─────────────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ 数据流转图 │
└─────────────────────────────────────────────────────────────────────────────────────┘
Excel文件 映射配置表 数据清洗 指标明细表
│ │ │ │
▼ ▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 上传Excel │ → │ 读取列映射 │ → │ 空值检测 │ → │ 插入业务数据 │
│ 选择考核期次 │ │ 字段类型转换 │ │ 业务字段检查│ │ 状态：待计算 │
└──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘
│ │
│ 分值计算流程 │
│ ▼
│ ┌──────────────┐
│ │ 触发计算引擎 │
│ │ - C101计算 │
│ │ - C102计算 │
│ │ - C103计算 │
│ │ - ... │
│ └──────────────┘
│ │
▼ ▼
┌──────────────┐ ┌──────────────┐
│ 更新记录状态 │ ←─────────────────────────────────│ 更新得分字段 │
│ 计算完成 │ │ indicator_ │
│ 显示结果 │ │ score │
└──────────────┘ └──────────────┘🚀 性能优化
5.1 批量计算优化
# 使用批量更新替代逐条更新
async def batch_update_scores(records, scores):
"""批量更新得分"""
update_data = [
{'id': r.id, 'indicator_score': s}
for r, s in zip(records, scores)
]
await CRUD.batch_update(update_data)5.2 异步处理
# 使用BackgroundTasks进行异步计算
from fastapi import BackgroundTasks
@router.post("/calculate/{id}")
async def calculate(
id: int,
background_tasks: BackgroundTasks,
auth: AuthSchema = Depends(...)
):
# 立即返回响应
background_tasks.add_task(
calculate_scores_task,
auth=auth,
record_id=id
)
return {"msg": "计算任务已启动"}5.3 数据库索引优化
-- 为常用查询字段添加索引
CREATE INDEX idx_assessment_period ON *_pa_kpi_cXXX(assessment_period);
CREATE INDEX idx_data_date ON *_pa_kpi_cXXX(data_date);
CREATE INDEX idx_judgment_condition ON *_pa_kpi_cXXX(judgment_condition);
CREATE INDEX idx_indicator_code ON *_pa_perf_indicator_score(indicator_level1_code);🔍 常见问题
Q1: 如何新增一个判断条件？
A: 只需在*_pa_perf_indicator_score表中插入新记录，无需修改代码：
INSERT INTO *_pa_perf_indicator_score (
indicator_level1_code,
judgment_condition,
indicator_score,
unit
) VALUES (
'C101',
'新增条件',
15.00,
'个'
);Q2: 计算结果为什么不正确？
A: 检查以下几点：
-
judgment_condition字段值是否与配置表匹配 -
indicator_level1_code是否正确对应 -
unit单位类型是否正确（个/天） -
对于"天"单位，
work_days字段是否有值
Q3: 如何支持新的计量单位？
A: 在计算服务中添加新的分支：
if unit == '次':
count = getattr(record, 'occurrence_count', 0)
score = Decimal(str(rule.indicator_score)) * Decimal(str(count))Q4: 如何处理计算失败的情况？
A: 系统已内置错误处理：
try:
# 计算逻辑
except Exception as e:
log.error(f"计算失败，记录ID: {record.id}, 错误: {str(e)}")
# 继续处理下一条，不中断整个流程📈 更新日志
Day 3 (2026-03-26)
新增功能：
-
✅ 指标得分计算引擎
-
✅ 支持多种计量单位（个、天）
-
✅ 完成7个指标实现（C101-C103、C105-C107）
-
✅ 动态规则配置
-
✅ 批量计算优化
技术亮点：
-
配置驱动的计算逻辑
-
灵活的规则扩展机制
-
完善的错误处理和日志记录
📚 系列文章
-
Day 3: 指标规则设计与计算引擎（本文）
💡 温馨提示：本系列文章持续更新中，欢迎关注收藏！如有问题欢迎在评论区留言讨论。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
1
0- 0
已为社区贡献1条内容
所有评论(0)