Files
zkxw/.sisyphus/plans/2025-05-17-qc-item-params-and-task-results.md
T
2026-05-17 23:08:38 +08:00

22 KiB

检测参数模板 + 质检任务结果录入 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 为检测项目(实验室自检 + 第三方)增加参数配置能力,在质检任务中支持结构化检测结果录入(填参数值 + 上传附件)。

Architecture: 新增 biz_qc_item_param(参数模板表)和 biz_qc_task_result(任务结果值表)。参数模板挂在检测项目下,工单配置生成任务时自动携带参数快照,检测人员在任务中填写参数值后提交完成任务。

Tech Stack: MySQL 5.7 / SQLAlchemy / FastAPI / Pydantic v2 / Vue 3 / Element Plus / TypeScript


File Structure

New files

  • docs/sql/qc_item_param_and_task_result.sql — DDL for 2 new tables
  • backend/app/models/qc_param.py — SQLAlchemy models for BizQcItemParam, BizQcTaskResult
  • backend/app/crud/qc_item_param.py — CRUD for item params
  • backend/app/crud/qc_task_result.py — CRUD for task results
  • web-admin/src/api/qc-item-param.ts — Frontend API for item params
  • web-admin/src/api/qc-task-result.ts — Frontend API for task results

Modified files

  • backend/app/models/models.py — Import new models
  • backend/app/api/qc.py — New endpoints for param CRUD + task result submission
  • backend/app/crud/qc_task.py — on task creation, snapshot params; on complete, validate
  • backend/app/api/qc.py — Task complete endpoint accepts result values
  • web-admin/src/api/qc-task.ts — Update types, add result submission API
  • web-admin/src/views/quality-inspection/lab-test-item/index.vue — Add param config in dialog
  • web-admin/src/views/quality-inspection/third-party-company/index.vue — Add param config in item dialog
  • web-admin/src/views/quality-inspection/qc-task/index.vue — Replace simple "完成" with "填写结果" dialog

Task 1: Create DDL for new tables

Files:

  • Create: docs/sql/qc_item_param_and_task_result.sql

  • Step 1: Write the DDL

-- 检测项目参数模板
CREATE TABLE IF NOT EXISTS `biz_qc_item_param` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `item_source` varchar(16) NOT NULL COMMENT 'lab实验室 third_party第三方',
  `item_id` varchar(64) NOT NULL COMMENT '检测项目ID',
  `param_name` varchar(200) NOT NULL COMMENT '参数名称',
  `param_code` varchar(64) NOT NULL COMMENT '参数编码',
  `param_unit` varchar(32) DEFAULT NULL COMMENT '单位',
  `param_type` varchar(16) NOT NULL DEFAULT 'text' COMMENT 'number/text/select',
  `required_flag` char(1) NOT NULL DEFAULT '0' COMMENT '1必填 0非必填',
  `tenant_no` varchar(64) NOT NULL DEFAULT 'zkxw' COMMENT '租户编号',
  `order_num` int(11) DEFAULT NULL COMMENT '排序号',
  `remark` varchar(512) DEFAULT NULL COMMENT '备注',
  `create_by` varchar(64) DEFAULT NULL COMMENT '创建人id',
  `create_dept_id` varchar(64) NOT NULL DEFAULT '' COMMENT '创建人部门ID',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记',
  PRIMARY KEY (`id`),
  KEY `idx_qc_param_item` (`item_source`, `item_id`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='检测项目参数模板';

-- 质检任务检测结果
CREATE TABLE IF NOT EXISTS `biz_qc_task_result` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `task_id` varchar(64) NOT NULL COMMENT '质检任务ID',
  `param_id` varchar(64) DEFAULT NULL COMMENT '参数模板ID',
  `param_name` varchar(200) NOT NULL COMMENT '参数名快照',
  `param_code` varchar(64) DEFAULT NULL COMMENT '参数编码快照',
  `param_unit` varchar(32) DEFAULT NULL COMMENT '单位快照',
  `param_type` varchar(16) NOT NULL DEFAULT 'text' COMMENT '类型快照',
  `param_value` varchar(500) DEFAULT NULL COMMENT '填写值',
  `tenant_no` varchar(64) NOT NULL DEFAULT 'zkxw' COMMENT '租户编号',
  `remark` varchar(512) DEFAULT NULL COMMENT '备注',
  `create_by` varchar(64) DEFAULT NULL COMMENT '创建人id',
  `create_dept_id` varchar(64) NOT NULL DEFAULT '' COMMENT '创建人部门ID',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记',
  PRIMARY KEY (`id`),
  KEY `idx_qc_result_task` (`task_id`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='质检任务检测结果';

Task 2: Backend — SQLAlchemy models

Files:

  • Create: backend/app/models/qc_param.py

  • Modify: backend/app/models/models.py — add import at bottom

  • Step 1: Create backend/app/models/qc_param.py

"""检测参数模板 + 质检任务结果"""
from app.models.base import Base
from sqlalchemy import Column, DateTime, Integer, SmallInteger, String
from datetime import datetime


class BizQcItemParam(Base):
    """检测项目参数模板 biz_qc_item_param"""
    __tablename__ = "biz_qc_item_param"

    id = Column(String(64), primary_key=True, comment="主键ID")
    item_source = Column(String(16), nullable=False, comment="lab实验室 third_party第三方")
    item_id = Column(String(64), nullable=False, comment="检测项目ID")
    param_name = Column(String(200), nullable=False, comment="参数名称")
    param_code = Column(String(64), nullable=False, comment="参数编码")
    param_unit = Column(String(32), nullable=True, comment="单位")
    param_type = Column(String(16), nullable=False, default="text", comment="number/text/select")
    required_flag = Column(String(1), nullable=False, default="0", comment="1必填 0非必填")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    order_num = Column(Integer, nullable=True, comment="排序号")
    remark = Column(String(512), nullable=True, comment="备注")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")


class BizQcTaskResult(Base):
    """质检任务检测结果 biz_qc_task_result"""
    __tablename__ = "biz_qc_task_result"

    id = Column(String(64), primary_key=True, comment="主键ID")
    task_id = Column(String(64), nullable=False, comment="质检任务ID")
    param_id = Column(String(64), nullable=True, comment="参数模板ID")
    param_name = Column(String(200), nullable=False, comment="参数名快照")
    param_code = Column(String(64), nullable=True, comment="参数编码快照")
    param_unit = Column(String(32), nullable=True, comment="单位快照")
    param_type = Column(String(16), nullable=False, default="text", comment="类型快照")
    param_value = Column(String(500), nullable=True, comment="填写值")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    remark = Column(String(512), nullable=True, comment="备注")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")
  • Step 2: Import in models.py

Add at the end of backend/app/models/models.py:

from app.models.qc_param import BizQcItemParam, BizQcTaskResult  # noqa: E402,F401

Task 3: Backend — CRUD for item params

Files:

  • Create: backend/app/crud/qc_item_param.py

This CRUD provides:

  • list_params_by_item(db, item_source, item_id, tenant_no) — get all params for an item
  • save_params_for_item(db, item_source, item_id, params_list, tenant_no, user_id) — full replace strategy: soft-delete existing, create new (simplest approach)
  • _row_to_dict(r) — serialize
"""检测项目参数模板 CRUD"""
from __future__ import annotations
from datetime import datetime
from typing import Any, Dict, List, Optional
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.models.qc_param import BizQcItemParam


def _row_to_dict(r: BizQcItemParam) -> Dict[str, Any]:
    return {
        "id": r.id,
        "paramName": r.param_name,
        "paramCode": r.param_code,
        "paramUnit": r.param_unit or "",
        "paramType": r.param_type or "text",
        "requiredFlag": r.required_flag or "0",
        "orderNum": r.order_num if r.order_num is not None else 0,
    }


def list_params_by_item(
    db: Session, item_source: str, item_id: str, tenant_no: str
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcItemParam)
        .filter(
            BizQcItemParam.item_source == item_source,
            BizQcItemParam.item_id == item_id,
            BizQcItemParam.tenant_no == tenant_no,
            BizQcItemParam.del_flag == 0,
        )
        .order_by(BizQcItemParam.order_num.asc(), BizQcItemParam.create_time.asc())
        .all()
    )
    return [_row_to_dict(r) for r in rows]


def save_params_for_item(
    db: Session,
    item_source: str,
    item_id: str,
    params: List[Dict[str, Any]],
    tenant_no: str,
    user_id: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """Full-replace: soft-delete old params, create new ones."""
    # soft-delete existing
    existing = (
        db.query(BizQcItemParam)
        .filter(
            BizQcItemParam.item_source == item_source,
            BizQcItemParam.item_id == item_id,
            BizQcItemParam.tenant_no == tenant_no,
            BizQcItemParam.del_flag == 0,
        )
        .all()
    )
    now = datetime.now()
    for old in existing:
        old.del_flag = 1
        old.update_by = user_id
        old.update_time = now
        db.add(old)

    # create new
    created = []
    for idx, p in enumerate(params):
        obj = BizQcItemParam(
            id=generate_uuid(),
            item_source=item_source,
            item_id=item_id,
            param_name=p.get("paramName", "").strip(),
            param_code=p.get("paramCode", "").strip(),
            param_unit=(p.get("paramUnit") or "").strip() or None,
            param_type=p.get("paramType", "text"),
            required_flag=p.get("requiredFlag", "0"),
            tenant_no=tenant_no,
            order_num=idx,
            remark=None,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(obj)
        created.append(obj)
    db.commit()
    for obj in created:
        db.refresh(obj)
    return [_row_to_dict(r) for r in created]

Task 4: Backend — CRUD for task results

Files:

  • Create: backend/app/crud/qc_task_result.py
"""质检任务检测结果 CRUD"""
from __future__ import annotations
from datetime import datetime
from typing import Any, Dict, List, Optional
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.models.qc_param import BizQcTaskResult


def _row_to_dict(r: BizQcTaskResult) -> Dict[str, Any]:
    return {
        "id": r.id,
        "paramId": r.param_id or "",
        "paramName": r.param_name,
        "paramCode": r.param_code or "",
        "paramUnit": r.param_unit or "",
        "paramType": r.param_type or "text",
        "paramValue": r.param_value or "",
    }


def list_results_by_task(
    db: Session, task_id: str, tenant_no: str
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcTaskResult)
        .filter(
            BizQcTaskResult.task_id == task_id,
            BizQcTaskResult.tenant_no == tenant_no,
            BizQcTaskResult.del_flag == 0,
        )
        .order_by(BizQcTaskResult.create_time.asc())
        .all()
    )
    return [_row_to_dict(r) for r in rows]


def save_results_for_task(
    db: Session,
    task_id: str,
    results: List[Dict[str, Any]],
    tenant_no: str,
    user_id: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """Full-replace: soft-delete old results, create new ones."""
    existing = (
        db.query(BizQcTaskResult)
        .filter(
            BizQcTaskResult.task_id == task_id,
            BizQcTaskResult.tenant_no == tenant_no,
            BizQcTaskResult.del_flag == 0,
        )
        .all()
    )
    now = datetime.now()
    for old in existing:
        old.del_flag = 1
        old.update_by = user_id
        old.update_time = now
        db.add(old)

    created = []
    for p in results:
        obj = BizQcTaskResult(
            id=generate_uuid(),
            task_id=task_id,
            param_id=(p.get("paramId") or "").strip() or None,
            param_name=p.get("paramName", "").strip(),
            param_code=(p.get("paramCode") or "").strip() or None,
            param_unit=(p.get("paramUnit") or "").strip() or None,
            param_type=p.get("paramType", "text"),
            param_value=(p.get("paramValue") or "").strip() or None,
            tenant_no=tenant_no,
            remark=None,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(obj)
        created.append(obj)
    db.commit()
    for obj in created:
        db.refresh(obj)
    return [_row_to_dict(r) for r in created]

Task 5: Backend — API endpoints

Files:

  • Modify: backend/app/api/qc.py

Add these endpoints (import the new CRUD modules at top):

a) Item param endpoints:

from app.crud import qc_item_param as param_svc
from app.crud import qc_task_result as result_svc


# ---------- 检测项目参数模板 ----------

class ItemParamBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    params: List[Dict[str, Any]] = Field(default_factory=list, alias="params")


@router.get("/qc-item-params")
async def list_item_params(
    itemSource: str = Query(...),
    itemId: str = Query(...),
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        return success_response(param_svc.list_params_by_item(db, itemSource, itemId, tenant_no))
    except Exception as e:
        return error_response(str(e))


@router.put("/qc-item-params")
async def save_item_params(
    body: ItemParamBody,
    itemSource: str = Query(...),
    itemId: str = Query(...),
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        return success_response(
            param_svc.save_params_for_item(db, itemSource, itemId, body.params, tenant_no, _get_current_user_id(user)),
            message="保存成功"
        )
    except Exception as e:
        return error_response(str(e))

b) Task result endpoints:

class TaskResultBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    results: List[Dict[str, Any]] = Field(default_factory=list, alias="results")


@router.get("/qc-tasks/{task_id}/results")
async def get_task_results(
    task_id: str,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcTask:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        return success_response(result_svc.list_results_by_task(db, task_id, tenant_no))
    except Exception as e:
        return error_response(str(e))


class TaskCompleteWithResultBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    results: List[Dict[str, Any]] = Field(default_factory=list, alias="results")
    third_party_attachments: Optional[str] = Field(None, alias="thirdPartyAttachments")


@router.post("/qc-tasks/{task_id}/complete-with-result")
async def complete_task_with_result(
    task_id: str,
    body: TaskCompleteWithResultBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcTask:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        # save results
        if body.results:
            result_svc.save_results_for_task(db, task_id, body.results, tenant_no, user_id)
        # save attachments if provided
        if body.third_party_attachments is not None:
            task_svc.update_task_attachments(db, task_id, attachments=body.third_party_attachments, tenant_no=tenant_no, user_id=user_id)
        # complete task
        obj = task_svc.complete_task(db, task_id, tenant_no=tenant_no, user_id=user_id)
        return success_response(task_svc._task_to_dict(obj), message="已完成")
    except Exception as e:
        return error_response(str(e))

Also add Dict, Any to the from typing import line at top of qc.py.


Task 6: Frontend — API modules

Files:

  • Create: web-admin/src/api/qc-item-param.ts
  • Create: web-admin/src/api/qc-task-result.ts

qc-item-param.ts:

import request from '@/utils/http'

export type QcItemParam = {
  id?: string
  paramName: string
  paramCode: string
  paramUnit?: string
  paramType: string
  requiredFlag: string
  orderNum?: number
}

export function fetchQcItemParams(itemSource: string, itemId: string) {
  return request.get<QcItemParam[]>({
    url: '/web/admin/qc-item-params',
    params: { itemSource, itemId }
  })
}

export function fetchSaveQcItemParams(itemSource: string, itemId: string, params: QcItemParam[]) {
  return request.put<QcItemParam[]>({
    url: '/web/admin/qc-item-params',
    params: { itemSource, itemId },
    data: { params }
  })
}

qc-task-result.ts:

import request from '@/utils/http'

export type QcTaskResultItem = {
  paramId?: string
  paramName: string
  paramCode?: string
  paramUnit?: string
  paramType: string
  paramValue: string
}

export function fetchQcTaskResults(taskId: string) {
  return request.get<QcTaskResultItem[]>({
    url: `/web/admin/qc-tasks/${taskId}/results`
  })
}

export function fetchCompleteTaskWithResult(
  taskId: string,
  data: { results: QcTaskResultItem[]; thirdPartyAttachments?: string }
) {
  return request.post({
    url: `/web/admin/qc-tasks/${taskId}/complete-with-result`,
    data
  })
}

Task 7: Frontend — Lab test item page: add param config

Files:

  • Modify: web-admin/src/views/quality-inspection/lab-test-item/index.vue

In the edit/add dialog form, add a "参数配置" section after the existing form fields (before 备注). The section shows a dynamic table where users can add/remove parameter rows.

Key additions:

  • Import fetchQcItemParams, fetchSaveQcItemParams, type QcItemParam from @/api/qc-item-param
  • Add itemParams ref: ref<QcItemParam[]>([])
  • On dialog open (handleAdd/handleEdit), load existing params: fetchQcItemParams('lab', row.id)
  • On submit, save params alongside the item: fetchSaveQcItemParams('lab', savedId, itemParams.value)
  • Template: form-section with section-title "参数配置", containing a dynamic table with columns: 参数名称, 参数编码, 单位, 类型(select: text/number), 必填(switch), 操作(删除)
  • Add row button at bottom
  • CSS: reuse .form-section + .section-title pattern from third-party-company page

Task 8: Frontend — Third-party company page: add param config to item dialog

Files:

  • Modify: web-admin/src/views/quality-inspection/third-party-company/index.vue

Same pattern as Task 7, but inside the nested item dialog. The item_source is 'third_party'.

Key additions:

  • Import param API
  • Add itemParams ref
  • On handleEditItem / handleAddItem, load params for the item
  • On submitItem, save params alongside the item
  • Template: form-section "参数配置" inside item form dialog

Task 9: Frontend — QC task page: "填写结果" dialog

Files:

  • Modify: web-admin/src/views/quality-inspection/qc-task/index.vue
  • Modify: web-admin/src/api/qc-task.ts — update QcTaskDetail type to include results

Major changes:

a) New "填写结果" dialog replaces the simple "完成" action for lab tasks (sysz):

  • When clicking "完成" on a lab task in jxz status, open a dialog
  • Dialog loads params for this task's item via fetchQcTaskResults(taskId) (if results already exist) and fetchQcItemParams(itemSource, itemId) (for param template)
  • Dynamically render form fields based on param type:
    • number → el-input-number
    • text → el-input
  • Include attachment upload (reuse ProtocolFinanceConfirmFields for third-party tasks)
  • Submit calls fetchCompleteTaskWithResult(taskId, { results, thirdPartyAttachments })

b) Update QcTaskDetail type to include results:

export type QcTaskDetail = QcTask & {
  protocolNo?: string
  sampleCode?: string
  attachmentFiles?: Array<{ id?: string; fileUrl: string; fileName?: string }>
  results?: Array<{
    paramName: string
    paramUnit?: string
    paramType: string
    paramValue: string
  }>
}

c) Update SystemViewDialog handleView to include a "检测结果" section showing param values.

d) For third-party tasks (dsfjc): existing "上传材料" stays, "完成" now opens a combined dialog that has both attachment upload AND result params.


Task 10: Verification

  • Run vite build on web-admin — must pass with 0 errors
  • Run LSP diagnostics on all modified Vue files — 0 errors
  • Manually test the flow:
    1. Create a lab test item with params (e.g. "活率", "细胞总数")
    2. Create a third-party item with params
    3. Go through work order → task flow
    4. Fill in result values and complete task
    5. View completed task — see result values