zuowei/erpnew
1
0
Fork 0
forked from erp-dev/erp
Code Pull Requests Activity

fix: current_state has been changed to a computed property

Browse Source
This commit is contained in:
colaftc
2025-11-19 10:23:17 +08:00
parent 7f822b0064
commit 400c0db6db
14 changed files with 964 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
stateflow/admin.py
View File

@@ -186,26 +186,32 @@ class BusinessObjectAdmin(admin.ModelAdmin):
@admin.display(description='当前状态', ordering='process')
def current_state_display(self, obj: models.BusinessObject):
from django.conf import settings
from . import services
current_state = obj.get_current_state()
overall_status = services.get_overall_status(obj)
if overall_status == 'not_started':
return '未开始'
elif overall_status == 'completed':
return f'已完成 ({current_state.name if current_state else "-"})'
elif current_state:
# 显示最后完成的状态
return f'进行中 (下一步: {current_state.name})'
overall_status = services.get_overall_status(obj)
mode = getattr(settings, 'STATEFLOW_CURRENT_STATE_MODE', 'NEXT')
if overall_status == 'completed':
return '已完成'
# 进行中状态
current_state = obj.get_current_state()
if current_state:
if mode == 'NEXT':
return f'进行中 (下一步: {current_state.name})'
else: # LAST
return f'进行中 (已完成: {current_state.name})'
return '-'
@admin.display(description='进度')
def progress(self, obj: models.BusinessObject):
return f"{obj.get_progress_percentage():.1f}%"
@admin.display(description='最后完成状态参数')
@admin.display(description='当前状态参数')
def params(self, obj: models.BusinessObject):
"""显示最后完成状态的参数"""
"""显示当前状态的参数（根据配置模式决定是最后完成的还是下一个待执行的）"""
current_state = obj.get_current_state()
if current_state:
params = current_state.parameters.all()

28
stateflow/models.py
View File

@@ -130,10 +130,36 @@ class BusinessObject(ModelBase):
return f"{self.name} ({self.process.name})"
def get_current_state(self) -> 'State':
"""获取当前状态（通过services推导）"""
"""
获取当前状态（通过services推导，行为可配置）
根据 settings.STATEFLOW_CURRENT_STATE_MODE 决定返回：
- 'NEXT' (默认): 下一个待执行的节点
- 'LAST': 最后完成的节点
注意：推荐在新代码中直接使用 get_last_completed_state() 或 get_next_pending_state() 以明确语义
"""
from . import services
return services.get_business_object_current_state(self)
def get_last_completed_state(self) -> 'State':
"""
获取最后完成的状态
返回：最后一个已完成的状态节点，如果没有则返回 None
"""
from . import services
return services.get_last_completed_state(self)
def get_next_pending_state(self) -> 'State':
"""
获取下一个待执行的状态
返回：下一个待执行的状态节点，如果没有则返回 None
"""
from . import services
return services.get_next_pending_state_simple(self)
def get_completed_node_ids(self) -> List[int]:
"""获取已完成的节点ID列表"""
from . import services

103
stateflow/services.py
View File

@@ -9,27 +9,69 @@ from . import models
User = get_user_model()
def get_last_completed_state(business_object: 'models.BusinessObject') -> Optional['models.State']:
"""
获取最后完成的状态
返回：最后一个已完成的状态节点，如果没有则返回 None
应用场景：
- 显示"刚完成了什么"
- 查询最后完成状态的参数
- 审计日志
"""
last_log = business_object.state_logs.filter(
is_cancelled=False
).order_by('-completed_at', '-id').first()
return last_log.state if last_log else None
def get_next_pending_state_simple(business_object: 'models.BusinessObject') -> Optional['models.State']:
"""
获取下一个待执行的状态（简化版，只返回State对象）
返回：下一个待执行的状态节点，如果没有则返回 None
应用场景：
- 显示"接下来要做什么"
- 获取下一步需要的参数
- 流程推进前的检查
"""
next_pending = get_next_pending_state(business_object, include_parameters=False)
return next_pending['state'] if next_pending else None
def get_business_object_current_state(business_object: 'models.BusinessObject') -> Optional['models.State']:
"""
获取业务对象的当前状态（下一个待执行的节点）
获取业务对象的当前状态（可配置模式）
规则：
根据 settings.STATEFLOW_CURRENT_STATE_MODE 决定返回：
- 'NEXT' (默认): 下一个待执行的节点
- 'LAST': 最后完成的节点
注意：此函数保持原有接口不变，但行为可配置
推荐：在新代码中直接使用 get_last_completed_state() 或 get_next_pending_state_simple() 以明确语义
NEXT模式的规则：
- 返回下一个待执行的节点（第一个未完成的节点）
- 如果所有节点都已完成，返回 None
- 配合进度可以判断具体状态：
- 进度 0% + current_state 非空 → 显示"未开始"
- 进度 > 0% + current_state 非空 → 显示 current_state.name（正在进行）
- current_state = None → 显示"已完成"
注意：current_state 始终返回"下一个待执行的节点"，配合进度可以判断具体状态
- 进度 0% + current_state 非空 → 显示"未开始"（还没开始执行，但知道第一个节点是什么）
- 进度 > 0% + current_state 非空 → 显示 current_state.name（正在进行）
- current_state = None → 显示"已完成"（没有待执行的节点了）
LAST模式的规则：
- 返回最后一个已完成的节点
- 如果没有完成任何节点，返回 None
"""
# 使用 get_next_pending_state 获取下一个待执行的节点
next_pending = get_next_pending_state(business_object, include_parameters=False)
from django.conf import settings
mode = getattr(settings, 'STATEFLOW_CURRENT_STATE_MODE', 'NEXT')
if next_pending:
return next_pending['state']
# 所有节点都已完成，返回 None
return None
if mode == 'LAST':
return get_last_completed_state(business_object)
else: # 默认 'NEXT'
return get_next_pending_state_simple(business_object)
def get_business_object_state_status(business_object: 'models.BusinessObject', state: 'models.State') -> str:
@@ -187,21 +229,42 @@ def get_current_state_parameters(business_object: 'models.BusinessObject') -> Li
return []
def get_display_status(business_object: 'models.BusinessObject') -> str:
"""
获取用于显示的状态文本
规则：
- 如果没有 business_object：返回第一个节点名称（如果有流程），否则返回空字符串
- 如果 current_state 为 None：返回"已完成"
- 否则：返回 current_state 的名称
注意：此函数废除了"未开始"的概念，直接显示节点名称
返回值: 状态显示文本（字符串）
"""
if not business_object:
return ''
current_state = get_business_object_current_state(business_object)
if current_state is None:
# 没有待执行的节点，已完成
return '已完成'
# 直接返回当前状态名称，不管进度如何
return current_state.name
def get_overall_status(business_object: 'models.BusinessObject') -> str:
"""
获取业务对象的整体状态
返回值:
- 'not_started': 未开始（没有任何有效的完成记录）
- 'in_progress': 进行中（有部分状态已完成，但未完成所有）
- 'completed': 已完成（所有状态都已完成）
注意：废除了 'not_started' 状态，进度为0时也返回 'in_progress'
"""
# 检查是否有任何有效的完成记录
has_completed = business_object.state_logs.filter(is_cancelled=False).exists()
if not has_completed:
return 'not_started'
# 检查是否有下一个待执行节点
next_pending = get_next_pending_state(business_object, include_parameters=False)

4
stateflow/tests/test_business_object_api.py
View File

@@ -623,8 +623,8 @@ class BusinessObjectAPITestCase(TestCase):
def test_step_back_when_not_started(self):
"""测试未开始时尝试回退（预期错误）"""
# 验证未开始
self.assertEqual(services.get_overall_status(self.business_object), 'not_started')
# 验证进行中（废除了 not_started，初始状态也是 in_progress）
self.assertEqual(services.get_overall_status(self.business_object), 'in_progress')
# 尝试回退（应该失败）
response = self.client.post(

4
stateflow/tests/test_services.py
View File

@@ -48,9 +48,9 @@ class StateFlowServicesTestCase(TestCase):
self.assertEqual(current_state.name, self.state1.name)
self.assertEqual(self.business_object.get_progress_percentage(), 0.0)
# 整体状态应该是 not_started
# 整体状态应该是 in_progress（废除了 not_started）
status = services.get_overall_status(self.business_object)
self.assertEqual(status, 'not_started')
self.assertEqual(status, 'in_progress')
def test_advance_to_next_state(self):
"""测试推进到下一个状态"""

6
stateflow/tests/test_step_back.py
View File

@@ -43,7 +43,7 @@ class StepBackTestCase(TestCase):
self.assertIn('没有任何状态流转记录', message)
# 验证仍然是未开始状态
self.assertEqual(services.get_overall_status(self.business_object), 'not_started')
self.assertEqual(services.get_overall_status(self.business_object), 'in_progress')
def test_step_back_from_first_state(self):
"""测试从第一个状态回退到未开始"""
@@ -61,7 +61,7 @@ class StepBackTestCase(TestCase):
self.assertIn('状态1', message)
# 验证回到未开始状态
self.assertEqual(services.get_overall_status(self.business_object), 'not_started')
self.assertEqual(services.get_overall_status(self.business_object), 'in_progress')
current_state = services.get_business_object_current_state(self.business_object)
self.assertEqual(current_state.name, self.state1.name)
@@ -143,7 +143,7 @@ class StepBackTestCase(TestCase):
# 第三次回退
success, _ = services.step_back_one_state(self.business_object, self.user)
self.assertTrue(success)
self.assertEqual(services.get_overall_status(self.business_object), 'not_started')
self.assertEqual(services.get_overall_status(self.business_object), 'in_progress')
# 第四次回退（应该失败）
success, message = services.step_back_one_state(self.business_object, self.user)