提交

backend/app.pid

@@ -1 +1 @@
-11478
+7085

backend/app/api/menu.py

@@ -254,7 +254,7 @@ STATIC_MENU_TREE = [
                 "name": "QualityInspectionQcWorkOrder",
                 "path": "qc-work-order",
                 "component": "/quality-inspection/qc-work-order/index",
-                "meta": {"title": "质检工单", "keepAlive": True, "menuCode": "biz:qcWorkOrder:list"},
+                "meta": {"title": "入库质检工单", "keepAlive": True, "menuCode": "biz:qcWorkOrder:list"},
             },
             {
                 "id": "70000000000000000000000000000008",

backend/app/api/production.py

@@ -14,6 +14,20 @@ from app.crud import process_category as process_category_svc
 from app.crud import output_type as output_type_svc
 from app.crud import consumable_pack as consumable_pack_svc
 from app.crud import process as process_svc
+from app.crud import process_route as route_svc
+from app.schemas.process_route import (
+    ProcessRouteCreateBody,
+    ProcessRouteUpdateBody,
+    SetCurrentBody,
+    StepAddBody,
+    StepCopyBody,
+    StepUpdateBody,
+    StepReorderBody,
+    StepReagentUpdateBody,
+    StepConsumableUpdateBody,
+    StepOutputUpdateBody,
+    StepDataCollectionUpdateBody,
+)
 from app.models.models import SysUser
 
 router = APIRouter(prefix="/web/admin", tags=["生产管理"])
@@ -941,3 +955,258 @@ async def delete_process(
         return success_response(None, message="删除成功")
     except ValueError as e:
         return error_response(400, str(e))
+
+
+# ---------------------------------------------------------------------------
+# 8. 工艺路线管理
+# ---------------------------------------------------------------------------
+
+
+@router.get("/process-routes")
+async def list_process_routes(
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
+    routeName: Optional[str] = None,
+    routeCode: Optional[str] = None,
+    productTypeId: Optional[str] = None,
+    user: dict = Depends(require_menu_code("production:process-route:list")),
+    db: Session = Depends(get_db),
+):
+    tenant_no = _get_tenant_no(db, user)
+    items, total = route_svc.list_routes(
+        db, page=page, page_size=pageSize,
+        route_name=routeName, route_code=routeCode,
+        product_type_id=productTypeId, tenant_no=tenant_no,
+    )
+    tp = route_svc.total_pages(total, pageSize)
+    return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
+
+
+@router.get("/process-routes/{route_id}")
+async def get_process_route_detail(
+    route_id: str,
+    _user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    detail = route_svc.get_route_detail(db, route_id)
+    if not detail:
+        return error_response(404, "路线不存在")
+    return success_response(detail)
+
+
+@router.post("/process-routes")
+async def create_process_route(
+    body: ProcessRouteCreateBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = route_svc.create_route(
+            db,
+            route_code=body.route_code,
+            route_name=body.route_name,
+            product_type_id=body.product_type_id,
+            version=body.version,
+            remark=body.remark,
+            tenant_no=tenant_no,
+            user_id=user.get("user_id"),
+        )
+        return success_response(route_svc._route_row_to_dict(obj), message="创建成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.put("/process-routes/{route_id}")
+async def update_process_route(
+    route_id: str,
+    body: ProcessRouteUpdateBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        payload = body.model_dump(exclude_unset=True, by_alias=False)
+        obj = route_svc.update_route(
+            db, route_id,
+            user_id=user.get("user_id"),
+            route_code=payload.get("route_code"),
+            route_name=payload.get("route_name"),
+            product_type_id=payload.get("product_type_id"),
+            version=payload.get("version"),
+            remark=payload.get("remark"),
+        )
+        return success_response(route_svc._route_row_to_dict(obj), message="更新成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.put("/process-routes/set-current")
+async def set_current_process_route(
+    body: SetCurrentBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = route_svc.set_current_route(db, body.route_id, tenant_no)
+        return success_response(route_svc._route_row_to_dict(obj), message="设置成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.delete("/process-routes/{route_id}")
+async def delete_process_route(
+    route_id: str,
+    _user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        route_svc.soft_delete_route(db, route_id)
+        return success_response(None, message="删除成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+# ----- 步骤操作 -----
+
+@router.post("/process-routes/step/add")
+async def add_route_step(
+    body: StepAddBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        step_data = route_svc.add_step(
+            db,
+            route_id=body.route_id,
+            process_id=body.process_id,
+            step_order=body.step_order,
+            tenant_no=tenant_no,
+            user_id=user.get("user_id"),
+        )
+        return success_response(step_data, message="添加成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.post("/process-routes/step/copy")
+async def copy_route_step(
+    body: StepCopyBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        step_data = route_svc.copy_step(
+            db,
+            step_id=body.step_id,
+            tenant_no=tenant_no,
+            user_id=user.get("user_id"),
+        )
+        return success_response(step_data, message="复制成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.put("/process-routes/step/update")
+async def update_route_step(
+    body: StepUpdateBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        payload = body.model_dump(exclude_unset=True, by_alias=False)
+        payload.pop("step_id", None)
+        obj = route_svc.update_step(db, body.step_id, user_id=user.get("user_id"), **payload)
+        return success_response(route_svc._step_row_to_dict(obj), message="更新成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.put("/process-routes/step/reorder")
+async def reorder_route_steps(
+    body: StepReorderBody,
+    _user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        route_svc.reorder_steps(db, body.route_id, body.step_ids)
+        return success_response(None, message="排序成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.delete("/process-routes/step/{step_id}")
+async def delete_route_step(
+    step_id: str,
+    _user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        route_svc.soft_delete_step(db, step_id)
+        return success_response(None, message="删除成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+# ----- 步骤子表 -----
+
+@router.put("/process-routes/step/reagent/update")
+async def update_step_reagents(
+    body: StepReagentUpdateBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items_data = [it.model_dump(by_alias=False) for it in body.items]
+        route_svc.update_step_reagents(db, body.step_id, items_data, tenant_no, user.get("user_id"))
+        return success_response(None, message="更新成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.put("/process-routes/step/consumable/update")
+async def update_step_consumables(
+    body: StepConsumableUpdateBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items_data = [it.model_dump(by_alias=False) for it in body.items]
+        route_svc.update_step_consumables(db, body.step_id, items_data, tenant_no, user.get("user_id"))
+        return success_response(None, message="更新成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.put("/process-routes/step/output/update")
+async def update_step_outputs(
+    body: StepOutputUpdateBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items_data = [it.model_dump(by_alias=False) for it in body.items]
+        route_svc.update_step_outputs(db, body.step_id, items_data, tenant_no, user.get("user_id"))
+        return success_response(None, message="更新成功")
+    except ValueError as e:
+        return error_response(400, str(e))
+
+
+@router.put("/process-routes/step/data-collection/update")
+async def update_step_data_collections(
+    body: StepDataCollectionUpdateBody,
+    user: dict = Depends(require_admin_or_super_admin),
+    db: Session = Depends(get_db),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items_data = [it.model_dump(by_alias=False) for it in body.items]
+        route_svc.update_step_data_collections(db, body.step_id, items_data, tenant_no, user.get("user_id"))
+        return success_response(None, message="更新成功")
+    except ValueError as e:
+        return error_response(400, str(e))

backend/app/api/qc.py

@@ -369,7 +369,9 @@ class TaskResultBody(BaseModel):
 
 class TaskCompleteWithResultBody(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
+    conclusion: str = Field(..., alias="conclusion")
     results: List[Dict[str, Any]] = Field(default_factory=list, alias="results")
+    remark: Optional[str] = Field(None, alias="remark")
     third_party_attachments: Optional[str] = Field(None, alias="thirdPartyAttachments")
 
 
@@ -624,6 +626,7 @@ async def list_qc_tasks(
     taskType: Optional[str] = None,
     taskStatus: Optional[str] = None,
     workOrderId: Optional[str] = None,
+    qcBizType: Optional[str] = None,
     db: Session = Depends(get_db),
     user: dict = Depends(require_menu_code("biz:qcTask:list")),
 ):
@@ -637,6 +640,7 @@ async def list_qc_tasks(
             task_type=taskType,
             task_status=taskStatus,
             work_order_id=workOrderId,
+            qc_biz_type=qcBizType,
             tenant_no=tenant_no,
         )
         return success_response(_page_payload(items, total, page, pageSize))
@@ -727,15 +731,32 @@ async def complete_task_with_result(
     try:
         tenant_no = _get_tenant_no(db, user)
         user_id = _get_current_user_id(user)
+        t = task_svc.get_task(db, task_id, tenant_no=tenant_no)
+        if not t:
+            return error_response("任务不存在")
+        was_done = (t.task_status or "").strip() == "ywc"
         if body.results:
             result_svc.save_results_for_task(db, task_id, body.results, tenant_no, user_id)
+        concl = (body.conclusion or "").strip()
+        if concl not in ("tg", "btg"):
+            return error_response("请选择检测结论（通过/不通过）")
         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,
             )
-        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="已完成")
+        obj = task_svc.apply_task_result(
+            db,
+            task_id,
+            tenant_no=tenant_no,
+            user_id=user_id,
+            conclusion=concl,
+            remark=body.remark,
+        )
+        return success_response(
+            task_svc._task_to_dict(obj, work_order_no="", db=db, tenant_no=tenant_no),
+            message="已保存" if was_done else "已完成",
+        )
     except PermissionError as e:
         return error_response(str(e), code=403)
     except ValueError as e:

backend/app/api/sample_reception.py

@@ -200,13 +200,15 @@ async def list_pending_samples(
     page: int = Query(1, ge=1),
     pageSize: int = Query(10, ge=1, le=100),
     keyword: Optional[str] = None,
+    sourceType: Optional[str] = None,
     db: Session = Depends(get_db),
     user: dict = Depends(require_menu_code("biz:sampleReception:list")),
 ):
     try:
         tenant_no = _get_tenant_no(db, user)
         items, total = sr_svc.list_pending_protocol_samples(
-            db, page=page, page_size=pageSize, tenant_no=tenant_no, keyword=keyword
+            db, page=page, page_size=pageSize, tenant_no=tenant_no, keyword=keyword,
+            source_type=sourceType,
         )
         return success_response(_page_payload(items, total, page, pageSize))
     except PermissionError as e:
@@ -220,22 +222,17 @@ async def list_received_samples(
     page: int = Query(1, ge=1),
     pageSize: int = Query(10, ge=1, le=100),
     keyword: Optional[str] = None,
+    sourceType: Optional[str] = None,
     db: Session = Depends(get_db),
     user: dict = Depends(require_menu_code("biz:sampleReception:list")),
 ):
     try:
         tenant_no = _get_tenant_no(db, user)
-        query = db.query(BizSampleReception).filter(
-            BizSampleReception.del_flag == 0,
-            BizSampleReception.tenant_no == tenant_no,
+        items, total = sr_svc.list_received_protocol_samples(
+            db, page=page, page_size=pageSize, tenant_no=tenant_no, keyword=keyword,
+            source_type=sourceType,
         )
-        if keyword:
-            query = query.filter(BizSampleReception.reception_no.like(f"%{keyword}%"))
-        query = apply_data_scope(query, BizSampleReception, user, db, tenant_no)
-        total = query.count()
-        items = query.order_by(BizSampleReception.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
-        result_list = [sr_svc.get_reception_by_id(db, r.id, tenant_no) for r in items]
-        return success_response(_page_payload(result_list, total, page, pageSize))
+        return success_response(_page_payload(items, total, page, pageSize))
     except PermissionError as e:
         return error_response(str(e), code=403)
     except Exception as e:
@@ -281,6 +278,8 @@ class SampleReceptionCreateBody(BaseModel):
     accept_conclusion: str = Field(..., alias="acceptConclusion")
     receive_attachments: Optional[str] = Field(None, alias="receiveAttachments")
     remark: Optional[str] = None
+    source_type: Optional[str] = Field("cell_storage", alias="sourceType")
+    donation_sample_id: Optional[str] = Field(None, alias="donationSampleId")
 
 
 @router.post("/sample-receptions")
@@ -302,7 +301,9 @@ async def post_sample_reception(
         ]
         obj = sr_svc.create_sample_reception(
             db,
-            protocol_sample_id=body.protocol_sample_id,
+            reception_source=body.source_type or "cell_storage",
+            protocol_sample_id=body.protocol_sample_id if body.source_type != "research_donation" else None,
+            donation_sample_id=body.donation_sample_id if body.source_type == "research_donation" else None,
             tenant_no=tenant_no,
             user_id=user_id,
             create_dept_id=dept_id,

backend/app/crud/collection_protocol.py

@@ -752,11 +752,17 @@ def get_protocol_detail_bundle(
     reception_by_sample = sample_reception_svc.get_receptions_map_for_samples(
         db, sample_ids, tenant_no
     )
+    from app.crud import qc_work_order as qc_work_order_svc
+
+    qc_by_sample = qc_work_order_svc.get_qc_info_map_for_samples(
+        db, sample_ids, tenant_no
+    )
     samples_out: List[Dict[str, Any]] = []
     for s in s_rows:
         item = _sample_to_dict(s)
         item["transport"] = transport_by_sample.get(s.id)
         item["sampleReception"] = reception_by_sample.get(s.id)
+        item["qcInfo"] = qc_by_sample.get(s.id)
         samples_out.append(item)
     out["samples"] = samples_out
     out["storageCert"] = _storage_cert_to_dict(sc) if sc else None
@@ -1849,6 +1855,7 @@ def save_sample_transports(
             id=generate_uuid(),
             collection_protocol_id=collection_protocol_id,
             sample_id=sample_obj.id,
+            sample_code=sample_obj.sample_code or None,
             transport_type=transport_type,
             express_company_id=(tnorm.get("express_company_id") or "").strip() or None,
             waybill_no=(tnorm.get("waybill_no") or "").strip() or None,

backend/app/crud/process.py

@@ -375,7 +375,8 @@ def update_process(
         new_code = process_code.strip()
         if not new_code:
             raise ValueError("工艺编码不能为空")
-        if new_code != obj.process_code and process_code_exists(db, new_code, obj.tenant_no, exclude_id=process_id):
+        old_code = (obj.process_code or "").strip()
+        if new_code != old_code and process_code_exists(db, new_code, obj.tenant_no, exclude_id=process_id):
             raise ValueError("工艺编码已存在")
         obj.process_code = new_code
     if process_name is not None:

backend/app/crud/process_route.py

@@ -0,0 +1,610 @@
+"""工艺路线 CRUD（biz_process_route + biz_process_route_step + 4张步骤子表）"""
+from __future__ import annotations
+
+import math
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple
+
+from sqlalchemy import and_, func, or_
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.models import (
+    BizProcess,
+    BizProcessReagent,
+    BizProcessConsumable,
+    BizProcessOutput,
+    BizProcessDataCollection,
+    BizProcessRoute,
+    BizProcessRouteStep,
+    BizProcessRouteStepReagent,
+    BizProcessRouteStepConsumable,
+    BizProcessRouteStepOutput,
+    BizProcessRouteStepDataCollection,
+)
+
+
+# ---------------------------------------------------------------------------
+# 序列化函数
+# ---------------------------------------------------------------------------
+
+def _route_row_to_dict(r: BizProcessRoute) -> Dict[str, Any]:
+    return {
+        "id": r.id,
+        "routeCode": r.route_code,
+        "routeName": r.route_name,
+        "productTypeId": r.product_type_id or "",
+        "version": r.version or "",
+        "isCurrent": r.is_current if r.is_current is not None else "0",
+        "tenantNo": r.tenant_no,
+        "orderNum": r.order_num if r.order_num is not None else 0,
+        "remark": r.remark or "",
+        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
+        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
+    }
+
+
+def _step_row_to_dict(s: BizProcessRouteStep) -> Dict[str, Any]:
+    return {
+        "id": s.id,
+        "routeId": s.route_id,
+        "processId": s.process_id or "",
+        "stepOrder": s.step_order if s.step_order is not None else 0,
+        "processCode": s.process_code or "",
+        "categoryId": s.category_id or "",
+        "processName": s.process_name or "",
+        "description": s.description or "",
+        "equipmentName": s.equipment_name or "",
+        "estimatedDuration": s.estimated_duration or "",
+        "sopIds": s.sop_ids or "",
+        "remark": s.remark or "",
+    }
+
+
+def _step_reagent_row_to_dict(r: BizProcessRouteStepReagent) -> Dict[str, Any]:
+    return {
+        "id": r.id,
+        "stepId": r.step_id,
+        "reagentCode": r.reagent_code or "",
+        "reagentName": r.reagent_name or "",
+        "spec": r.spec or "",
+        "standardUsage": r.standard_usage or "",
+        "unit": r.unit or "",
+        "isKeyMaterial": r.is_key_material if r.is_key_material is not None else "0",
+        "remark": r.remark or "",
+    }
+
+
+def _step_consumable_row_to_dict(c: BizProcessRouteStepConsumable) -> Dict[str, Any]:
+    return {
+        "id": c.id,
+        "stepId": c.step_id,
+        "consumableCode": c.consumable_code or "",
+        "consumableName": c.consumable_name or "",
+        "specModel": c.spec_model or "",
+        "standardUsage": c.standard_usage or "",
+        "unit": c.unit or "",
+        "isRequired": c.is_required if c.is_required is not None else "1",
+        "remark": c.remark or "",
+    }
+
+
+def _step_output_row_to_dict(o: BizProcessRouteStepOutput) -> Dict[str, Any]:
+    return {
+        "id": o.id,
+        "stepId": o.step_id,
+        "outputTypeId": o.output_type_id or "",
+        "outputName": o.output_name or "",
+        "outputBarcode": o.output_barcode or "",
+        "remark": o.remark or "",
+    }
+
+
+def _step_data_collection_row_to_dict(d: BizProcessRouteStepDataCollection) -> Dict[str, Any]:
+    return {
+        "id": d.id,
+        "stepId": d.step_id,
+        "collectionName": d.collection_name or "",
+        "unit": d.unit or "",
+        "orderNum": d.order_num if d.order_num is not None else 0,
+        "remark": d.remark or "",
+    }
+
+
+# ---------------------------------------------------------------------------
+# 路线主表 CRUD
+# ---------------------------------------------------------------------------
+
+def list_routes(db: Session, *, page: int = 1, page_size: int = 10,
+                route_name: Optional[str] = None, route_code: Optional[str] = None,
+                product_type_id: Optional[str] = None,
+                tenant_no: Optional[str] = None) -> Tuple[List[Dict], int]:
+    def _filtered():
+        q = db.query(BizProcessRoute).filter(BizProcessRoute.del_flag == 0)
+        if tenant_no:
+            q = q.filter(BizProcessRoute.tenant_no == tenant_no)
+        if route_name and route_name.strip():
+            q = q.filter(BizProcessRoute.route_name.like(f"%{route_name.strip()}%"))
+        if route_code and route_code.strip():
+            q = q.filter(BizProcessRoute.route_code == route_code.strip())
+        if product_type_id and product_type_id.strip():
+            q = q.filter(BizProcessRoute.product_type_id == product_type_id.strip())
+        return q
+
+    total = _filtered().order_by(None).with_entities(func.count(BizProcessRoute.id)).scalar() or 0
+    page = max(1, page)
+    page_size = min(max(1, page_size), 100)
+    rows = (
+        _filtered()
+        .order_by(
+            func.coalesce(BizProcessRoute.order_num, 999999).asc(),
+            BizProcessRoute.create_time.desc(),
+        )
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+        .all()
+    )
+    return [_route_row_to_dict(x) for x in rows], int(total)
+
+
+def get_route(db: Session, route_id: str) -> Optional[BizProcessRoute]:
+    return db.query(BizProcessRoute).filter(
+        and_(BizProcessRoute.id == route_id, BizProcessRoute.del_flag == 0)
+    ).first()
+
+
+def get_route_detail(db: Session, route_id: str) -> Optional[Dict[str, Any]]:
+    route = get_route(db, route_id)
+    if not route:
+        return None
+    result = _route_row_to_dict(route)
+
+    steps = (
+        db.query(BizProcessRouteStep)
+        .filter(and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0))
+        .order_by(BizProcessRouteStep.step_order.asc(), BizProcessRouteStep.create_time.asc())
+        .all()
+    )
+    step_list = []
+    for s in steps:
+        sd = _step_row_to_dict(s)
+        reagents = db.query(BizProcessRouteStepReagent).filter(
+            and_(BizProcessRouteStepReagent.step_id == s.id, BizProcessRouteStepReagent.del_flag == 0)
+        ).order_by(BizProcessRouteStepReagent.create_time.asc()).all()
+        consumables = db.query(BizProcessRouteStepConsumable).filter(
+            and_(BizProcessRouteStepConsumable.step_id == s.id, BizProcessRouteStepConsumable.del_flag == 0)
+        ).order_by(BizProcessRouteStepConsumable.create_time.asc()).all()
+        outputs = db.query(BizProcessRouteStepOutput).filter(
+            and_(BizProcessRouteStepOutput.step_id == s.id, BizProcessRouteStepOutput.del_flag == 0)
+        ).order_by(BizProcessRouteStepOutput.create_time.asc()).all()
+        data_collections = db.query(BizProcessRouteStepDataCollection).filter(
+            and_(BizProcessRouteStepDataCollection.step_id == s.id, BizProcessRouteStepDataCollection.del_flag == 0)
+        ).order_by(BizProcessRouteStepDataCollection.order_num.asc()).all()
+
+        sd["reagents"] = [_step_reagent_row_to_dict(r) for r in reagents]
+        sd["consumables"] = [_step_consumable_row_to_dict(c) for c in consumables]
+        sd["outputs"] = [_step_output_row_to_dict(o) for o in outputs]
+        sd["dataCollections"] = [_step_data_collection_row_to_dict(d) for d in data_collections]
+        step_list.append(sd)
+
+    result["steps"] = step_list
+    return result
+
+
+def create_route(db: Session, *, route_code: str, route_name: str,
+                 product_type_id: Optional[str] = None, version: Optional[str] = None,
+                 remark: Optional[str] = None, tenant_no: str = "zkxw",
+                 user_id: Optional[str] = None) -> BizProcessRoute:
+    rc = (route_code or "").strip()
+    if not rc:
+        raise ValueError("路线编码不能为空")
+    now = datetime.now()
+    obj = BizProcessRoute(
+        id=generate_uuid(),
+        route_code=rc,
+        route_name=(route_name or "").strip() or rc,
+        product_type_id=(product_type_id or "").strip() or None,
+        version=(version or "").strip() or None,
+        is_current="0",
+        tenant_no=tenant_no,
+        order_num=0,
+        remark=(remark or "").strip() or None,
+        create_by=user_id,
+        create_dept_id="",
+        create_time=now,
+        update_by=user_id,
+        update_time=now,
+        del_flag=0,
+    )
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def update_route(db: Session, route_id: str, *, user_id: Optional[str] = None, **kwargs) -> BizProcessRoute:
+    obj = get_route(db, route_id)
+    if not obj:
+        raise ValueError("路线不存在")
+    for field in ["route_code", "route_name", "product_type_id", "version", "remark"]:
+        val = kwargs.get(field)
+        if val is not None:
+            setattr(obj, field, (val or "").strip() or None if field != "route_code" else (val or "").strip())
+    obj.update_by = user_id
+    obj.update_time = datetime.now()
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def set_current_route(db: Session, route_id: str, tenant_no: str) -> BizProcessRoute:
+    obj = get_route(db, route_id)
+    if not obj:
+        raise ValueError("路线不存在")
+    now = datetime.now()
+    siblings = db.query(BizProcessRoute).filter(
+        and_(BizProcessRoute.route_code == obj.route_code,
+             BizProcessRoute.tenant_no == tenant_no,
+             BizProcessRoute.del_flag == 0)
+    ).all()
+    for s in siblings:
+        s.is_current = "0"
+        s.update_time = now
+        db.add(s)
+    obj.is_current = "1"
+    obj.update_time = now
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+# ---------------------------------------------------------------------------
+# 步骤子表辅助
+# ---------------------------------------------------------------------------
+
+def _soft_delete_sub_records(db: Session, model_class, step_id: str, now: datetime) -> None:
+    rows = db.query(model_class).filter(
+        and_(model_class.step_id == step_id, model_class.del_flag == 0)
+    ).all()
+    for r in rows:
+        r.del_flag = 1
+        r.update_time = now
+        db.add(r)
+
+
+def _create_step_reagent(db, step_id, tenant_no, user_id, now, data):
+    db.add(BizProcessRouteStepReagent(
+        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
+        reagent_code=(data.get("reagent_code") or "").strip() or None,
+        reagent_name=(data.get("reagent_name") or "").strip() or None,
+        spec=(data.get("spec") or "").strip() or None,
+        standard_usage=(data.get("standard_usage") or "").strip() or None,
+        unit=(data.get("unit") or "").strip() or None,
+        is_key_material=data.get("is_key_material", "0"),
+        remark=(data.get("remark") or "").strip() or None,
+        order_num=0, create_by=user_id, create_dept_id="",
+        create_time=now, update_by=user_id, update_time=now, del_flag=0,
+    ))
+
+
+def _create_step_consumable(db, step_id, tenant_no, user_id, now, data):
+    db.add(BizProcessRouteStepConsumable(
+        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
+        consumable_code=(data.get("consumable_code") or "").strip() or None,
+        consumable_name=(data.get("consumable_name") or "").strip() or None,
+        spec_model=(data.get("spec_model") or "").strip() or None,
+        standard_usage=(data.get("standard_usage") or "").strip() or None,
+        unit=(data.get("unit") or "").strip() or None,
+        is_required=data.get("is_required", "1"),
+        remark=(data.get("remark") or "").strip() or None,
+        order_num=0, create_by=user_id, create_dept_id="",
+        create_time=now, update_by=user_id, update_time=now, del_flag=0,
+    ))
+
+
+def _create_step_output(db, step_id, tenant_no, user_id, now, data):
+    db.add(BizProcessRouteStepOutput(
+        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
+        output_type_id=(data.get("output_type_id") or "").strip() or None,
+        output_name=(data.get("output_name") or "").strip() or None,
+        output_barcode=(data.get("output_barcode") or "").strip() or None,
+        remark=(data.get("remark") or "").strip() or None,
+        order_num=0, create_by=user_id, create_dept_id="",
+        create_time=now, update_by=user_id, update_time=now, del_flag=0,
+    ))
+
+
+def _create_step_data_collection(db, step_id, tenant_no, user_id, now, data):
+    db.add(BizProcessRouteStepDataCollection(
+        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
+        collection_name=(data.get("collection_name") or "").strip() or None,
+        unit=(data.get("unit") or "").strip() or None,
+        order_num=data.get("order_num", 0) or 0,
+        remark=(data.get("remark") or "").strip() or None,
+        create_by=user_id, create_dept_id="",
+        create_time=now, update_by=user_id, update_time=now, del_flag=0,
+    ))
+
+
+# ---------------------------------------------------------------------------
+# 步骤 CRUD
+# ---------------------------------------------------------------------------
+
+def add_step(db: Session, *, route_id: str, process_id: str, step_order: Optional[int] = None,
+             tenant_no: str = "zkxw", user_id: Optional[str] = None) -> Dict[str, Any]:
+    route = get_route(db, route_id)
+    if not route:
+        raise ValueError("路线不存在")
+
+    proc = db.query(BizProcess).filter(
+        and_(BizProcess.id == process_id, BizProcess.del_flag == 0)
+    ).first()
+    if not proc:
+        raise ValueError("工艺不存在")
+
+    now = datetime.now()
+    if step_order is None:
+        max_order = db.query(func.max(BizProcessRouteStep.step_order)).filter(
+            and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0)
+        ).scalar()
+        step_order = (max_order or 0) + 1
+
+    step = BizProcessRouteStep(
+        id=generate_uuid(), route_id=route_id, process_id=process_id,
+        step_order=step_order,
+        process_code=proc.process_code, category_id=proc.category_id,
+        process_name=proc.process_name, description=proc.description,
+        equipment_name=proc.equipment_name, estimated_duration=proc.estimated_duration,
+        sop_ids=proc.sop_ids,
+        tenant_no=tenant_no, order_num=0,
+        remark=None, create_by=user_id, create_dept_id="",
+        create_time=now, update_by=user_id, update_time=now, del_flag=0,
+    )
+    db.add(step)
+    db.flush()
+
+    src_reagents = db.query(BizProcessReagent).filter(
+        and_(BizProcessReagent.process_id == process_id, BizProcessReagent.del_flag == 0)
+    ).all()
+    for r in src_reagents:
+        _create_step_reagent(db, step.id, tenant_no, user_id, now, {
+            "reagent_code": r.reagent_code, "reagent_name": r.reagent_name,
+            "spec": r.spec, "standard_usage": r.standard_usage,
+            "unit": r.unit, "is_key_material": r.is_key_material,
+        })
+
+    src_consumables = db.query(BizProcessConsumable).filter(
+        and_(BizProcessConsumable.process_id == process_id, BizProcessConsumable.del_flag == 0)
+    ).all()
+    for c in src_consumables:
+        _create_step_consumable(db, step.id, tenant_no, user_id, now, {
+            "consumable_code": c.consumable_code, "consumable_name": c.consumable_name,
+            "spec_model": c.spec_model, "standard_usage": c.standard_usage,
+            "unit": c.unit, "is_required": c.is_required, "remark": c.remark,
+        })
+
+    src_outputs = db.query(BizProcessOutput).filter(
+        and_(BizProcessOutput.process_id == process_id, BizProcessOutput.del_flag == 0)
+    ).all()
+    for o in src_outputs:
+        _create_step_output(db, step.id, tenant_no, user_id, now, {
+            "output_type_id": o.output_type_id, "output_name": o.output_name,
+            "output_barcode": o.output_barcode,
+        })
+
+    src_collections = db.query(BizProcessDataCollection).filter(
+        and_(BizProcessDataCollection.process_id == process_id, BizProcessDataCollection.del_flag == 0)
+    ).all()
+    for d in src_collections:
+        _create_step_data_collection(db, step.id, tenant_no, user_id, now, {
+            "collection_name": d.collection_name, "unit": d.unit, "order_num": d.order_num,
+        })
+
+    db.commit()
+    detail = get_route_detail(db, route_id)
+    return detail["steps"][-1] if detail and detail["steps"] else _step_row_to_dict(step)
+
+
+def update_step(db: Session, step_id: str, *, user_id: Optional[str] = None, **kwargs) -> BizProcessRouteStep:
+    step = db.query(BizProcessRouteStep).filter(
+        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
+    ).first()
+    if not step:
+        raise ValueError("步骤不存在")
+    for field in ["process_code", "category_id", "process_name", "description",
+                   "equipment_name", "estimated_duration", "sop_ids"]:
+        val = kwargs.get(field)
+        if val is not None:
+            setattr(step, field, (val or "").strip() or None)
+    step.update_by = user_id
+    step.update_time = datetime.now()
+    db.add(step)
+    db.commit()
+    db.refresh(step)
+    return step
+
+
+def reorder_steps(db: Session, route_id: str, step_ids: List[str]) -> None:
+    now = datetime.now()
+    for idx, sid in enumerate(step_ids):
+        step = db.query(BizProcessRouteStep).filter(
+            and_(BizProcessRouteStep.id == sid, BizProcessRouteStep.route_id == route_id,
+                 BizProcessRouteStep.del_flag == 0)
+        ).first()
+        if step:
+            step.step_order = idx + 1
+            step.update_time = now
+            db.add(step)
+    db.commit()
+
+
+def copy_step(db: Session, step_id: str, *, tenant_no: str = "zkxw",
+              user_id: Optional[str] = None) -> Dict[str, Any]:
+    """复制步骤，插入到原步骤之后，名称追加（复制）"""
+    src = db.query(BizProcessRouteStep).filter(
+        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
+    ).first()
+    if not src:
+        raise ValueError("步骤不存在")
+
+    now = datetime.now()
+    new_order = src.step_order + 1
+
+    # 后续步骤 step_order +1
+    later_steps = db.query(BizProcessRouteStep).filter(
+        and_(
+            BizProcessRouteStep.route_id == src.route_id,
+            BizProcessRouteStep.del_flag == 0,
+            BizProcessRouteStep.step_order > src.step_order,
+        )
+    ).order_by(BizProcessRouteStep.step_order.asc()).all()
+    for ls in later_steps:
+        ls.step_order += 1
+        ls.update_time = now
+        db.add(ls)
+
+    # 创建新步骤
+    new_name = (src.process_name or "").strip()
+    if not new_name.endswith("（复制）"):
+        new_name += "（复制）"
+    new_step = BizProcessRouteStep(
+        id=generate_uuid(), route_id=src.route_id, process_id=src.process_id,
+        step_order=new_order,
+        process_code=src.process_code, category_id=src.category_id,
+        process_name=new_name, description=src.description,
+        equipment_name=src.equipment_name, estimated_duration=src.estimated_duration,
+        sop_ids=src.sop_ids,
+        tenant_no=tenant_no, order_num=0,
+        remark=src.remark, create_by=user_id, create_dept_id="",
+        create_time=now, update_by=user_id, update_time=now, del_flag=0,
+    )
+    db.add(new_step)
+    db.flush()
+
+    # 复制子表：试剂
+    for r in db.query(BizProcessRouteStepReagent).filter(
+        and_(BizProcessRouteStepReagent.step_id == step_id, BizProcessRouteStepReagent.del_flag == 0)
+    ).all():
+        _create_step_reagent(db, new_step.id, tenant_no, user_id, now, {
+            "reagent_code": r.reagent_code, "reagent_name": r.reagent_name,
+            "spec": r.spec, "standard_usage": r.standard_usage,
+            "unit": r.unit, "is_key_material": r.is_key_material,
+            "remark": r.remark,
+        })
+
+    # 复制子表：耗材
+    for c in db.query(BizProcessRouteStepConsumable).filter(
+        and_(BizProcessRouteStepConsumable.step_id == step_id, BizProcessRouteStepConsumable.del_flag == 0)
+    ).all():
+        _create_step_consumable(db, new_step.id, tenant_no, user_id, now, {
+            "consumable_code": c.consumable_code, "consumable_name": c.consumable_name,
+            "spec_model": c.spec_model, "standard_usage": c.standard_usage,
+            "unit": c.unit, "is_required": c.is_required,
+            "remark": c.remark,
+        })
+
+    # 复制子表：产出
+    for o in db.query(BizProcessRouteStepOutput).filter(
+        and_(BizProcessRouteStepOutput.step_id == step_id, BizProcessRouteStepOutput.del_flag == 0)
+    ).all():
+        _create_step_output(db, new_step.id, tenant_no, user_id, now, {
+            "output_type_id": o.output_type_id, "output_name": o.output_name,
+            "output_barcode": o.output_barcode, "remark": o.remark,
+        })
+
+    # 复制子表：数据采集
+    for d in db.query(BizProcessRouteStepDataCollection).filter(
+        and_(BizProcessRouteStepDataCollection.step_id == step_id, BizProcessRouteStepDataCollection.del_flag == 0)
+    ).all():
+        _create_step_data_collection(db, new_step.id, tenant_no, user_id, now, {
+            "collection_name": d.collection_name, "unit": d.unit,
+            "order_num": d.order_num, "remark": d.remark,
+        })
+
+    db.commit()
+    detail = get_route_detail(db, src.route_id)
+    return detail["steps"][-1] if detail and detail["steps"] else _step_row_to_dict(new_step)
+
+
+def soft_delete_step(db: Session, step_id: str) -> None:
+    step = db.query(BizProcessRouteStep).filter(
+        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
+    ).first()
+    if not step:
+        raise ValueError("步骤不存在")
+    now = datetime.now()
+    step.del_flag = 1
+    step.update_time = now
+    db.add(step)
+    _soft_delete_sub_records(db, BizProcessRouteStepReagent, step_id, now)
+    _soft_delete_sub_records(db, BizProcessRouteStepConsumable, step_id, now)
+    _soft_delete_sub_records(db, BizProcessRouteStepOutput, step_id, now)
+    _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, step_id, now)
+    db.commit()
+
+
+def soft_delete_route(db: Session, route_id: str) -> None:
+    route = get_route(db, route_id)
+    if not route:
+        raise ValueError("路线不存在")
+    now = datetime.now()
+    route.del_flag = 1
+    route.update_time = now
+    db.add(route)
+
+    steps = db.query(BizProcessRouteStep).filter(
+        and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0)
+    ).all()
+    for s in steps:
+        s.del_flag = 1
+        s.update_time = now
+        db.add(s)
+        _soft_delete_sub_records(db, BizProcessRouteStepReagent, s.id, now)
+        _soft_delete_sub_records(db, BizProcessRouteStepConsumable, s.id, now)
+        _soft_delete_sub_records(db, BizProcessRouteStepOutput, s.id, now)
+        _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, s.id, now)
+    db.commit()
+
+
+# ---------------------------------------------------------------------------
+# 子表完整替换更新
+# ---------------------------------------------------------------------------
+
+def update_step_reagents(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
+    now = datetime.now()
+    _soft_delete_sub_records(db, BizProcessRouteStepReagent, step_id, now)
+    for it in items:
+        _create_step_reagent(db, step_id, tenant_no, user_id, now, it)
+    db.commit()
+
+
+def update_step_consumables(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
+    now = datetime.now()
+    _soft_delete_sub_records(db, BizProcessRouteStepConsumable, step_id, now)
+    for it in items:
+        _create_step_consumable(db, step_id, tenant_no, user_id, now, it)
+    db.commit()
+
+
+def update_step_outputs(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
+    now = datetime.now()
+    _soft_delete_sub_records(db, BizProcessRouteStepOutput, step_id, now)
+    for it in items:
+        _create_step_output(db, step_id, tenant_no, user_id, now, it)
+    db.commit()
+
+
+def update_step_data_collections(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
+    now = datetime.now()
+    _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, step_id, now)
+    for it in items:
+        _create_step_data_collection(db, step_id, tenant_no, user_id, now, it)
+    db.commit()
+
+
+def total_pages(total: int, page_size: int) -> int:
+    if total <= 0:
+        return 0
+    return int(math.ceil(total / float(page_size)))

backend/app/crud/qc_task.py

@@ -10,7 +10,21 @@ from sqlalchemy.orm import Session
 
 from app.core.utils import generate_uuid
 from app.crud.collection_protocol import urls_to_attachment_list
-from app.models.models import BizQcTask, BizQcWorkOrder
+from app.crud import sample_reception as sample_reception_svc
+from app.crud.qc_work_order import (
+    _load_sample_type_name_map,
+    _qc_biz_type_label,
+    _reception_source_label,
+    _sample_type_name_for_reception,
+    _work_order_qc_biz_type,
+)
+from app.models.models import (
+    BizProtocolSample,
+    BizQcTask,
+    BizQcWorkOrder,
+    BizSampleReception,
+)
+from app.models.research_donation import BizResearchDonationSample
 
 
 def _dec(v: Any) -> Optional[float]:
@@ -27,6 +41,14 @@ def _gen_task_no(db: Session, work_order: BizQcWorkOrder, seq: int) -> str:
     return f"{work_order.work_order_no}-{seq:02d}"
 
 
+_CONCLUSION_LABELS = {"tg": "通过", "btg": "不通过"}
+
+
+def _conclusion_label(v: Optional[str]) -> str:
+    key = (v or "").strip()
+    return _CONCLUSION_LABELS.get(key, key)
+
+
 def _task_to_dict(
     t: BizQcTask,
     work_order_no: str = "",
@@ -47,6 +69,8 @@ def _task_to_dict(
         "companyName": t.company_name or "",
         "feeAmount": _dec(t.fee_amount),
         "taskStatus": t.task_status,
+        "conclusion": t.conclusion or "",
+        "conclusionLabel": _conclusion_label(t.conclusion),
         "thirdPartyAttachments": t.third_party_attachments or "",
         "attachmentFiles": urls_to_attachment_list(t.third_party_attachments, db, tn) if db else [],
         "completedTime": _fmt_dt(t.completed_time),
@@ -57,6 +81,102 @@ def _task_to_dict(
     }
 
 
+def _task_work_order_context(wo: BizQcWorkOrder, t: BizQcTask) -> Dict[str, Any]:
+    biz = _work_order_qc_biz_type(wo)
+    source = getattr(wo, "reception_source", "cell_storage") or "cell_storage"
+    return {
+        "receptionId": wo.reception_id,
+        "protocolNo": wo.protocol_no or "",
+        "sampleCode": wo.sample_code or t.sample_code or "",
+        "qcBizType": biz,
+        "qcBizTypeLabel": _qc_biz_type_label(biz),
+        "receptionSource": source,
+        "receptionSourceLabel": _reception_source_label(source),
+        "receptionTime": "",
+        "handlerName": "",
+        "sampleTypeName": "",
+        "workOrderStatus": wo.work_order_status or "",
+    }
+
+
+def _enrich_tasks_reception_fields(
+    db: Session, items: List[Dict[str, Any]]
+) -> List[Dict[str, Any]]:
+    if not items:
+        return items
+    rec_ids = [x.get("receptionId") for x in items if x.get("receptionId")]
+    if not rec_ids:
+        return items
+    rec_map = {
+        r.id: r
+        for r in db.query(BizSampleReception)
+        .filter(BizSampleReception.id.in_(rec_ids), BizSampleReception.del_flag == 0)
+        .all()
+    }
+    cell_sample_ids = [
+        r.protocol_sample_id
+        for r in rec_map.values()
+        if (r.reception_source or "cell_storage") != "research_donation" and r.protocol_sample_id
+    ]
+    donation_sample_ids = [
+        r.donation_sample_id
+        for r in rec_map.values()
+        if (r.reception_source or "") == "research_donation" and r.donation_sample_id
+    ]
+    protocol_samples: Dict[str, BizProtocolSample] = {}
+    if cell_sample_ids:
+        protocol_samples = {
+            s.id: s
+            for s in db.query(BizProtocolSample)
+            .filter(BizProtocolSample.id.in_(cell_sample_ids), BizProtocolSample.del_flag == 0)
+            .all()
+        }
+    donation_samples: Dict[str, BizResearchDonationSample] = {}
+    if donation_sample_ids:
+        donation_samples = {
+            s.id: s
+            for s in db.query(BizResearchDonationSample)
+            .filter(
+                BizResearchDonationSample.id.in_(donation_sample_ids),
+                BizResearchDonationSample.del_flag == 0,
+            )
+            .all()
+        }
+    type_ids = list(
+        {
+            s.sample_type_id
+            for s in list(protocol_samples.values()) + list(donation_samples.values())
+            if s.sample_type_id
+        }
+    )
+    type_map = _load_sample_type_name_map(db, type_ids)
+    for item in items:
+        rec = rec_map.get(item.get("receptionId") or "")
+        if not rec:
+            continue
+        item["receptionTime"] = _fmt_dt(rec.reception_time)
+        item["handlerName"] = rec.handler_name or ""
+        item["sampleTypeName"] = _sample_type_name_for_reception(
+            db,
+            rec,
+            type_map,
+            protocol_samples=protocol_samples,
+            donation_samples=donation_samples,
+        )
+    return items
+
+
+def _task_row_to_dict(
+    t: BizQcTask,
+    wo: BizQcWorkOrder,
+    db: Session,
+    tenant_no: Optional[str],
+) -> Dict[str, Any]:
+    out = _task_to_dict(t, wo.work_order_no or "", db, tenant_no)
+    out.update(_task_work_order_context(wo, t))
+    return out
+
+
 def create_task_for_line(
     db: Session,
     *,
@@ -79,6 +199,7 @@ def create_task_for_line(
         task_no=_gen_task_no(db, work_order, seq),
         task_type=task_type,
         source_line_id=source_line_id,
+        sample_code=work_order.sample_code or None,
         item_code=item_code,
         item_name=item_name,
         company_id=company_id,
@@ -108,11 +229,12 @@ def list_tasks(
     task_type: Optional[str] = None,
     task_status: Optional[str] = None,
     work_order_id: Optional[str] = None,
+    qc_biz_type: Optional[str] = None,
     tenant_no: Optional[str] = None,
 ) -> Tuple[List[Dict[str, Any]], int]:
     def _filtered():
         q0 = (
-            db.query(BizQcTask, BizQcWorkOrder.work_order_no)
+            db.query(BizQcTask, BizQcWorkOrder)
             .join(
                 BizQcWorkOrder,
                 and_(
@@ -134,6 +256,8 @@ def list_tasks(
                     BizQcTask.task_no.like(like),
                     BizQcTask.item_name.like(like),
                     BizQcWorkOrder.work_order_no.like(like),
+                    BizQcWorkOrder.sample_code.like(like),
+                    BizQcWorkOrder.protocol_no.like(like),
                 )
             )
         tt = (task_type or "").strip()
@@ -142,6 +266,9 @@ def list_tasks(
         st = (task_status or "").strip()
         if st:
             q0 = q0.filter(BizQcTask.task_status == st)
+        bt = (qc_biz_type or "").strip()
+        if bt:
+            q0 = q0.filter(BizQcWorkOrder.qc_biz_type == bt)
         return q0
 
     total = _filtered().order_by(None).with_entities(func.count(BizQcTask.id)).scalar() or 0
@@ -158,22 +285,25 @@ def list_tasks(
         .limit(page_size)
         .all()
     )
-    return [_task_to_dict(t, wno or "", db, tenant_no) for t, wno in rows], int(total)
+    items = [_task_row_to_dict(t, wo, db, tenant_no) for t, wo in rows]
+    return _enrich_tasks_reception_fields(db, items), int(total)
 
 
 def list_tasks_by_work_order(db: Session, work_order_id: str, tenant_no: str) -> List[Dict[str, Any]]:
     rows = (
-        db.query(BizQcTask, BizQcWorkOrder.work_order_no)
+        db.query(BizQcTask, BizQcWorkOrder)
         .join(BizQcWorkOrder, BizQcWorkOrder.id == BizQcTask.work_order_id)
         .filter(
             BizQcTask.work_order_id == work_order_id,
             BizQcTask.tenant_no == tenant_no,
             BizQcTask.del_flag == 0,
+            BizQcWorkOrder.del_flag == 0,
         )
         .order_by(BizQcTask.order_num.asc())
         .all()
     )
-    return [_task_to_dict(t, wno or "", db, tenant_no) for t, wno in rows]
+    items = [_task_row_to_dict(t, wo, db, tenant_no) for t, wo in rows]
+    return _enrich_tasks_reception_fields(db, items)
 
 
 def get_task(
@@ -189,15 +319,22 @@ def get_task_detail(db: Session, row_id: str, tenant_no: str) -> Dict[str, Any]:
     row = (
         db.query(BizQcTask, BizQcWorkOrder)
         .join(BizQcWorkOrder, BizQcWorkOrder.id == BizQcTask.work_order_id)
-        .filter(BizQcTask.id == row_id, BizQcTask.tenant_no == tenant_no, BizQcTask.del_flag == 0)
+        .filter(
+            BizQcTask.id == row_id,
+            BizQcTask.tenant_no == tenant_no,
+            BizQcTask.del_flag == 0,
+            BizQcWorkOrder.del_flag == 0,
+        )
         .first()
     )
     if not row:
         raise ValueError("任务不存在")
     t, wo = row
-    out = _task_to_dict(t, wo.work_order_no or "", db, tenant_no)
-    out["protocolNo"] = wo.protocol_no or ""
-    out["sampleCode"] = wo.sample_code or ""
+    out = _task_row_to_dict(t, wo, db, tenant_no)
+    out = _enrich_tasks_reception_fields(db, [out])[0]
+    reception_detail = sample_reception_svc.get_reception_by_id(db, wo.reception_id, tenant_no)
+    if reception_detail:
+        out["reception"] = reception_detail
     return out
 
 
@@ -218,21 +355,37 @@ def start_task(db: Session, row_id: str, *, tenant_no: str, user_id: Optional[st
     return t
 
 
-def complete_task(db: Session, row_id: str, *, tenant_no: str, user_id: Optional[str]) -> BizQcTask:
+def _set_task_remark(t: BizQcTask, remark: Optional[str]) -> None:
+    t.remark = (remark or "").strip() or None
+
+
+def complete_task(
+    db: Session,
+    row_id: str,
+    *,
+    tenant_no: str,
+    user_id: Optional[str],
+    conclusion: Optional[str] = None,
+    remark: Optional[str] = None,
+) -> BizQcTask:
     t = get_task(db, row_id, tenant_no=tenant_no)
     if not t:
         raise ValueError("任务不存在")
+    concl = (conclusion or "").strip() if conclusion is not None else ""
+    if concl and concl not in ("tg", "btg"):
+        raise ValueError("检测结论无效")
     if t.task_type == "sysz":
         if t.task_status not in ("dzx", "jxz"):
             raise ValueError("当前状态不可完成")
     elif t.task_type == "dsfjc":
         if not (t.third_party_attachments or "").strip():
             raise ValueError("请先上传第三方检测材料")
-        if t.task_status == "ywc":
-            raise ValueError("任务已完成")
     else:
         raise ValueError("任务类型无效")
     t.task_status = "ywc"
+    if concl:
+        t.conclusion = concl
+    _set_task_remark(t, remark)
     t.completed_time = datetime.now()
     t.completed_by = user_id
     t.update_by = user_id
@@ -246,6 +399,41 @@ def complete_task(db: Session, row_id: str, *, tenant_no: str, user_id: Optional
     return t
 
 
+def apply_task_result(
+    db: Session,
+    row_id: str,
+    *,
+    tenant_no: str,
+    user_id: Optional[str],
+    conclusion: str,
+    remark: Optional[str] = None,
+) -> BizQcTask:
+    """保存检测结果与结论；任务未完成时顺带标记为已完成。"""
+    t = get_task(db, row_id, tenant_no=tenant_no)
+    if not t:
+        raise ValueError("任务不存在")
+    concl = (conclusion or "").strip()
+    if concl not in ("tg", "btg"):
+        raise ValueError("请选择检测结论（通过/不通过）")
+    if (t.task_status or "").strip() == "ywc":
+        t.conclusion = concl
+        _set_task_remark(t, remark if remark is not None else "")
+        t.update_by = user_id
+        t.update_time = datetime.now()
+        db.add(t)
+        db.commit()
+        db.refresh(t)
+        return t
+    return complete_task(
+        db,
+        row_id,
+        tenant_no=tenant_no,
+        user_id=user_id,
+        conclusion=concl,
+        remark=remark,
+    )
+
+
 def update_task_attachments(
     db: Session,
     row_id: str,
@@ -257,12 +445,12 @@ def update_task_attachments(
     t = get_task(db, row_id, tenant_no=tenant_no)
     if not t:
         raise ValueError("任务不存在")
-    if t.task_type != "dsfjc":
-        raise ValueError("仅第三方检测任务可上传材料")
-    if t.task_status == "ywc":
-        raise ValueError("任务已完成，不可修改附件")
     t.third_party_attachments = (attachments or "").strip() or None
-    if t.task_status == "dzx" and t.third_party_attachments:
+    if (
+        t.task_type == "dsfjc"
+        and t.task_status == "dzx"
+        and t.third_party_attachments
+    ):
         t.task_status = "jxz"
     t.update_by = user_id
     t.update_time = datetime.now()

backend/app/crud/qc_task_result.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from sqlalchemy.orm import Session
 
 from app.core.utils import generate_uuid
+from app.crud import qc_item_param as param_svc
+from app.crud.qc_task import get_task
+from app.models.models import BizQcWorkOrderLabItem, BizQcWorkOrderThirdItem
 from app.models.qc_param import BizQcTaskResult
 
 
@@ -21,7 +24,7 @@ def _row_to_dict(r: BizQcTaskResult) -> Dict[str, Any]:
     }
 
 
-def list_results_by_task(
+def _list_saved_results_by_task(
     db: Session, task_id: str, tenant_no: str
 ) -> List[Dict[str, Any]]:
     rows = (
@@ -37,6 +40,74 @@ def list_results_by_task(
     return [_row_to_dict(r) for r in rows]
 
 
+def _resolve_task_item_ref(db: Session, task) -> Tuple[Optional[str], Optional[str]]:
+    """返回 (item_source, item_id) 供 biz_qc_item_param 查询。"""
+    if task.task_type == "sysz":
+        line = (
+            db.query(BizQcWorkOrderLabItem)
+            .filter(
+                BizQcWorkOrderLabItem.id == task.source_line_id,
+                BizQcWorkOrderLabItem.del_flag == 0,
+            )
+            .first()
+        )
+        if line:
+            return "lab", line.lab_test_item_id
+    elif task.task_type == "dsfjc":
+        line = (
+            db.query(BizQcWorkOrderThirdItem)
+            .filter(
+                BizQcWorkOrderThirdItem.id == task.source_line_id,
+                BizQcWorkOrderThirdItem.del_flag == 0,
+            )
+            .first()
+        )
+        if line:
+            return "third_party", line.third_party_item_id
+    return None, None
+
+
+def _merge_saved_into_template(
+    templates: List[Dict[str, Any]], saved: List[Dict[str, Any]]
+) -> List[Dict[str, Any]]:
+    by_id = {s["paramId"]: s for s in saved if s.get("paramId")}
+    by_code = {s["paramCode"]: s for s in saved if s.get("paramCode")}
+    out: List[Dict[str, Any]] = []
+    for p in templates:
+        pid = p.get("id") or ""
+        code = p.get("paramCode") or ""
+        hit = by_id.get(pid) or (by_code.get(code) if code else None)
+        out.append(
+            {
+                "paramId": pid,
+                "paramName": p.get("paramName", ""),
+                "paramCode": code,
+                "paramUnit": p.get("paramUnit", ""),
+                "paramType": p.get("paramType", "text"),
+                "paramValue": (hit.get("paramValue") if hit else "") or "",
+                "requiredFlag": p.get("requiredFlag", "0"),
+            }
+        )
+    return out
+
+
+def list_results_by_task(
+    db: Session, task_id: str, tenant_no: str
+) -> List[Dict[str, Any]]:
+    """返回任务结果填写表单：项目参数模板 + 已保存值（无模板时仅返回已保存记录）。"""
+    saved = _list_saved_results_by_task(db, task_id, tenant_no)
+    task = get_task(db, task_id, tenant_no=tenant_no)
+    if not task:
+        return saved
+    item_source, item_id = _resolve_task_item_ref(db, task)
+    if not item_source or not item_id:
+        return saved
+    templates = param_svc.list_params_by_item(db, item_source, item_id, tenant_no)
+    if not templates:
+        return saved
+    return _merge_saved_into_template(templates, saved)
+
+
 def save_results_for_task(
     db: Session,
     task_id: str,
@@ -44,6 +115,9 @@ def save_results_for_task(
     tenant_no: str,
     user_id: Optional[str] = None,
 ) -> List[Dict[str, Any]]:
+    task = get_task(db, task_id, tenant_no=tenant_no)
+    if not task:
+        raise ValueError("任务不存在")
     now = datetime.now()
     db.query(BizQcTaskResult).filter(
         BizQcTaskResult.task_id == task_id,

backend/app/crud/qc_work_order.py

@@ -13,7 +13,11 @@ from app.crud import qc_lab_test_item as lab_item_svc
 from app.crud import qc_task as task_svc
 from app.crud import qc_third_party_item as third_item_svc
 from app.crud.qc_third_party_company import get_company
+from app.crud.collection_protocol import urls_to_attachment_list
+from app.crud import sample_reception as sample_reception_svc
+from app.crud import sample_reception_check_result as check_result_svc
 from app.models.models import (
+    BizCellSampleType,
     BizCollectionProtocol,
     BizProtocolSample,
     BizQcLabTestItem,
@@ -24,6 +28,22 @@ from app.models.models import (
     BizQcWorkOrderThirdItem,
     BizSampleReception,
 )
+from app.models.qc_param import BizQcTaskResult
+from app.models.research_donation import BizResearchDonation, BizResearchDonationSample
+
+_RECEPTION_SOURCE_LABELS = {
+    "cell_storage": "细胞存储",
+    "research_donation": "科研捐赠",
+}
+_QC_BIZ_TYPE_LABELS = {
+    "inbound": "入库质检",
+    "outbound": "出库质检",
+}
+_ACCEPT_CONCLUSION_LABELS = {"js": "接收", "bj": "拒收"}
+_WORK_ORDER_STATUS_LABELS = {"dpz": "待配置", "ypz": "已配置", "ywc": "已完成"}
+_TASK_TYPE_LABELS = {"sysz": "实验室自检", "dsfjc": "第三方检测"}
+_TASK_STATUS_LABELS = {"dzx": "待执行", "jxz": "进行中", "ywc": "已完成"}
+_CONCLUSION_LABELS = {"tg": "通过", "btg": "不通过"}
 
 
 def _dec(v: Any) -> Optional[float]:
@@ -36,6 +56,126 @@ def _fmt_dt(t: Optional[datetime]) -> str:
     return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""
 
 
+def _reception_source_label(source: Optional[str]) -> str:
+    key = (source or "").strip() or "cell_storage"
+    return _RECEPTION_SOURCE_LABELS.get(key, key)
+
+
+def _qc_biz_type_label(biz_type: Optional[str]) -> str:
+    key = (biz_type or "").strip() or "inbound"
+    return _QC_BIZ_TYPE_LABELS.get(key, key)
+
+
+def _work_order_qc_biz_type(wo: BizQcWorkOrder) -> str:
+    return getattr(wo, "qc_biz_type", None) or "inbound"
+
+
+def _accept_conclusion_label(v: Optional[str]) -> str:
+    key = (v or "").strip()
+    return _ACCEPT_CONCLUSION_LABELS.get(key, key)
+
+
+def _load_sample_type_name_map(db: Session, type_ids: List[str]) -> Dict[str, str]:
+    ids = [x for x in type_ids if x]
+    if not ids:
+        return {}
+    return {
+        t.id: t.type_name or ""
+        for t in db.query(BizCellSampleType)
+        .filter(BizCellSampleType.id.in_(ids), BizCellSampleType.del_flag == 0)
+        .all()
+    }
+
+
+def _sample_type_name_for_reception(
+    db: Session,
+    rec: BizSampleReception,
+    type_map: Dict[str, str],
+    *,
+    protocol_samples: Dict[str, BizProtocolSample],
+    donation_samples: Dict[str, BizResearchDonationSample],
+) -> str:
+    if (rec.reception_source or "cell_storage") == "research_donation":
+        sample = donation_samples.get(rec.donation_sample_id or "")
+        tid = sample.sample_type_id if sample else ""
+    else:
+        sample = protocol_samples.get(rec.protocol_sample_id or "")
+        tid = sample.sample_type_id if sample else ""
+    return type_map.get(tid or "", "")
+
+
+def _enrich_work_order_list_items(
+    db: Session, items: List[Dict[str, Any]]
+) -> List[Dict[str, Any]]:
+    if not items:
+        return items
+    rec_ids = [x["receptionId"] for x in items if x.get("receptionId")]
+    if not rec_ids:
+        return items
+    rec_map = {
+        r.id: r
+        for r in db.query(BizSampleReception)
+        .filter(BizSampleReception.id.in_(rec_ids), BizSampleReception.del_flag == 0)
+        .all()
+    }
+    cell_sample_ids = [
+        r.protocol_sample_id
+        for r in rec_map.values()
+        if (r.reception_source or "cell_storage") != "research_donation" and r.protocol_sample_id
+    ]
+    donation_sample_ids = [
+        r.donation_sample_id
+        for r in rec_map.values()
+        if (r.reception_source or "") == "research_donation" and r.donation_sample_id
+    ]
+    protocol_samples: Dict[str, BizProtocolSample] = {}
+    if cell_sample_ids:
+        protocol_samples = {
+            s.id: s
+            for s in db.query(BizProtocolSample)
+            .filter(BizProtocolSample.id.in_(cell_sample_ids), BizProtocolSample.del_flag == 0)
+            .all()
+        }
+    donation_samples: Dict[str, BizResearchDonationSample] = {}
+    if donation_sample_ids:
+        donation_samples = {
+            s.id: s
+            for s in db.query(BizResearchDonationSample)
+            .filter(
+                BizResearchDonationSample.id.in_(donation_sample_ids),
+                BizResearchDonationSample.del_flag == 0,
+            )
+            .all()
+        }
+    type_ids = list(
+        {
+            s.sample_type_id
+            for s in list(protocol_samples.values()) + list(donation_samples.values())
+            if s.sample_type_id
+        }
+    )
+    type_map = _load_sample_type_name_map(db, type_ids)
+    for item in items:
+        rec = rec_map.get(item.get("receptionId") or "")
+        source = item.get("receptionSource") or (rec.reception_source if rec else "cell_storage")
+        item["receptionSourceLabel"] = _reception_source_label(source)
+        if not rec:
+            item.setdefault("receptionTime", "")
+            item.setdefault("handlerName", "")
+            item.setdefault("sampleTypeName", "")
+            continue
+        item["receptionTime"] = _fmt_dt(rec.reception_time)
+        item["handlerName"] = rec.handler_name or ""
+        item["sampleTypeName"] = _sample_type_name_for_reception(
+            db,
+            rec,
+            type_map,
+            protocol_samples=protocol_samples,
+            donation_samples=donation_samples,
+        )
+    return items
+
+
 def _gen_daily_no(db: Session, tenant_no: str, prefix: str) -> str:
     day = datetime.now().strftime("%Y%m%d")
     like = f"{prefix}{day}%"
@@ -59,6 +199,15 @@ def _wo_to_dict(wo: BizQcWorkOrder) -> Dict[str, Any]:
         "collectionProtocolId": wo.collection_protocol_id,
         "protocolNo": wo.protocol_no or "",
         "sampleCode": wo.sample_code or "",
+        "receptionSource": getattr(wo, "reception_source", "cell_storage") or "cell_storage",
+        "receptionSourceLabel": _reception_source_label(
+            getattr(wo, "reception_source", "cell_storage") or "cell_storage"
+        ),
+        "qcBizType": _work_order_qc_biz_type(wo),
+        "qcBizTypeLabel": _qc_biz_type_label(_work_order_qc_biz_type(wo)),
+        "receptionTime": "",
+        "handlerName": "",
+        "sampleTypeName": "",
         "workOrderStatus": wo.work_order_status,
         "hasLab": wo.has_lab == "1",
         "hasThirdParty": wo.has_third_party == "1",
@@ -74,7 +223,7 @@ def _wo_to_dict(wo: BizQcWorkOrder) -> Dict[str, Any]:
 def create_work_order_from_reception(
     db: Session,
     *,
-    reception: BizSampleReception,
+    reception,
     tenant_no: str,
     user_id: Optional[str] = None,
     create_dept_id: str = "",
@@ -93,44 +242,94 @@ def create_work_order_from_reception(
     )
     if exists:
         return exists
-    protocol_no = ""
-    p = (
-        db.query(BizCollectionProtocol)
-        .filter(
-            BizCollectionProtocol.id == reception.collection_protocol_id,
-            BizCollectionProtocol.del_flag == 0,
+
+    source = getattr(reception, "reception_source", "cell_storage") or "cell_storage"
+
+    if source == "research_donation":
+        protocol_no = ""
+        d = (
+            db.query(BizResearchDonation)
+            .filter(
+                BizResearchDonation.id == (reception.research_donation_id or ""),
+                BizResearchDonation.del_flag == 0,
+            )
+            .first()
+        )
+        if d:
+            protocol_no = d.donation_no or ""
+        sample_code = ""
+        ds = (
+            db.query(BizResearchDonationSample)
+            .filter(
+                BizResearchDonationSample.id == (reception.donation_sample_id or ""),
+                BizResearchDonationSample.del_flag == 0,
+            )
+            .first()
+        )
+        if ds:
+            sample_code = ds.sample_code or ""
+        wo = BizQcWorkOrder(
+            id=generate_uuid(),
+            work_order_no=_gen_daily_no(db, tenant_no, "ZJ"),
+            reception_id=reception.id,
+            protocol_sample_id=reception.donation_sample_id or "",
+            collection_protocol_id=reception.research_donation_id or "",
+            protocol_no=protocol_no or None,
+            sample_code=sample_code or None,
+            reception_source="research_donation",
+            qc_biz_type="inbound",
+            work_order_status="dpz",
+            has_lab="0",
+            has_third_party="0",
+            tenant_no=tenant_no,
+            create_by=user_id,
+            create_dept_id=create_dept_id or "",
+            update_by=user_id,
+            create_time=datetime.now(),
+            update_time=datetime.now(),
+            del_flag=0,
+        )
+    else:
+        protocol_no = ""
+        p = (
+            db.query(BizCollectionProtocol)
+            .filter(
+                BizCollectionProtocol.id == reception.collection_protocol_id,
+                BizCollectionProtocol.del_flag == 0,
+            )
+            .first()
+        )
+        if p:
+            protocol_no = p.protocol_no or ""
+        sample_code = ""
+        s = (
+            db.query(BizProtocolSample)
+            .filter(BizProtocolSample.id == reception.protocol_sample_id, BizProtocolSample.del_flag == 0)
+            .first()
+        )
+        if s:
+            sample_code = s.sample_code or ""
+        wo = BizQcWorkOrder(
+            id=generate_uuid(),
+            work_order_no=_gen_daily_no(db, tenant_no, "ZJ"),
+            reception_id=reception.id,
+            protocol_sample_id=reception.protocol_sample_id,
+            collection_protocol_id=reception.collection_protocol_id,
+            protocol_no=protocol_no or None,
+            sample_code=sample_code or None,
+            reception_source="cell_storage",
+            qc_biz_type="inbound",
+            work_order_status="dpz",
+            has_lab="0",
+            has_third_party="0",
+            tenant_no=tenant_no,
+            create_by=user_id,
+            create_dept_id=create_dept_id or "",
+            update_by=user_id,
+            create_time=datetime.now(),
+            update_time=datetime.now(),
+            del_flag=0,
         )
-        .first()
-    )
-    if p:
-        protocol_no = p.protocol_no or ""
-    sample_code = ""
-    s = (
-        db.query(BizProtocolSample)
-        .filter(BizProtocolSample.id == reception.protocol_sample_id, BizProtocolSample.del_flag == 0)
-        .first()
-    )
-    if s:
-        sample_code = s.sample_code or ""
-    wo = BizQcWorkOrder(
-        id=generate_uuid(),
-        work_order_no=_gen_daily_no(db, tenant_no, "ZJ"),
-        reception_id=reception.id,
-        protocol_sample_id=reception.protocol_sample_id,
-        collection_protocol_id=reception.collection_protocol_id,
-        protocol_no=protocol_no or None,
-        sample_code=sample_code or None,
-        work_order_status="dpz",
-        has_lab="0",
-        has_third_party="0",
-        tenant_no=tenant_no,
-        create_by=user_id,
-        create_dept_id=create_dept_id or "",
-        update_by=user_id,
-        create_time=datetime.now(),
-        update_time=datetime.now(),
-        del_flag=0,
-    )
     db.add(wo)
     db.flush()
     return wo
@@ -174,7 +373,8 @@ def list_work_orders(
         .limit(page_size)
         .all()
     )
-    return [_wo_to_dict(x) for x in rows], int(total)
+    items = [_wo_to_dict(x) for x in rows]
+    return _enrich_work_order_list_items(db, items), int(total)
 
 
 def get_work_order(
@@ -231,6 +431,49 @@ def _third_lines(db: Session, work_order_id: str) -> List[Dict[str, Any]]:
     ]
 
 
+def get_qc_info_map_for_samples(
+    db: Session,
+    sample_ids: List[str],
+    tenant_no: str,
+) -> Dict[str, Dict[str, Any]]:
+    """按协议/捐赠样本 ID 批量返回质检工单及检测任务摘要，供协议/捐赠查看弹窗使用。"""
+    if not sample_ids or not tenant_no:
+        return {}
+    wos = (
+        db.query(BizQcWorkOrder)
+        .filter(
+            BizQcWorkOrder.protocol_sample_id.in_(sample_ids),
+            BizQcWorkOrder.tenant_no == tenant_no,
+            BizQcWorkOrder.del_flag == 0,
+        )
+        .all()
+    )
+    out: Dict[str, Dict[str, Any]] = {}
+    for wo in wos:
+        sid = wo.protocol_sample_id
+        if not sid:
+            continue
+        tasks = task_svc.list_tasks_by_work_order(db, wo.id, tenant_no)
+        for t in tasks:
+            tt = (t.get("taskType") or "").strip()
+            ts = (t.get("taskStatus") or "").strip()
+            tc = (t.get("conclusion") or "").strip()
+            t["taskTypeLabel"] = _TASK_TYPE_LABELS.get(tt, tt)
+            t["taskStatusLabel"] = _TASK_STATUS_LABELS.get(ts, ts)
+            t["conclusionLabel"] = t.get("conclusionLabel") or _CONCLUSION_LABELS.get(tc, tc or "—")
+        wo_status = (wo.work_order_status or "").strip()
+        biz = _work_order_qc_biz_type(wo)
+        out[sid] = {
+            **_wo_to_dict(wo),
+            "workOrderStatusLabel": _WORK_ORDER_STATUS_LABELS.get(wo_status, wo_status),
+            "qcBizTypeLabel": _qc_biz_type_label(biz),
+            "labItems": _lab_lines(db, wo.id),
+            "thirdPartyItems": _third_lines(db, wo.id),
+            "tasks": tasks,
+        }
+    return out
+
+
 def get_work_order_detail(
     db: Session, row_id: str, tenant_no: str
 ) -> Dict[str, Any]:
@@ -238,22 +481,83 @@ def get_work_order_detail(
     if not wo:
         raise ValueError("工单不存在")
     out = _wo_to_dict(wo)
-    rec = (
-        db.query(BizSampleReception)
-        .filter(BizSampleReception.id == wo.reception_id, BizSampleReception.del_flag == 0)
-        .first()
+    enriched = _enrich_work_order_list_items(db, [out])[0]
+    out.update(
+        {
+            "receptionTime": enriched.get("receptionTime", ""),
+            "handlerName": enriched.get("handlerName", ""),
+            "sampleTypeName": enriched.get("sampleTypeName", ""),
+            "receptionSourceLabel": enriched.get("receptionSourceLabel", ""),
+        }
     )
-    out["reception"] = {
-        "receptionTime": _fmt_dt(rec.reception_time) if rec else "",
-        "handlerName": (rec.handler_name if rec else "") or "",
-        "acceptConclusion": (rec.accept_conclusion if rec else "") or "",
-    }
+    reception_detail = sample_reception_svc.get_reception_by_id(db, wo.reception_id, tenant_no)
+    if reception_detail:
+        out["reception"] = reception_detail
+    else:
+        rec = (
+            db.query(BizSampleReception)
+            .filter(BizSampleReception.id == wo.reception_id, BizSampleReception.del_flag == 0)
+            .first()
+        )
+        cr = (
+            check_result_svc.get_check_results_for_reception(db, rec, tenant_no) if rec else []
+        )
+        out["reception"] = {
+            "receptionTime": _fmt_dt(rec.reception_time) if rec else "",
+            "handlerName": (rec.handler_name if rec else "") or "",
+            "acceptConclusion": (rec.accept_conclusion if rec else "") or "",
+            "acceptConclusionLabel": _accept_conclusion_label(
+                rec.accept_conclusion if rec else ""
+            ),
+            "sourceType": wo.reception_source or "cell_storage",
+            "sourceTypeLabel": _reception_source_label(wo.reception_source),
+            "sampleTypeName": enriched.get("sampleTypeName", ""),
+            "checkResults": cr,
+            "receiveAttachments": (rec.receive_attachments if rec else "") or "",
+            "receiveAttachmentFiles": urls_to_attachment_list(
+                rec.receive_attachments if rec else None, db, tenant_no
+            ),
+        }
     out["labItems"] = _lab_lines(db, wo.id)
     out["thirdPartyItems"] = _third_lines(db, wo.id)
     out["tasks"] = task_svc.list_tasks_by_work_order(db, wo.id, tenant_no)
     return out
 
 
+def _soft_delete_pending_work_order_tasks(
+    db: Session, work_order_id: str, tenant_no: str
+) -> None:
+    """调整检测前软删待执行任务；存在进行中/已完成任务时不允许调整。"""
+    tasks = (
+        db.query(BizQcTask)
+        .filter(
+            BizQcTask.work_order_id == work_order_id,
+            BizQcTask.tenant_no == tenant_no,
+            BizQcTask.del_flag == 0,
+        )
+        .all()
+    )
+    if any(t.task_status in ("jxz", "ywc") for t in tasks):
+        raise ValueError("存在进行中或已完成的质检任务，无法调整检测项目")
+    if not tasks:
+        return
+    now = datetime.now()
+    task_ids = [t.id for t in tasks]
+    if task_ids:
+        db.query(BizQcTaskResult).filter(
+            BizQcTaskResult.task_id.in_(task_ids),
+            BizQcTaskResult.tenant_no == tenant_no,
+            BizQcTaskResult.del_flag == 0,
+        ).update(
+            {"del_flag": 1, "update_time": now},
+            synchronize_session="fetch",
+        )
+    for t in tasks:
+        t.del_flag = 1
+        t.update_time = now
+        db.add(t)
+
+
 def configure_work_order(
     db: Session,
     work_order_id: str,
@@ -269,15 +573,12 @@ def configure_work_order(
     wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
     if not wo:
         raise ValueError("工单不存在")
-    if wo.work_order_status != "dpz":
-        raise ValueError("仅待配置工单可配置检测项目")
-    task_cnt = (
-        db.query(BizQcTask)
-        .filter(BizQcTask.work_order_id == wo.id, BizQcTask.del_flag == 0)
-        .count()
-    )
-    if task_cnt:
-        raise ValueError("工单已存在任务，无法重新配置")
+    if wo.work_order_status == "ywc":
+        raise ValueError("工单已完成，无法调整检测项目")
+    if wo.work_order_status not in ("dpz", "ypz"):
+        raise ValueError("当前工单状态不可配置检测项目")
+    if wo.work_order_status == "ypz":
+        _soft_delete_pending_work_order_tasks(db, wo.id, tenant_no)
 
     if not has_lab and not has_third_party:
         raise ValueError("请至少选择一种检测方式")

backend/app/crud/research_donation.py

@@ -588,10 +588,22 @@ def get_donation_detail(
     s_rows = list_samples(db, donation_id, tenant_no)
     transport_rows = list_sample_transports(db, donation_id, tenant_no)
     transport_by_sample = {t.sample_id: _transport_to_dict(t) for t in transport_rows}
+    from app.crud import sample_reception as sample_reception_svc
+    from app.crud import qc_work_order as qc_work_order_svc
+
+    sample_ids = [s.id for s in s_rows]
+    reception_by_sample = sample_reception_svc.get_receptions_map_for_donation_samples(
+        db, sample_ids, tenant_no
+    )
+    qc_by_sample = qc_work_order_svc.get_qc_info_map_for_samples(
+        db, sample_ids, tenant_no
+    )
     samples_out: List[Dict[str, Any]] = []
     for s in s_rows:
         item = _sample_to_dict(s)
         item["transport"] = transport_by_sample.get(s.id)
+        item["sampleReception"] = reception_by_sample.get(s.id)
+        item["qcInfo"] = qc_by_sample.get(s.id)
         samples_out.append(item)
     out["samples"] = samples_out
 
@@ -1246,6 +1258,7 @@ def convert_from_storage(
                 id=generate_uuid(),
                 research_donation_id=did,
                 sample_id=new_sample.id,
+                sample_code=ss.sample_code or None,
                 transport_type=src_t.transport_type,
                 express_company_id=src_t.express_company_id,
                 waybill_no=src_t.waybill_no,
@@ -1484,6 +1497,7 @@ def convert_to_storage(
                 id=generate_uuid(),
                 collection_protocol_id=pid,
                 sample_id=new_sample.id,
+                sample_code=ss.sample_code or None,
                 transport_type=src_t.transport_type,
                 express_company_id=src_t.express_company_id,
                 waybill_no=src_t.waybill_no,
@@ -1686,7 +1700,7 @@ def _create_samples(
         # transport
         transport_raw = raw.get("transport")
         if isinstance(transport_raw, dict):
-            _create_transport(db, donation_id, obj.id, tenant_no, user_id, transport_raw, idx, now)
+            _create_transport(db, donation_id, obj.id, tenant_no, user_id, transport_raw, idx, now, sample_code=obj.sample_code)
 
     return result
 
@@ -1700,6 +1714,7 @@ def _create_transport(
     raw: dict,
     order_num: int,
     now: datetime,
+    sample_code: Optional[str] = None,
 ) -> Optional[BizResearchDonationSampleTransport]:
     tnorm = _normalize_fields(raw)
     transport_type = (tnorm.get("transport_type") or "").strip()
@@ -1720,6 +1735,7 @@ def _create_transport(
         id=generate_uuid(),
         research_donation_id=donation_id,
         sample_id=sample_id,
+        sample_code=sample_code or None,
         transport_type=transport_type,
         express_company_id=(tnorm.get("express_company_id") or "").strip() or None,
         waybill_no=(tnorm.get("waybill_no") or "").strip() or None,

backend/app/crud/sample_reception.py

@@ -20,6 +20,20 @@ from app.models.models import (
     BizSampleReception,
     SysUser,
 )
+from app.models.research_donation import (
+    BizResearchDonation,
+    BizResearchDonationParty,
+    BizResearchDonationSample,
+    BizResearchDonationSampleTransport,
+)
+
+# 科研捐赠表默认 general_ci，样本接收等为 unicode_ci，跨表比较需统一
+_UNICODE_CI = "utf8mb4_unicode_ci"
+
+
+def _donation_sample_id_col(col):
+    """捐赠样本 ID 与 biz_sample_reception.donation_sample_id 比较时统一排序规则。"""
+    return col.collate(_UNICODE_CI)
 
 
 def _fmt_dt(t: Optional[datetime]) -> str:
@@ -90,6 +104,152 @@ def _sample_detail_rows_query(db: Session):
     )
 
 
+def _donation_sample_detail_rows_query(db: Session):
+    """捐赠样本 + 捐赠单 + 当事人 + 运输（四表 JOIN）。"""
+    return (
+        db.query(
+            BizResearchDonationSample,
+            BizResearchDonation,
+            BizResearchDonationParty,
+            BizResearchDonationSampleTransport,
+        )
+        .select_from(BizResearchDonationSample)
+        .join(
+            BizResearchDonation,
+            and_(
+                BizResearchDonation.id == BizResearchDonationSample.research_donation_id,
+                BizResearchDonation.del_flag == 0,
+            ),
+        )
+        .outerjoin(
+            BizResearchDonationParty,
+            and_(
+                BizResearchDonationParty.research_donation_id == BizResearchDonation.id,
+                BizResearchDonationParty.del_flag == 0,
+            ),
+        )
+        .outerjoin(
+            BizResearchDonationSampleTransport,
+            and_(
+                BizResearchDonationSampleTransport.sample_id == BizResearchDonationSample.id,
+                BizResearchDonationSampleTransport.del_flag == 0,
+            ),
+        )
+    )
+
+
+def _donation_sample_pending_query(db: Session, tenant_no: str, keyword: Optional[str] = None):
+    """待接收：捐赠样本且尚无有效接收记录。"""
+    received_donation_ids = (
+        db.query(BizSampleReception.donation_sample_id)
+        .filter(
+            BizSampleReception.tenant_no == tenant_no,
+            BizSampleReception.del_flag == 0,
+            BizSampleReception.reception_source == "research_donation",
+        )
+        .subquery()
+    )
+    q = _donation_sample_detail_rows_query(db).filter(
+        BizResearchDonationSample.del_flag == 0,
+        BizResearchDonationSample.tenant_no == tenant_no,
+        BizResearchDonation.tenant_no == tenant_no,
+        BizResearchDonation.donation_status.in_(["sxz", "ygq"]),
+        ~_donation_sample_id_col(BizResearchDonationSample.id).in_(received_donation_ids),
+    )
+    if keyword and keyword.strip():
+        kw = f"%{keyword.strip()}%"
+        q = q.filter(
+            or_(
+                BizResearchDonation.donation_no.like(kw),
+                BizResearchDonationSample.sample_code.like(kw),
+                BizResearchDonationParty.mother_name.like(kw),
+                BizResearchDonationParty.newborn_name.like(kw),
+            )
+        )
+    return q
+
+
+def _donation_rows_to_sample_list_items(
+    db: Session,
+    rows: list,
+    tenant_no: str,
+    *,
+    reception_map: Optional[Dict[str, BizSampleReception]] = None,
+) -> List[Dict[str, Any]]:
+    """将捐赠样本行转为统一列表项字典（字段名与细胞存储保持一致）。"""
+    type_ids = list({s.sample_type_id for s, *_ in rows if s.sample_type_id})
+    type_map: Dict[str, str] = {}
+    if type_ids:
+        for t in (
+            db.query(BizCellSampleType)
+            .filter(
+                BizCellSampleType.id.in_(type_ids),
+                BizCellSampleType.del_flag == 0,
+            )
+            .all()
+        ):
+            type_map[t.id] = t.type_name or ""
+
+    express_ids = list(
+        {tr.express_company_id for *_, tr in rows if tr and tr.express_company_id}
+    )
+    express_map: Dict[str, str] = {}
+    if express_ids:
+        for e in (
+            db.query(BizExpressCompany)
+            .filter(BizExpressCompany.id.in_(express_ids), BizExpressCompany.del_flag == 0)
+            .all()
+        ):
+            express_map[e.id] = e.express_name or ""
+
+    reception_ids = (
+        [reception_map[s.id].id for s, *_ in rows if s.id in reception_map]
+        if reception_map
+        else []
+    )
+    results_map = (
+        check_result_svc.list_results_by_reception_ids(db, reception_ids, tenant_no)
+        if reception_ids
+        else {}
+    )
+
+    items: List[Dict[str, Any]] = []
+    for sample, donation, party, transport in rows:
+        item: Dict[str, Any] = {
+            "protocolSampleId": sample.id,
+            "collectionProtocolId": donation.id,
+            "protocolNo": donation.donation_no or "",
+            "sampleCode": sample.sample_code or "",
+            "sampleTypeId": sample.sample_type_id or "",
+            "sampleTypeName": type_map.get(sample.sample_type_id or "", ""),
+            "collectAmount": _dec(sample.collect_amount),
+            "motherName": (party.mother_name if party else "") or "",
+            "newbornName": (party.newborn_name if party else "") or "",
+            "sourceType": "research_donation",
+            "sourceTypeLabel": "科研捐赠",
+            "etaTime": _fmt_dt(transport.eta_time) if transport and transport.eta_time else "",
+            "transportType": (transport.transport_type if transport else "") or "",
+            "waybillNo": (transport.waybill_no if transport else "") or "",
+            "expressCompanyId": (transport.express_company_id if transport else "") or "",
+            "expressCompanyName": express_map.get(
+                (transport.express_company_id if transport else "") or "", ""
+            ),
+            "carrierName": (transport.carrier_name if transport else "") or "",
+            "carrierPhone": (transport.carrier_phone if transport else "") or "",
+        }
+        if reception_map and sample.id in reception_map:
+            rec = reception_map[sample.id]
+            cr = results_map.get(rec.id)
+            if not cr:
+                cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
+            item.update(_reception_to_dict(rec, check_results=cr))
+            item["receiveAttachmentFiles"] = urls_to_attachment_list(
+                rec.receive_attachments, db, tenant_no
+            )
+        items.append(item)
+    return items
+
+
 def _sample_pending_query(db: Session, tenant_no: str, keyword: Optional[str] = None):
     """待接收：协议样本且尚无有效接收记录。"""
     received_ids = (
@@ -184,6 +344,9 @@ def _rows_to_sample_list_items(
                 (transport.express_company_id if transport else "") or "", ""
             ),
             "carrierName": (transport.carrier_name if transport else "") or "",
+            "carrierPhone": (transport.carrier_phone if transport else "") or "",
+            "sourceType": "cell_storage",
+            "sourceTypeLabel": "细胞存储",
         }
         if reception_map and sample.id in reception_map:
             rec = reception_map[sample.id]
@@ -205,24 +368,53 @@ def list_pending_protocol_samples(
     page_size: int = 10,
     tenant_no: Optional[str] = None,
     keyword: Optional[str] = None,
+    source_type: Optional[str] = None,
 ) -> Tuple[List[Dict[str, Any]], int]:
     if not tenant_no:
         return [], 0
-    q_base = _sample_pending_query(db, tenant_no, keyword)
-    total = q_base.order_by(None).with_entities(func.count(BizProtocolSample.id)).scalar() or 0
+
+    # 收集两种来源的样本
+    all_items: List[Dict[str, Any]] = []
+
+    if not source_type or source_type == "cell_storage":
+        q_base = _sample_pending_query(db, tenant_no, keyword)
+        cell_total = q_base.order_by(None).with_entities(func.count(BizProtocolSample.id)).scalar() or 0
+        cell_rows = (
+            q_base.order_by(
+                BizProtocolSampleTransport.eta_time.is_(None),
+                BizProtocolSampleTransport.eta_time.asc(),
+                BizProtocolSample.create_time.asc(),
+            )
+            .all()
+        )
+        all_items.extend(_rows_to_sample_list_items(db, cell_rows, tenant_no))
+    else:
+        cell_total = 0
+
+    if not source_type or source_type == "research_donation":
+        q_donation = _donation_sample_pending_query(db, tenant_no, keyword)
+        donation_total = q_donation.order_by(None).with_entities(func.count(BizResearchDonationSample.id)).scalar() or 0
+        donation_rows = (
+            q_donation.order_by(
+                BizResearchDonationSampleTransport.eta_time.is_(None),
+                BizResearchDonationSampleTransport.eta_time.asc(),
+                BizResearchDonationSample.create_time.asc(),
+            )
+            .all()
+        )
+        all_items.extend(_donation_rows_to_sample_list_items(db, donation_rows, tenant_no))
+    else:
+        donation_total = 0
+
+    # 统一按 etaTime 排序（空值排后）
+    all_items.sort(key=lambda x: (not x.get("etaTime"), x.get("etaTime") or ""))
+
+    total = cell_total + donation_total
     page = max(1, page)
     page_size = min(max(1, page_size), 100)
-    rows = (
-        q_base.order_by(
-            BizProtocolSampleTransport.eta_time.is_(None),
-            BizProtocolSampleTransport.eta_time.asc(),
-            BizProtocolSample.create_time.asc(),
-        )
-        .offset((page - 1) * page_size)
-        .limit(page_size)
-        .all()
-    )
-    return _rows_to_sample_list_items(db, rows, tenant_no), int(total)
+    start = (page - 1) * page_size
+    page_items = all_items[start:start + page_size]
+    return page_items, int(total)
 
 
 def list_received_protocol_samples(
@@ -232,16 +424,24 @@ def list_received_protocol_samples(
     page_size: int = 10,
     tenant_no: Optional[str] = None,
     keyword: Optional[str] = None,
+    source_type: Optional[str] = None,
 ) -> Tuple[List[Dict[str, Any]], int]:
     if not tenant_no:
         return [], 0
+
+    # 查询所有接收记录，按 source_type 过滤
     q_rec = db.query(BizSampleReception).filter(
         BizSampleReception.tenant_no == tenant_no,
         BizSampleReception.del_flag == 0,
     )
+    if source_type:
+        q_rec = q_rec.filter(BizSampleReception.reception_source == source_type)
+
+    # 关键词搜索
     if keyword and keyword.strip():
         kw = f"%{keyword.strip()}%"
-        sample_ids_kw = (
+        # 细胞存储样本关键词
+        cell_sample_ids_kw = (
             _sample_detail_rows_query(db)
             .with_entities(BizProtocolSample.id)
             .filter(
@@ -257,10 +457,28 @@ def list_received_protocol_samples(
             )
             .subquery()
         )
+        # 捐赠样本关键词
+        donation_sample_ids_kw = (
+            _donation_sample_detail_rows_query(db)
+            .with_entities(_donation_sample_id_col(BizResearchDonationSample.id))
+            .filter(
+                BizResearchDonationSample.del_flag == 0,
+                BizResearchDonationSample.tenant_no == tenant_no,
+                BizResearchDonation.tenant_no == tenant_no,
+                or_(
+                    BizResearchDonation.donation_no.like(kw),
+                    BizResearchDonationSample.sample_code.like(kw),
+                    BizResearchDonationParty.mother_name.like(kw),
+                    BizResearchDonationParty.newborn_name.like(kw),
+                ),
+            )
+            .subquery()
+        )
         q_rec = q_rec.filter(
             or_(
                 BizSampleReception.handler_name.like(kw),
-                BizSampleReception.protocol_sample_id.in_(sample_ids_kw),
+                BizSampleReception.protocol_sample_id.in_(cell_sample_ids_kw),
+                BizSampleReception.donation_sample_id.in_(donation_sample_ids_kw),
             )
         )
 
@@ -276,19 +494,38 @@ def list_received_protocol_samples(
     if not receptions:
         return [], int(total)
 
-    sample_ids = [r.protocol_sample_id for r in receptions]
-    reception_map = {r.protocol_sample_id: r for r in receptions}
+    # 按 reception_source 分组查询详情
+    cell_receptions = [r for r in receptions if r.reception_source == "cell_storage"]
+    donation_receptions = [r for r in receptions if r.reception_source == "research_donation"]
 
-    rows = (
-        _sample_detail_rows_query(db)
-        .filter(BizProtocolSample.id.in_(sample_ids))
-        .all()
-    )
-    row_by_sample = {row[0].id: row for row in rows}
-    ordered_rows = [row_by_sample[sid] for sid in sample_ids if sid in row_by_sample]
-    items = _rows_to_sample_list_items(
-        db, ordered_rows, tenant_no, reception_map=reception_map
-    )
+    items: List[Dict[str, Any]] = []
+
+    if cell_receptions:
+        cell_sample_ids = [r.protocol_sample_id for r in cell_receptions]
+        cell_reception_map = {r.protocol_sample_id: r for r in cell_receptions}
+        cell_rows = (
+            _sample_detail_rows_query(db)
+            .filter(BizProtocolSample.id.in_(cell_sample_ids))
+            .all()
+        )
+        cell_row_by_sample = {row[0].id: row for row in cell_rows}
+        cell_ordered = [cell_row_by_sample[sid] for sid in cell_sample_ids if sid in cell_row_by_sample]
+        items.extend(_rows_to_sample_list_items(db, cell_ordered, tenant_no, reception_map=cell_reception_map))
+
+    if donation_receptions:
+        donation_sample_ids = [r.donation_sample_id for r in donation_receptions]
+        donation_reception_map = {r.donation_sample_id: r for r in donation_receptions}
+        donation_rows = (
+            _donation_sample_detail_rows_query(db)
+            .filter(_donation_sample_id_col(BizResearchDonationSample.id).in_(donation_sample_ids))
+            .all()
+        )
+        donation_row_by_sample = {row[0].id: row for row in donation_rows}
+        donation_ordered = [donation_row_by_sample[sid] for sid in donation_sample_ids if sid in donation_row_by_sample]
+        items.extend(_donation_rows_to_sample_list_items(db, donation_ordered, tenant_no, reception_map=donation_reception_map))
+
+    # 按 reception_time DESC 排序
+    items.sort(key=lambda x: x.get("receptionTime") or "", reverse=True)
     return items, int(total)
 
 
@@ -306,6 +543,36 @@ def get_reception_by_id(
     )
     if not rec:
         return None
+    if (rec.reception_source or "cell_storage") == "research_donation":
+        dsid = rec.donation_sample_id or ""
+        if not dsid:
+            cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
+            out = _reception_to_dict(rec, check_results=cr)
+            out["sourceType"] = "research_donation"
+            out["sourceTypeLabel"] = "科研捐赠"
+            out["receiveAttachmentFiles"] = urls_to_attachment_list(
+                rec.receive_attachments, db, tenant_no
+            )
+            return out
+        rows = (
+            _donation_sample_detail_rows_query(db)
+            .filter(_donation_sample_id_col(BizResearchDonationSample.id) == dsid)
+            .all()
+        )
+        if not rows:
+            cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
+            out = _reception_to_dict(rec, check_results=cr)
+            out["sourceType"] = "research_donation"
+            out["sourceTypeLabel"] = "科研捐赠"
+            out["receiveAttachmentFiles"] = urls_to_attachment_list(
+                rec.receive_attachments, db, tenant_no
+            )
+            return out
+        items = _donation_rows_to_sample_list_items(
+            db, rows, tenant_no, reception_map={dsid: rec}
+        )
+        return items[0] if items else None
+
     sample = (
         db.query(BizProtocolSample)
         .filter(
@@ -342,7 +609,10 @@ def get_reception_by_id(
 def create_sample_reception(
     db: Session,
     *,
-    protocol_sample_id: str,
+    reception_source: str = "cell_storage",
+    protocol_sample_id: Optional[str] = None,
+    donation_sample_id: Optional[str] = None,
+    research_donation_id: Optional[str] = None,
     tenant_no: str,
     user_id: Optional[str],
     create_dept_id: str = "",
@@ -357,28 +627,71 @@ def create_sample_reception(
     receive_attachments: Optional[str],
     remark: Optional[str],
 ) -> BizSampleReception:
-    sample = (
-        db.query(BizProtocolSample)
-        .filter(
-            BizProtocolSample.id == protocol_sample_id,
-            BizProtocolSample.tenant_no == tenant_no,
-            BizProtocolSample.del_flag == 0,
+    source = (reception_source or "").strip()
+    if source not in ("cell_storage", "research_donation"):
+        raise ValueError("来源类型无效")
+
+    # 按来源校验样本并获取关联 ID
+    if source == "cell_storage":
+        if not protocol_sample_id:
+            raise ValueError("细胞存储来源必须传入 protocolSampleId")
+        sample = (
+            db.query(BizProtocolSample)
+            .filter(
+                BizProtocolSample.id == protocol_sample_id,
+                BizProtocolSample.tenant_no == tenant_no,
+                BizProtocolSample.del_flag == 0,
+            )
+            .first()
         )
-        .first()
-    )
-    if not sample:
-        raise ValueError("样本不存在")
-    exists = (
-        db.query(BizSampleReception)
-        .filter(
-            BizSampleReception.protocol_sample_id == protocol_sample_id,
-            BizSampleReception.tenant_no == tenant_no,
-            BizSampleReception.del_flag == 0,
+        if not sample:
+            raise ValueError("样本不存在")
+        exists = (
+            db.query(BizSampleReception)
+            .filter(
+                BizSampleReception.protocol_sample_id == protocol_sample_id,
+                BizSampleReception.tenant_no == tenant_no,
+                BizSampleReception.del_flag == 0,
+            )
+            .first()
         )
-        .first()
-    )
-    if exists:
-        raise ValueError("该样本已登记接收，请勿重复操作")
+        if exists:
+            raise ValueError("该样本已登记接收，请勿重复操作")
+        collection_protocol_id = sample.collection_protocol_id
+        _donation_sample_id = None
+        _research_donation_id = None
+        _sample_code = sample.sample_code or ""
+    else:
+        # research_donation
+        if not donation_sample_id:
+            raise ValueError("科研捐赠来源必须传入 donationSampleId")
+        dsample = (
+            db.query(BizResearchDonationSample)
+            .filter(
+                _donation_sample_id_col(BizResearchDonationSample.id) == donation_sample_id,
+                BizResearchDonationSample.tenant_no == tenant_no,
+                BizResearchDonationSample.del_flag == 0,
+            )
+            .first()
+        )
+        if not dsample:
+            raise ValueError("捐赠样本不存在")
+        exists = (
+            db.query(BizSampleReception)
+            .filter(
+                BizSampleReception.donation_sample_id == donation_sample_id,
+                BizSampleReception.tenant_no == tenant_no,
+                BizSampleReception.del_flag == 0,
+            )
+            .first()
+        )
+        if exists:
+            raise ValueError("该捐赠样本已登记接收，请勿重复操作")
+        collection_protocol_id = ""
+        protocol_sample_id = ""
+        _donation_sample_id = donation_sample_id
+        _research_donation_id = dsample.research_donation_id
+        _sample_code = dsample.sample_code or ""
 
     conclusion = (accept_conclusion or "").strip()
     if conclusion not in ("js", "bj"):
@@ -392,8 +705,12 @@ def create_sample_reception(
 
     obj = BizSampleReception(
         id=generate_uuid(),
-        protocol_sample_id=protocol_sample_id,
-        collection_protocol_id=sample.collection_protocol_id,
+        protocol_sample_id=protocol_sample_id or "",
+        collection_protocol_id=collection_protocol_id or "",
+        reception_source=source,
+        donation_sample_id=_donation_sample_id,
+        research_donation_id=_research_donation_id,
+        sample_code=_sample_code or None,
         reception_time=reception_time,
         handler_user_id=handler_user_id,
         handler_name=handler_display or None,
@@ -484,3 +801,40 @@ def get_receptions_map_for_samples(
         )
         out[rec.protocol_sample_id] = item
     return out
+
+
+def get_receptions_map_for_donation_samples(
+    db: Session,
+    sample_ids: List[str],
+    tenant_no: str,
+) -> Dict[str, Dict[str, Any]]:
+    """按捐赠样本 ID 批量查询接收记录，供捐赠协议单查看弹窗使用。"""
+    if not sample_ids or not tenant_no:
+        return {}
+    rows = (
+        db.query(BizSampleReception)
+        .filter(
+            BizSampleReception.donation_sample_id.in_(sample_ids),
+            BizSampleReception.tenant_no == tenant_no,
+            BizSampleReception.del_flag == 0,
+        )
+        .all()
+    )
+    if not rows:
+        return {}
+    reception_ids = [r.id for r in rows]
+    results_map = check_result_svc.list_results_by_reception_ids(db, reception_ids, tenant_no)
+    out: Dict[str, Dict[str, Any]] = {}
+    for rec in rows:
+        sid = rec.donation_sample_id
+        if not sid:
+            continue
+        cr = results_map.get(rec.id)
+        if not cr:
+            cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
+        item = _reception_to_dict(rec, check_results=cr)
+        item["receiveAttachmentFiles"] = urls_to_attachment_list(
+            rec.receive_attachments, db, tenant_no
+        )
+        out[sid] = item
+    return out

backend/app/models/__init__.py

@@ -31,6 +31,12 @@ from app.models.models import (
     BizProcessConsumable,
     BizProcessOutput,
     BizProcessDataCollection,
+    BizProcessRoute,
+    BizProcessRouteStep,
+    BizProcessRouteStepReagent,
+    BizProcessRouteStepConsumable,
+    BizProcessRouteStepOutput,
+    BizProcessRouteStepDataCollection,
 )
 from app.models.research_donation import (  # noqa: F401
     BizResearchDonation,
@@ -80,4 +86,10 @@ __all__ = [
     "BizProcessConsumable",
     "BizProcessOutput",
     "BizProcessDataCollection",
+    "BizProcessRoute",
+    "BizProcessRouteStep",
+    "BizProcessRouteStepReagent",
+    "BizProcessRouteStepConsumable",
+    "BizProcessRouteStepOutput",
+    "BizProcessRouteStepDataCollection",
 ]

backend/app/models/models.py

@@ -918,6 +918,7 @@ class BizProtocolSampleTransport(Base):
     id = Column(String(64), primary_key=True, comment="主键ID")
     collection_protocol_id = Column(String(64), nullable=False, comment="采集协议单ID")
     sample_id = Column(String(64), nullable=False, comment="样本ID")
+    sample_code = Column(String(32), nullable=True, comment="样本编码快照")
     transport_type = Column(String(16), nullable=False, comment="kdys快递运输 ryxd人员携带")
     express_company_id = Column(String(64), nullable=True, comment="快递公司ID")
     waybill_no = Column(String(128), nullable=True, comment="快递单号")
@@ -1047,6 +1048,10 @@ class BizSampleReception(Base):
     check_no_stratification = Column(SmallInteger, nullable=False, default=0, comment="无分层")
     accept_conclusion = Column(String(16), nullable=False, comment="结论 js接收 bj拒收")
     receive_attachments = Column(String(4000), nullable=True, comment="接收附件")
+    reception_source = Column(String(20), nullable=False, default="cell_storage", comment="来源类型 cell_storage细胞存储 research_donation科研捐赠")
+    donation_sample_id = Column(String(64), nullable=True, comment="捐赠样本ID(仅科研捐赠)")
+    research_donation_id = Column(String(64), nullable=True, comment="科研捐赠单ID(仅科研捐赠)")
+    sample_code = Column(String(32), nullable=True, comment="样本编码快照")
     tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
     order_num = Column(Integer, nullable=True, comment="排序号")
     remark = Column(String(512), nullable=True, comment="备注信息")
@@ -1145,6 +1150,13 @@ class BizQcWorkOrder(Base):
     collection_protocol_id = Column(String(64), nullable=False, comment="协议单ID")
     protocol_no = Column(String(64), nullable=True, comment="协议号快照")
     sample_code = Column(String(32), nullable=True, comment="样本编码快照")
+    reception_source = Column(String(20), nullable=False, default="cell_storage", comment="来源类型 cell_storage/research_donation")
+    qc_biz_type = Column(
+        String(20),
+        nullable=False,
+        default="inbound",
+        comment="质检业务类型 inbound入库质检 outbound出库质检",
+    )
     work_order_status = Column(String(16), nullable=False, default="dpz", comment="工单状态")
     has_lab = Column(String(1), nullable=False, default="0", comment="含实验室自检")
     has_third_party = Column(String(1), nullable=False, default="0", comment="含第三方检测")
@@ -1217,12 +1229,14 @@ class BizQcTask(Base):
     task_no = Column(String(32), nullable=False, comment="任务号")
     task_type = Column(String(16), nullable=False, comment="sysz实验室 dsfjc第三方")
     source_line_id = Column(String(64), nullable=False, comment="工单明细行ID")
+    sample_code = Column(String(32), nullable=True, comment="样本编码快照")
     item_code = Column(String(64), nullable=False, comment="项目编码快照")
     item_name = Column(String(200), nullable=False, comment="项目名称快照")
     company_id = Column(String(64), nullable=True, comment="第三方公司ID")
     company_name = Column(String(200), nullable=True, comment="第三方公司名")
     fee_amount = Column(Numeric(18, 2), nullable=True, comment="费用快照")
     task_status = Column(String(16), nullable=False, default="dzx", comment="任务状态")
+    conclusion = Column(String(16), nullable=True, comment="检测结论 tg通过 btg不通过")
     third_party_attachments = Column(String(4000), nullable=True, comment="第三方材料")
     completed_time = Column(DateTime, nullable=True, comment="完成时间")
     completed_by = Column(String(64), nullable=True, comment="完成人")
@@ -1962,4 +1976,142 @@ class BizProcessDataCollection(Base):
     del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")
 
 
+class BizProcessRoute(Base):
+    """工艺路线 biz_process_route"""
+
+    __tablename__ = "biz_process_route"
+
+    id = Column(String(64), primary_key=True, comment="主键ID")
+    route_code = Column(String(64), nullable=False, comment="路线编码")
+    route_name = Column(String(100), nullable=False, comment="路线名称")
+    product_type_id = Column(String(64), nullable=True, comment="产品类型id")
+    version = Column(String(50), nullable=True, comment="版本号")
+    is_current = Column(CHAR(1), nullable=True, default="0", comment="是否当前版本(0-否 1-是)")
+    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 BizProcessRouteStep(Base):
+    """工艺路线步骤 biz_process_route_step"""
+
+    __tablename__ = "biz_process_route_step"
+
+    id = Column(String(64), primary_key=True, comment="主键ID")
+    route_id = Column(String(64), nullable=False, comment="路线id")
+    process_id = Column(String(64), nullable=True, comment="关联工艺id")
+    step_order = Column(Integer, nullable=True, comment="步骤顺序")
+    process_code = Column(String(64), nullable=True, comment="工序代码")
+    category_id = Column(String(64), nullable=True, comment="工序分类id")
+    process_name = Column(String(100), nullable=True, comment="工序名称")
+    description = Column(String(500), nullable=True, comment="描述")
+    equipment_name = Column(String(200), nullable=True, comment="预计使用设备名称")
+    estimated_duration = Column(String(100), nullable=True, comment="预估耗时")
+    sop_ids = Column(String(1000), nullable=True, comment="关联SOP文档id(逗号分隔)")
+    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 BizProcessRouteStepReagent(Base):
+    """工艺路线步骤试剂 biz_process_route_step_reagent"""
+
+    __tablename__ = "biz_process_route_step_reagent"
+
+    id = Column(String(64), primary_key=True, comment="主键ID")
+    step_id = Column(String(64), nullable=False, comment="步骤id")
+    reagent_code = Column(String(64), nullable=True, comment="试剂编码")
+    reagent_name = Column(String(100), nullable=True, comment="试剂名称")
+    spec = Column(String(200), nullable=True, comment="规格")
+    standard_usage = Column(String(100), nullable=True, comment="标准用量")
+    unit = Column(String(64), nullable=True, comment="单位")
+    is_key_material = Column(CHAR(1), nullable=True, default="0", comment="是否关键物料(0-否 1-是)")
+    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 BizProcessRouteStepConsumable(Base):
+    """工艺路线步骤耗材 biz_process_route_step_consumable"""
+
+    __tablename__ = "biz_process_route_step_consumable"
+
+    id = Column(String(64), primary_key=True, comment="主键ID")
+    step_id = Column(String(64), nullable=False, comment="步骤id")
+    consumable_code = Column(String(64), nullable=True, comment="耗材编码")
+    consumable_name = Column(String(100), nullable=True, comment="耗材名称")
+    spec_model = Column(String(200), nullable=True, comment="耗材规格型号")
+    standard_usage = Column(String(100), nullable=True, comment="标准用量")
+    unit = Column(String(64), nullable=True, comment="单位")
+    is_required = Column(CHAR(1), nullable=True, default="1", comment="是否必需(1-必须 0-可选)")
+    remark = Column(String(512), nullable=True, comment="备注")
+    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
+    order_num = Column(Integer, 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 BizProcessRouteStepOutput(Base):
+    """工艺路线步骤产出物 biz_process_route_step_output"""
+
+    __tablename__ = "biz_process_route_step_output"
+
+    id = Column(String(64), primary_key=True, comment="主键ID")
+    step_id = Column(String(64), nullable=False, comment="步骤id")
+    output_type_id = Column(String(64), nullable=True, comment="产出类型id")
+    output_name = Column(String(200), nullable=True, comment="产出物名称")
+    output_barcode = Column(String(200), nullable=True, comment="产出物条码")
+    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 BizProcessRouteStepDataCollection(Base):
+    """工艺路线步骤数据采集项 biz_process_route_step_data_collection"""
+
+    __tablename__ = "biz_process_route_step_data_collection"
+
+    id = Column(String(64), primary_key=True, comment="主键ID")
+    step_id = Column(String(64), nullable=False, comment="步骤id")
+    collection_name = Column(String(100), nullable=True, comment="采集项名称")
+    unit = Column(String(64), nullable=True, comment="单位")
+    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="删除标记")
+
+
 from app.models.qc_param import BizQcItemParam, BizQcTaskResult  # noqa: E402,F401

backend/app/models/research_donation.py

@@ -159,6 +159,7 @@ class BizResearchDonationSampleTransport(Base):
     id = Column(String(64), primary_key=True, comment="主键ID")
     research_donation_id = Column(String(64), nullable=False, comment="科研捐赠ID")
     sample_id = Column(String(64), nullable=False, comment="样本ID")
+    sample_code = Column(String(32), nullable=True, comment="样本编码快照")
     transport_type = Column(String(16), nullable=False, comment="kdys快递运输 ryxd人员携带")
     express_company_id = Column(String(64), nullable=True, comment="快递公司ID")
     waybill_no = Column(String(128), nullable=True, comment="快递单号")

backend/app/schemas/process_route.py

@@ -0,0 +1,122 @@
+"""工艺路线管理 Pydantic Schemas"""
+from pydantic import BaseModel, ConfigDict, Field
+from typing import Optional, List
+
+
+# Route main table
+class ProcessRouteCreateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    route_code: str = Field(..., alias="routeCode")
+    route_name: str = Field(..., alias="routeName")
+    product_type_id: Optional[str] = Field(None, alias="productTypeId")
+    version: Optional[str] = Field(None, alias="version")
+    remark: Optional[str] = None
+
+
+class ProcessRouteUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    route_code: Optional[str] = Field(None, alias="routeCode")
+    route_name: Optional[str] = Field(None, alias="routeName")
+    product_type_id: Optional[str] = Field(None, alias="productTypeId")
+    version: Optional[str] = Field(None, alias="version")
+    remark: Optional[str] = None
+
+
+class SetCurrentBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    route_id: str = Field(..., alias="routeId")
+
+
+# Step operations
+class StepAddBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    route_id: str = Field(..., alias="routeId")
+    process_id: str = Field(..., alias="processId")
+    step_order: Optional[int] = Field(None, alias="stepOrder")
+
+
+class StepCopyBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    step_id: str = Field(..., alias="stepId")
+
+
+class StepUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    step_id: str = Field(..., alias="stepId")
+    process_code: Optional[str] = Field(None, alias="processCode")
+    category_id: Optional[str] = Field(None, alias="categoryId")
+    process_name: Optional[str] = Field(None, alias="processName")
+    description: Optional[str] = None
+    equipment_name: Optional[str] = Field(None, alias="equipmentName")
+    estimated_duration: Optional[str] = Field(None, alias="estimatedDuration")
+    sop_ids: Optional[str] = Field(None, alias="sopIds")
+
+
+class StepReorderBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    route_id: str = Field(..., alias="routeId")
+    step_ids: List[str] = Field(..., alias="stepIds")
+
+
+# Step sub-table items
+class StepReagentItem(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    reagent_code: Optional[str] = Field(None, alias="reagentCode")
+    reagent_name: Optional[str] = Field(None, alias="reagentName")
+    spec: Optional[str] = None
+    standard_usage: Optional[str] = Field(None, alias="standardUsage")
+    unit: Optional[str] = None
+    is_key_material: Optional[str] = Field("0", alias="isKeyMaterial")
+    remark: Optional[str] = None
+
+
+class StepConsumableItem(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    consumable_code: Optional[str] = Field(None, alias="consumableCode")
+    consumable_name: Optional[str] = Field(None, alias="consumableName")
+    spec_model: Optional[str] = Field(None, alias="specModel")
+    standard_usage: Optional[str] = Field(None, alias="standardUsage")
+    unit: Optional[str] = None
+    is_required: Optional[str] = Field("1", alias="isRequired")
+    remark: Optional[str] = None
+
+
+class StepOutputItem(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    output_type_id: Optional[str] = Field(None, alias="outputTypeId")
+    output_name: Optional[str] = Field(None, alias="outputName")
+    output_barcode: Optional[str] = Field(None, alias="outputBarcode")
+    remark: Optional[str] = None
+
+
+class StepDataCollectionItem(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    collection_name: Optional[str] = Field(None, alias="collectionName")
+    unit: Optional[str] = None
+    order_num: Optional[int] = Field(0, alias="orderNum")
+    remark: Optional[str] = None
+
+
+# Sub-table update bodies (step_id + items[])
+class StepReagentUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    step_id: str = Field(..., alias="stepId")
+    items: List[StepReagentItem] = Field(...)
+
+
+class StepConsumableUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    step_id: str = Field(..., alias="stepId")
+    items: List[StepConsumableItem] = Field(...)
+
+
+class StepOutputUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    step_id: str = Field(..., alias="stepId")
+    items: List[StepOutputItem] = Field(...)
+
+
+class StepDataCollectionUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    step_id: str = Field(..., alias="stepId")
+    items: List[StepDataCollectionItem] = Field(...)