Compare commits

..
1 Commits
Author SHA1 Message Date
zlwuandCursor 9e3a67a8e4 fix(qc): 调整检测清空重配不再误标工单已完成
try_complete 在无活跃任务时也会把已配置打成已完成;清空全部后再添加会踩中。
现要求至少有一条活跃任务且全部完成后才自动完成,并强制全量重建时清理旧批次,前后端均校验批次须含检测项目。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-11-05 12:18:30 +08:00
5 changed files with 170 additions and 31 deletions

No files matched your search

+11 -4
View File
@@ -56,12 +56,14 @@ def _validate_batch_payload(batch: Dict[str, Any], index: int) -> None:
assignees = batch.get("assigneeUserIds") or batch.get("assignee_user_ids") or []
if not batch_name:
raise ValueError(f"{label}:请填写批次名称")
if not assignees:
raise ValueError(f"{label}:请指定负责人")
if not lab_ids and not third_ids:
raise ValueError(f"{label}:请至少选择一个检测项目")
if len(lab_ids) != len(set(lab_ids)):
raise ValueError(f"{label}:实验室项目不能重复")
if len(third_ids) != len(set(third_ids)):
raise ValueError(f"{label}:第三方项目不能重复")
if not assignees:
raise ValueError(f"{label}:请指定负责人")
def _order_source_for_wo(wo: BizQcWorkOrder) -> str:
@@ -834,8 +836,13 @@ def configure_work_order_batches(
}
keep_batch_ids = {((b.get("id") or "").strip()) for b in batches if (b.get("id") or "").strip()}
# 提交空 batches = 显式清空:强制软删批次及关联任务(含进行中/已完成)
# 提交空 batches = 显式清空;提交批次与库中无交集 = 前端「清空全部后再添加」全量重建
# 两种情况都强制软删旧批次及关联任务(含进行中/已完成)
force_clear_all = not batches
full_replace = bool(batches) and bool(existing_batches) and keep_batch_ids.isdisjoint(
existing_batches.keys()
)
force_remove = is_voided or force_clear_all or full_replace
for bid, batch_row in existing_batches.items():
if bid not in keep_batch_ids:
_soft_delete_batch(
@@ -843,7 +850,7 @@ def configure_work_order_batches(
batch_row,
tenant_no=tenant_no,
user_id=user_id,
force=is_voided or force_clear_all,
force=force_remove,
)
fee_total = Decimal("0")
+56 -27
View File
@@ -3050,6 +3050,17 @@ def configure_work_order(
def try_complete_work_order(db: Session, work_order_id: str, tenant_no: str) -> None:
from app.services.qc_work_order_complete import can_auto_complete_work_order
active = (
db.query(BizQcTask)
.filter(
BizQcTask.work_order_id == work_order_id,
BizQcTask.tenant_no == tenant_no,
BizQcTask.del_flag == 0,
)
.count()
)
pending = (
db.query(BizQcTask)
.filter(
@@ -3060,28 +3071,31 @@ def try_complete_work_order(db: Session, work_order_id: str, tenant_no: str) ->
)
.count()
)
if pending:
return
wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
if wo and wo.work_order_status == "ypz":
wo.work_order_status = "ywc"
task_conclusions = [
c
for (c,) in db.query(BizQcTask.conclusion)
.filter(
BizQcTask.work_order_id == work_order_id,
BizQcTask.tenant_no == tenant_no,
BizQcTask.del_flag == 0,
)
.all()
]
derived = build_qc_conclusion_fields(task_conclusions)["qcConclusion"]
wo.conclusion = derived or None
wo.update_time = datetime.now()
db.add(wo)
from app.crud.pilot_outbound_status_sync import sync_schedule_from_qc_work_order
if not wo or not can_auto_complete_work_order(
status=wo.work_order_status or "",
active_task_count=active,
pending_task_count=pending,
):
return
wo.work_order_status = "ywc"
task_conclusions = [
c
for (c,) in db.query(BizQcTask.conclusion)
.filter(
BizQcTask.work_order_id == work_order_id,
BizQcTask.tenant_no == tenant_no,
BizQcTask.del_flag == 0,
)
.all()
]
derived = build_qc_conclusion_fields(task_conclusions)["qcConclusion"]
wo.conclusion = derived or None
wo.update_time = datetime.now()
db.add(wo)
from app.crud.pilot_outbound_status_sync import sync_schedule_from_qc_work_order
sync_schedule_from_qc_work_order(db, wo)
sync_schedule_from_qc_work_order(db, wo)
db.commit()
@@ -3418,6 +3432,18 @@ def configure_temp_qc_work_order(
def try_complete_temp_qc_work_order(
db: Session, work_order_id: str, tenant_no: str
) -> None:
from app.services.qc_work_order_complete import can_auto_complete_work_order
active = (
db.query(BizQcTask)
.filter(
BizQcTask.work_order_id == work_order_id,
BizQcTask.order_source == "temporary",
BizQcTask.tenant_no == tenant_no,
BizQcTask.del_flag == 0,
)
.count()
)
pending = (
db.query(BizQcTask)
.filter(
@@ -3429,14 +3455,17 @@ def try_complete_temp_qc_work_order(
)
.count()
)
if pending:
return
wo = get_temp_qc_work_order(db, work_order_id, tenant_no)
if wo and wo.work_order_status == "ypz":
wo.work_order_status = "ywc"
wo.update_time = datetime.now()
db.add(wo)
db.commit()
if not wo or not can_auto_complete_work_order(
status=wo.work_order_status or "",
active_task_count=active,
pending_task_count=pending,
):
return
wo.work_order_status = "ywc"
wo.update_time = datetime.now()
db.add(wo)
db.commit()
def delete_temp_qc_work_order(
@@ -0,0 +1,18 @@
"""质检工单自动完成判定(无 DB 依赖,供 CRUD 与单测共用)。"""
def can_auto_complete_work_order(
*, status: str, active_task_count: int, pending_task_count: int
) -> bool:
"""仅当工单为已配置、存在活跃任务且全部已完成时,才可标为已完成。
活跃任务数为 0 时不得完成:调整检测清空后重配若尚未产生任务,
否则会把「已配置」误标为「已完成」。
"""
if (status or "").strip() != "ypz":
return False
if active_task_count <= 0:
return False
if pending_task_count > 0:
return False
return True
@@ -0,0 +1,81 @@
"""质检工单自动完成条件:无活跃任务时不得标为已完成(issue #32)。"""
import unittest
class TestCanAutoCompleteWorkOrder(unittest.TestCase):
def test_zero_active_tasks_must_not_complete(self):
"""调整检测清空后重配若尚未产生任务,不能把已配置打成已完成。"""
from app.services.qc_work_order_complete import can_auto_complete_work_order
self.assertFalse(
can_auto_complete_work_order(
status="ypz", active_task_count=0, pending_task_count=0
)
)
def test_all_tasks_completed_can_complete(self):
from app.services.qc_work_order_complete import can_auto_complete_work_order
self.assertTrue(
can_auto_complete_work_order(
status="ypz", active_task_count=3, pending_task_count=0
)
)
def test_pending_tasks_block_complete(self):
from app.services.qc_work_order_complete import can_auto_complete_work_order
self.assertFalse(
can_auto_complete_work_order(
status="ypz", active_task_count=3, pending_task_count=1
)
)
def test_non_configured_status_not_completed(self):
from app.services.qc_work_order_complete import can_auto_complete_work_order
self.assertFalse(
can_auto_complete_work_order(
status="dpz", active_task_count=2, pending_task_count=0
)
)
self.assertFalse(
can_auto_complete_work_order(
status="ywc", active_task_count=2, pending_task_count=0
)
)
class TestValidateBatchRequiresItems(unittest.TestCase):
def test_batch_without_items_rejected(self):
from app.crud.qc_inspection_batch import _validate_batch_payload
with self.assertRaises(ValueError) as ctx:
_validate_batch_payload(
{
"batchName": "B1",
"assigneeUserIds": ["u1"],
"labItemIds": [],
"thirdPartyItemIds": [],
},
0,
)
self.assertIn("检测项目", str(ctx.exception))
def test_batch_with_lab_item_ok(self):
from app.crud.qc_inspection_batch import _validate_batch_payload
_validate_batch_payload(
{
"batchName": "B1",
"assigneeUserIds": ["u1"],
"labItemIds": ["lab-1"],
"thirdPartyItemIds": [],
},
0,
)
if __name__ == "__main__":
unittest.main()
@@ -576,6 +576,10 @@
ElMessage.warning(`${label}:请指定负责人`)
return false
}
if (!(b.labItemIds?.length || 0) && !(b.thirdPartyItemIds?.length || 0)) {
ElMessage.warning(`${label}:请至少选择一个检测项目`)
return false
}
const labSet = new Set(b.labItemIds)
if (labSet.size !== b.labItemIds.length) {
ElMessage.warning(`${label}:实验室项目不能重复`)