From 2e0b7dc3356b5138fa3f118d4d7c893fdb62655d Mon Sep 17 00:00:00 2001 From: colaftc Date: Fri, 16 Jan 2026 17:25:21 +0800 Subject: [PATCH] feat: improve for report --- api_v1/views/stateflow/business_object.py | 48 ++++- docs/2026-01-16_summary.md | 51 ++++++ ...ects_query_by_completed_state_and_param.md | 169 ++++++++++++++++++ ...logparameterrecord_parameters_gin_index.py | 20 +++ stateflow/services.py | 152 ++++++++-------- ...st_business_object_crud_and_filters_api.py | 50 ++++++ 6 files changed, 417 insertions(+), 73 deletions(-) create mode 100644 docs/2026-01-16_summary.md create mode 100644 docs/stateflow_business_objects_query_by_completed_state_and_param.md create mode 100644 stateflow/migrations/0022_statelogparameterrecord_parameters_gin_index.py diff --git a/api_v1/views/stateflow/business_object.py b/api_v1/views/stateflow/business_object.py index 86be6a1..0949e2f 100644 --- a/api_v1/views/stateflow/business_object.py +++ b/api_v1/views/stateflow/business_object.py @@ -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 @@ -30,6 +31,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): """过滤整体状态""" @@ -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): diff --git a/docs/2026-01-16_summary.md b/docs/2026-01-16_summary.md new file mode 100644 index 0000000..80a13be --- /dev/null +++ b/docs/2026-01-16_summary.md @@ -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=¶m_key=设计师名称¶m_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()` 的实现（此前曾临时禁用导致测试失败）。 + diff --git a/docs/stateflow_business_objects_query_by_completed_state_and_param.md b/docs/stateflow_business_objects_query_by_completed_state_and_param.md new file mode 100644 index 0000000..4204b72 --- /dev/null +++ b/docs/stateflow_business_objects_query_by_completed_state_and_param.md @@ -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¶m_key=设计师名称¶m_value=AAA&limit=50&offset=0 +Authorization: Bearer +``` + +### 示例 2：只按类型过滤（不启用 state+param） + +请求： + +```http +GET /api/v1/stateflow/business-objects/?content_type_str=printing.plateorder&limit=50&offset=0 +Authorization: Bearer +``` + +### 示例 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 +``` + +--- + +## 错误响应 + +- **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 索引 + diff --git a/stateflow/migrations/0022_statelogparameterrecord_parameters_gin_index.py b/stateflow/migrations/0022_statelogparameterrecord_parameters_gin_index.py new file mode 100644 index 0000000..54b79ed --- /dev/null +++ b/stateflow/migrations/0022_statelogparameterrecord_parameters_gin_index.py @@ -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"], + ), + ), + ] + diff --git a/stateflow/services.py b/stateflow/services.py index c929764..7f0d79b 100644 --- a/stateflow/services.py +++ b/stateflow/services.py @@ -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, diff --git a/stateflow/tests/test_business_object_crud_and_filters_api.py b/stateflow/tests/test_business_object_crud_and_filters_api.py index ffbb7e0..9d9d00e 100644 --- a/stateflow/tests/test_business_object_crud_and_filters_api.py +++ b/stateflow/tests/test_business_object_crud_and_filters_api.py @@ -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):