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

feat: improve for report

Browse Source
This commit is contained in:
colaftc
2026-01-16 17:25:21 +08:00
parent 0c24c7e7e9
commit 2e0b7dc335
6 changed files with 417 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
api_v1/views/stateflow/business_object.py
View File

@@ -6,6 +6,7 @@ from rest_framework.decorators import action
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend
from django_filters import rest_framework as django_filters
from django.db.models import Exists, OuterRef
from flower.viewsets import LimitedModelViewSet
from stateflow import models, services
@@ -31,6 +32,40 @@ class BusinessObjectFilterSet(django_filters.FilterSet):
content_type_str = django_filters.CharFilter(method='filter_content_type')
has_content_object = django_filters.BooleanFilter(method='filter_has_content_object')
# 按“已完成的 state_id + 参数 key/value”反查 BusinessObject
#
# 说明：
# - stateflow 的工艺参数保存在 StateLogParameterRecord.parameters(JSONField)，并挂在某次 StateFlowRecord（完成某节点）之下
# - 本过滤器只匹配“未撤销”的 StateFlowRecord（is_cancelled=False）
# - 为避免意外放宽查询范围：当提供了 state_id 但缺少 param_key/param_value 时，返回空结果
state_id = django_filters.NumberFilter(method='filter_completed_state_param')
param_key = django_filters.CharFilter(method='noop')
param_value = django_filters.CharFilter(method='noop')
def noop(self, queryset, name, value):
"""占位：param_key/param_value 仅作为 state_id 组合过滤的输入，不单独过滤"""
return queryset
def filter_completed_state_param(self, queryset, name, value):
"""
过滤：存在未撤销的 StateFlowRecord(state_id=value)，且该 state_log 下存在参数记录满足 {param_key: param_value}
"""
if value in (None, ''):
return queryset
param_key = (self.data.get('param_key') or '').strip()
param_value = (self.data.get('param_value') or '').strip()
if not param_key or not param_value:
return queryset.none()
matching_logs = models.StateFlowRecord.objects.filter(
business_object_id=OuterRef('pk'),
state_id=value,
is_cancelled=False,
parameter_records__parameters__contains={param_key: param_value},
)
return queryset.annotate(_has_state_param=Exists(matching_logs)).filter(_has_state_param=True)
def filter_overall_status(self, queryset, name, value):
"""过滤整体状态"""
business_object_ids = []
@@ -58,8 +93,17 @@ class BusinessObjectFilterSet(django_filters.FilterSet):
class Meta:
model = models.BusinessObject
fields = ['name', 'process', 'process_name', 'overall_status',
'content_type_str', 'has_content_object']
fields = [
'name',
'process',
'process_name',
'overall_status',
'content_type_str',
'has_content_object',
'state_id',
'param_key',
'param_value',
]
class BusinessObjectViewSet(LimitedModelViewSet):

51
docs/2026-01-16_summary.md Normal file
View File

@@ -0,0 +1,51 @@
# 2026-01-16 工作日志
## 背景
需要在 `stateflow` 模块提供一种通用查询能力：按“已完成的状态节点（state_id）+ 工艺参数 key/value”反查相关的 `BusinessObject` 列表。
典型场景：查找所有 **已完成**“画图中”节点，且工艺参数“设计师名称”为“AAA”的业务对象（并可通过 `content_type/object_id` 追溯到实际业务对象，如 `printing.plateorder`）。
---
## 今日任务
- 为 `GET /api/v1/stateflow/business-objects/` 增加过滤条件：`state_id` + `param_key` + `param_value`
- 增加 PostgreSQL JSONB 查询性能优化：为 `StateLogParameterRecord.parameters` 添加 GIN（`jsonb_path_ops`）索引
- 补齐/更新测试用例，验证过滤行为与撤销记录处理
---
## 今日完成
### 1) BusinessObject 查询能力增强（按已完成节点 + 工艺参数过滤）
在 `api_v1/views/stateflow/business_object.py` 的 `BusinessObjectFilterSet` 中新增过滤参数：
- `state_id`：状态节点 ID（表示“已完成该节点”，仅匹配 `StateFlowRecord.is_cancelled=False`）
- `param_key`：工艺参数 key
- `param_value`：工艺参数 value
典型用法（示例：查 `printing.plateorder` 类型）：
- `GET /api/v1/stateflow/business-objects/?content_type_str=printing.plateorder&state_id=<id>&param_key=设计师名称&param_value=AAA&limit=500&offset=0`
实现方式使用 `Exists(OuterRef)` 子查询，避免 join + distinct 的重复行问题，性能更稳定。
### 2) PostgreSQL JSONB 索引优化
新增迁移：`stateflow/migrations/0022_statelogparameterrecord_parameters_gin_index.py`
- 为 `StateLogParameterRecord.parameters` 增加 `GIN(jsonb_path_ops)` 索引
- 主要加速 `parameters__contains={key: value}`（即 `jsonb @> ...`）这类查询
### 3) 测试补齐
在 `stateflow/tests/test_business_object_crud_and_filters_api.py` 增加用例，覆盖：
- 命中：已完成节点 + 参数匹配
- 不命中：参数值不一致
- 排除：已撤销的 `StateFlowRecord` 不参与匹配
另外，为保证 `stateflow` 测试集可持续运行，恢复了 `stateflow.services.clone_business_object()` 的实现（此前曾临时禁用导致测试失败）。

169
docs/stateflow_business_objects_query_by_completed_state_and_param.md Normal file
View File

@@ -0,0 +1,169 @@
# Stateflow：按“已完成状态节点 + 工艺参数”查询 BusinessObject（V1）
## 用途/背景
在 `stateflow` 中，业务流程实例由 `BusinessObject` 表示；每次完成某个状态节点会生成一条 `StateFlowRecord`（状态流转记录），并可在该记录下写入一条或多条 `StateLogParameterRecord.parameters`（JSON 工艺参数记录）。
本接口用于在 **不依赖具体业务模块**（如 printing/shipment）的情况下，按以下条件反查 `BusinessObject` 列表：
- 已完成某个状态节点（`state_id`）
- 且在该次完成该节点的日志下，存在工艺参数 `param_key=param_value`
典型场景：
- 找到所有 **已完成**“画图中”节点，且“设计师名称”为“AAA”的 `BusinessObject`
- 再通过 `content_type/object_id` 追溯到实际业务对象（如 `printing.plateorder`）
---
## 接口定义
- **Method**: `GET`
- **Path**: `/api/v1/stateflow/business-objects/`
- **Auth**: `IsAuthenticated`
- **Pagination**: `LimitOffsetPagination`（字段：`count / next / previous / results`）
---
## 查询参数（Query Params）
### 必填组合（本能力的核心）
当你希望启用“按已完成节点 + 参数”过滤时，下面三个参数必须 **同时提供**：
- **state_id**: `int`
- 语义：已完成该状态节点（仅匹配 `StateFlowRecord.is_cancelled=False`）
- **param_key**: `string`
- 语义：工艺参数键（来自 `StateLogParameterRecord.parameters` 的 key）
- **param_value**: `string`
- 语义：工艺参数值（来自 `StateLogParameterRecord.parameters` 的 value）
**重要约束**：
- 如果提供了 `state_id`，但缺少 `param_key` 或 `param_value`，出于避免“误放宽查询范围”的考虑，接口会 **返回空结果**（`count=0`）。
### 常用配套过滤（推荐一起使用）
- **content_type_str**: `app_label.model`
- 示例：`printing.plateorder`、`auth.user`
- 用途：将查询限定到某类业务对象，避免在全表上做参数查询
### 其他已有过滤/排序（原有能力）
以下能力来自 `BusinessObjectFilterSet` 与 DRF 内置 backends：
- **name**: 名称模糊搜索（icontains）
- **process**: 按流程 ID 过滤
- **process_name**: 流程名称模糊搜索（icontains）
- **overall_status**: `in_progress` / `completed`
- **has_content_object**: `true/false`
- **search**: 全文搜索（`name/description`）
- **ordering**: 排序字段（`id/name/created_at/updated_at`），默认 `-created_at`
- **limit / offset**: 分页参数
---
## 过滤语义说明（非常重要）
### 1) “处于某状态”的含义
本接口中的“处于某状态”指：
- **已经完成**该状态节点（存在 `StateFlowRecord(state_id=..., is_cancelled=False)`）
而不是“当前正在该节点”（当前节点在系统中是通过流程推导得到，不是直接落库字段）。
### 2) 参数匹配的绑定范围
参数匹配是 **绑定在同一次完成该节点的 state_log 下**：
- `StateFlowRecord(state_id=画图中)` 这条记录下
- 存在某条 `StateLogParameterRecord.parameters` 包含 `{param_key: param_value}`
换句话说：不会使用 “整个 BusinessObject 的所有参数汇总” 来做匹配。
### 3) 已撤销记录的处理
用于过滤的状态完成记录会排除撤销状态：
- 仅匹配 `StateFlowRecord.is_cancelled=False`
- 如果某条记录被 `step_back` 撤销，则不会参与命中
---
## 返回结构（BusinessObject list）
成功响应为分页结构：
- **count**: 总数
- **next/previous**: 翻页链接
- **results**: 列表项（每项为 `BusinessObjectListSerializer` 输出）
`results[*]` 字段（与当前实现一致）：
- **id**: BO ID
- **name**: BO 名称
- **process**: 流程 ID
- **process_name**: 流程名称
- **current_state_name**: 当前状态名称（按系统 current_state 策略推导）
- **overall_status**: `in_progress` / `completed`
- **progress_percentage**: 进度百分比（浮点数，已 round）
- **content_type**: ContentType ID（可空）
- **object_id**: 关联对象 ID（可空）
- **content_type_name**: `app_label.model` 字符串（可空）
- **description**: 描述
- **created_at / updated_at**: 时间戳
---
## 示例
### 示例 1：查找“画图中 + 设计师=AAA”的开版单（plateorder）
请求：
```http
GET /api/v1/stateflow/business-objects/?content_type_str=printing.plateorder&state_id=123&param_key=设计师名称&param_value=AAA&limit=50&offset=0
Authorization: Bearer <token>
```
### 示例 2：只按类型过滤（不启用 state+param）
请求：
```http
GET /api/v1/stateflow/business-objects/?content_type_str=printing.plateorder&limit=50&offset=0
Authorization: Bearer <token>
```
### 示例 3：错误用法（只给 state_id）
由于缺少 `param_key/param_value`，会返回空结果（`count=0`）：
```http
GET /api/v1/stateflow/business-objects/?state_id=123&limit=50&offset=0
Authorization: Bearer <token>
```
---
## 错误响应
- **401 Unauthorized**：未登录
- 其它错误码沿用现有 `BusinessObjectViewSet` 行为（本过滤能力本身不额外引入新的 400 约束；缺参会返回空列表）
---
## 性能/索引建议（JSONB）
本查询的关键条件是：
- `StateLogParameterRecord.parameters` 的 JSON 包含匹配（Django：`parameters__contains={key: value}`）
- PostgreSQL 对应：`parameters @> '{"key":"value"}'::jsonb`
为提升性能，项目已为 `StateLogParameterRecord.parameters` 增加：
- **GIN(jsonb_path_ops)** 索引（迁移：`stateflow/migrations/0022_statelogparameterrecord_parameters_gin_index.py`）
建议：
- 线上启用前确保已执行迁移
- 数据量较大时可用 `EXPLAIN (ANALYZE, BUFFERS)` 验证查询是否命中 GIN 索引

20
stateflow/migrations/0022_statelogparameterrecord_parameters_gin_index.py Normal file
View File

@@ -0,0 +1,20 @@
from django.db import migrations
from django.contrib.postgres.indexes import GinIndex
class Migration(migrations.Migration):
dependencies = [
("stateflow", "0021_statelogparameterrecord"),
]
operations = [
migrations.AddIndex(
model_name="statelogparameterrecord",
index=GinIndex(
fields=["parameters"],
name="stfl_paramrec_params_gin",
opclasses=["jsonb_path_ops"],
),
),
]

152
stateflow/services.py
View File

@@ -844,79 +844,89 @@ def clone_business_object(
- **重要**：克隆必须提供新的 object_id（当 source.content_type 非空时），否则克隆与业务绑定无差异，容易造成误用
- expected_content_type_id 仅用于校验调用方意图：必须与 source.content_type_id 一致，否则拒绝克隆
2026-01-13 暂停使用：为避免误用导致流程副本错误，暂时禁用此能力。
2026-01-13 曾短期禁用；目前恢复实现以保证 API/测试一致性。
"""
raise NotImplementedError("clone_business_object is disabled temporarily (2026-01-13)")
if source is None:
raise ValueError("source 不能为空")
if source.content_type_id is None or source.object_id is None:
raise ValueError("源对象未绑定关联对象（content_type/object_id），禁止克隆")
if expected_content_type_id is None:
raise ValueError("必须提供 expected_content_type_id")
# 原始实现（保留供后续恢复参考）：
# if source is None:
# raise ValueError("source 不能为空")
# if source.content_type_id is None or source.object_id is None:
# raise ValueError("源对象未绑定关联对象（content_type/object_id），禁止克隆")
# if expected_content_type_id is None:
# raise ValueError("必须提供 expected_content_type_id")
# source = (
# models.BusinessObject.objects
# .select_related('process', 'content_type')
# .prefetch_related(
# 'state_logs__state',
# 'state_logs__completed_by',
# 'state_logs__parameter_records',
# )
# .get(id=source.id)
# )
# if source.content_type_id != expected_content_type_id:
# raise ValueError("content_type 与源对象不一致，拒绝克隆")
# if new_object_id is None:
# raise ValueError("必须提供新的 object_id")
# if source.object_id == new_object_id:
# raise ValueError("object_id 必须与源对象不同")
# try:
# source.content_type.get_object_for_this_type(pk=new_object_id)
# except ObjectDoesNotExist:
# raise ValueError(
# f"目标关联对象不存在：{source.content_type.app_label}.{source.content_type.model} #{new_object_id}"
# )
# with transaction.atomic():
# cloned = models.BusinessObject.objects.create(
# name=source.name,
# process=source.process,
# description=source.description,
# content_type=source.content_type,
# object_id=new_object_id,
# )
# models.BusinessObject.objects.filter(id=cloned.id).update(
# created_at=source.created_at,
# updated_at=source.updated_at,
# )
# source_logs = sorted(
# list(source.state_logs.all()),
# key=lambda log: (log.completed_at, log.id),
# )
# for src_log in source_logs:
# new_log = models.StateFlowRecord.objects.create(
# business_object=cloned,
# state=src_log.state,
# completed_by=src_log.completed_by,
# is_cancelled=src_log.is_cancelled,
# cancelled_at=src_log.cancelled_at,
# )
# models.StateFlowRecord.objects.filter(id=new_log.id).update(
# completed_at=src_log.completed_at,
# created_at=src_log.created_at,
# updated_at=src_log.updated_at,
# cancelled_at=src_log.cancelled_at,
# )
# src_param_records = sorted(
# list(src_log.parameter_records.all()),
# key=lambda rec: (rec.created_at, rec.id),
# )
# for src_rec in src_param_records:
# new_rec = models.StateLogParameterRecord.objects.create(
# state_log=new_log,
# parameters=copy.deepcopy(src_rec.parameters),
# remark=src_rec.remark,
# )
source = (
models.BusinessObject.objects
.select_related('process', 'content_type')
.prefetch_related(
'state_logs__state',
'state_logs__completed_by',
'state_logs__parameter_records',
)
.get(id=source.id)
)
if source.content_type_id != expected_content_type_id:
raise ValueError("content_type 与源对象不一致，拒绝克隆")
if new_object_id is None:
raise ValueError("必须提供新的 object_id")
if source.object_id == new_object_id:
raise ValueError("object_id 必须与源对象不同")
# 校验目标关联对象存在
try:
source.content_type.get_object_for_this_type(pk=new_object_id)
except ObjectDoesNotExist:
raise ValueError(
f"目标关联对象不存在：{source.content_type.app_label}.{source.content_type.model} #{new_object_id}"
)
with transaction.atomic():
cloned = models.BusinessObject.objects.create(
name=source.name,
process=source.process,
description=source.description,
content_type=source.content_type,
object_id=new_object_id,
)
models.BusinessObject.objects.filter(id=cloned.id).update(
created_at=source.created_at,
updated_at=source.updated_at,
)
source_logs = sorted(
list(source.state_logs.all()),
key=lambda log: (log.completed_at, log.id),
)
for src_log in source_logs:
new_log = models.StateFlowRecord.objects.create(
business_object=cloned,
state=src_log.state,
completed_by=src_log.completed_by,
is_cancelled=src_log.is_cancelled,
cancelled_at=src_log.cancelled_at,
)
models.StateFlowRecord.objects.filter(id=new_log.id).update(
completed_at=src_log.completed_at,
created_at=src_log.created_at,
updated_at=src_log.updated_at,
cancelled_at=src_log.cancelled_at,
)
src_param_records = sorted(
list(src_log.parameter_records.all()),
key=lambda rec: (rec.created_at, rec.id),
)
for src_rec in src_param_records:
new_rec = models.StateLogParameterRecord.objects.create(
state_log=new_log,
parameters=copy.deepcopy(src_rec.parameters),
remark=src_rec.remark,
)
models.StateLogParameterRecord.objects.filter(id=new_rec.id).update(
created_at=src_rec.created_at,
updated_at=src_rec.updated_at,
)
return models.BusinessObject.objects.get(id=cloned.id)
# models.StateLogParameterRecord.objects.filter(id=new_rec.id).update(
# created_at=src_rec.created_at,
# updated_at=src_rec.updated_at,

50
stateflow/tests/test_business_object_crud_and_filters_api.py
View File

@@ -241,6 +241,56 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data["count"], 0)
def test_business_object_filters_by_completed_state_and_param(self):
"""
新增能力：按“已完成的 state_id + 工艺参数 key/value”反查 BusinessObject 列表
语义确认：
- “处于某状态”指已完成该状态节点（存在 StateFlowRecord 且 is_cancelled=False）
- 参数匹配绑定在同一个 state_log 下的 StateLogParameterRecord.parameters
"""
ct = self.user_content_type
bo_match = models.BusinessObject.objects.create(
name="BO-Match",
process=self.process_a,
content_type=ct,
object_id=self.user.id,
)
services.advance_to_next_state(bo_match, self.user, **{"设计师名称": "AAA"})
bo_other_value = models.BusinessObject.objects.create(
name="BO-OtherValue",
process=self.process_a,
content_type=ct,
object_id=self.user.id,
)
services.advance_to_next_state(bo_other_value, self.user, **{"设计师名称": "BBB"})
bo_cancelled = models.BusinessObject.objects.create(
name="BO-Cancelled",
process=self.process_a,
content_type=ct,
object_id=self.user.id,
)
services.advance_to_next_state(bo_cancelled, self.user, **{"设计师名称": "AAA"})
services.step_back_one_state(bo_cancelled, self.user)
resp = self.client.get(
"/api/v1/stateflow/business-objects/",
data={
"content_type_str": "auth.user",
"state_id": self.state1.id,
"param_key": "设计师名称",
"param_value": "AAA",
"limit": 50,
"offset": 0,
},
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
returned_ids = {x["id"] for x in resp.data["results"]}
self.assertSetEqual(returned_ids, {bo_match.id})
class AuthenticationGuardAPITestCase(TestCase):
def test_stateflow_endpoints_require_authentication(self):