提交

backend/app/api/qc.py

@@ -0,0 +1,769 @@
+"""质量检测 — 第三方/实验室主数据、质检工单、质检任务 API"""
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, Query
+from pydantic import BaseModel, ConfigDict, Field
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.core.menu_guard import require_admin_or_super_admin, require_menu_code
+from app.core.response import error_response, success_response
+from app.crud import qc_lab_test_item as lab_svc
+from app.crud import qc_item_param as param_svc
+from app.crud import qc_task_result as result_svc
+from app.crud import qc_task as task_svc
+from app.crud import qc_third_party_company as tpc_svc
+from app.crud import qc_third_party_item as tpi_svc
+from app.crud import qc_work_order as wo_svc
+from app.crud.collection_protocol import total_pages
+from app.api.sample_reception import _get_current_user_id, _get_tenant_no, _page_payload
+
+router = APIRouter(prefix="/web/admin", tags=["质量检测"])
+
+
+# ---------- 第三方检测公司 ----------
+
+
+class ThirdPartyCompanyBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    company_code: str = Field(..., alias="companyCode")
+    company_name: str = Field(..., alias="companyName")
+    contact_name: Optional[str] = Field(None, alias="contactName")
+    contact_phone: Optional[str] = Field(None, alias="contactPhone")
+    address: Optional[str] = None
+    invoice_title: Optional[str] = Field(None, alias="invoiceTitle")
+    tax_id: Optional[str] = Field(None, alias="taxId")
+    bank_name: Optional[str] = Field(None, alias="bankName")
+    bank_account: Optional[str] = Field(None, alias="bankAccount")
+    status: str = "1"
+    order_num: Optional[int] = Field(None, alias="orderNum")
+    remark: Optional[str] = None
+
+
+class ThirdPartyCompanyUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    company_code: Optional[str] = Field(None, alias="companyCode")
+    company_name: Optional[str] = Field(None, alias="companyName")
+    contact_name: Optional[str] = Field(None, alias="contactName")
+    contact_phone: Optional[str] = Field(None, alias="contactPhone")
+    address: Optional[str] = None
+    invoice_title: Optional[str] = Field(None, alias="invoiceTitle")
+    tax_id: Optional[str] = Field(None, alias="taxId")
+    bank_name: Optional[str] = Field(None, alias="bankName")
+    bank_account: Optional[str] = Field(None, alias="bankAccount")
+    status: Optional[str] = None
+    order_num: Optional[int] = Field(None, alias="orderNum")
+    remark: Optional[str] = None
+
+
+@router.get("/qc-third-party-companies")
+async def list_qc_third_party_companies(
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100),
+    keyword: Optional[str] = None,
+    status: Optional[str] = None,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items, total = tpc_svc.list_companies(
+            db, page=page, page_size=pageSize, keyword=keyword, status=status, tenant_no=tenant_no
+        )
+        return success_response(_page_payload(items, total, page, pageSize))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.get("/qc-third-party-companies/options")
+async def list_qc_third_party_company_options(
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcWorkOrder:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        return success_response(tpc_svc.list_company_options(db, tenant_no))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.post("/qc-third-party-companies")
+async def create_qc_third_party_company(
+    body: ThirdPartyCompanyBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = tpc_svc.create_company(
+            db,
+            company_code=body.company_code,
+            company_name=body.company_name,
+            contact_name=body.contact_name,
+            contact_phone=body.contact_phone,
+            address=body.address,
+            invoice_title=body.invoice_title,
+            tax_id=body.tax_id,
+            bank_name=body.bank_name,
+            bank_account=body.bank_account,
+            status=body.status,
+            tenant_no=tenant_no,
+            order_num=body.order_num,
+            remark=body.remark,
+            user_id=_get_current_user_id(user),
+        )
+        return success_response(tpc_svc._row_to_dict(obj), message="创建成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.put("/qc-third-party-companies/{company_id}")
+async def update_qc_third_party_company(
+    company_id: str,
+    body: ThirdPartyCompanyUpdateBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = tpc_svc.update_company(
+            db,
+            company_id,
+            company_code=body.company_code,
+            company_name=body.company_name,
+            contact_name=body.contact_name,
+            contact_phone=body.contact_phone,
+            address=body.address,
+            invoice_title=body.invoice_title,
+            tax_id=body.tax_id,
+            bank_name=body.bank_name,
+            bank_account=body.bank_account,
+            status=body.status,
+            order_num=body.order_num,
+            remark=body.remark,
+            user_id=_get_current_user_id(user),
+            tenant_no=tenant_no,
+        )
+        return success_response(tpc_svc._row_to_dict(obj), message="更新成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.delete("/qc-third-party-companies/{company_id}")
+async def delete_qc_third_party_company(
+    company_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        tpc_svc.soft_delete_company(db, company_id, tenant_no=tenant_no)
+        return success_response(None, message="删除成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+# ---------- 第三方检测项目 ----------
+
+
+class ThirdPartyItemBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    company_id: str = Field(..., alias="companyId")
+    item_code: str = Field(..., alias="itemCode")
+    item_name: str = Field(..., alias="itemName")
+    fee_amount: float = Field(..., alias="feeAmount")
+    status: str = "1"
+    order_num: Optional[int] = Field(None, alias="orderNum")
+    remark: Optional[str] = None
+
+
+class ThirdPartyItemUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    company_id: Optional[str] = Field(None, alias="companyId")
+    item_code: Optional[str] = Field(None, alias="itemCode")
+    item_name: Optional[str] = Field(None, alias="itemName")
+    fee_amount: Optional[float] = Field(None, alias="feeAmount")
+    status: Optional[str] = None
+    order_num: Optional[int] = Field(None, alias="orderNum")
+    remark: Optional[str] = None
+
+
+@router.get("/qc-third-party-items")
+async def list_qc_third_party_items(
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100),
+    keyword: Optional[str] = None,
+    status: Optional[str] = None,
+    companyId: Optional[str] = None,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items, total = tpi_svc.list_items(
+            db,
+            page=page,
+            page_size=pageSize,
+            keyword=keyword,
+            status=status,
+            company_id=companyId,
+            tenant_no=tenant_no,
+        )
+        return success_response(_page_payload(items, total, page, pageSize))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.get("/qc-third-party-items/active")
+async def list_qc_third_party_items_active(
+    companyId: Optional[str] = None,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcWorkOrder:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        return success_response(tpi_svc.list_active_items(db, tenant_no, company_id=companyId))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.post("/qc-third-party-items")
+async def create_qc_third_party_item(
+    body: ThirdPartyItemBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = tpi_svc.create_item(
+            db,
+            company_id=body.company_id,
+            item_code=body.item_code,
+            item_name=body.item_name,
+            fee_amount=body.fee_amount,
+            status=body.status,
+            tenant_no=tenant_no,
+            order_num=body.order_num,
+            remark=body.remark,
+            user_id=_get_current_user_id(user),
+        )
+        return success_response(tpi_svc._row_to_dict(obj, ""), message="创建成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.put("/qc-third-party-items/{item_id}")
+async def update_qc_third_party_item(
+    item_id: str,
+    body: ThirdPartyItemUpdateBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = tpi_svc.update_item(
+            db,
+            item_id,
+            company_id=body.company_id,
+            item_code=body.item_code,
+            item_name=body.item_name,
+            fee_amount=body.fee_amount,
+            status=body.status,
+            order_num=body.order_num,
+            remark=body.remark,
+            user_id=_get_current_user_id(user),
+            tenant_no=tenant_no,
+        )
+        return success_response(tpi_svc._row_to_dict(obj, ""), message="更新成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.delete("/qc-third-party-items/{item_id}")
+async def delete_qc_third_party_item(
+    item_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcThirdPartyCompany:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        tpi_svc.soft_delete_item(db, item_id, tenant_no=tenant_no)
+        return success_response(None, message="删除成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+# ---------- 实验室自检项目 ----------
+
+
+class LabTestItemBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    item_code: str = Field(..., alias="itemCode")
+    item_name: str = Field(..., alias="itemName")
+    description: Optional[str] = None
+    status: str = "1"
+    order_num: Optional[int] = Field(None, alias="orderNum")
+    remark: Optional[str] = None
+
+
+class LabTestItemUpdateBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    item_code: Optional[str] = Field(None, alias="itemCode")
+    item_name: Optional[str] = Field(None, alias="itemName")
+    description: Optional[str] = None
+    status: Optional[str] = None
+    order_num: Optional[int] = Field(None, alias="orderNum")
+    remark: Optional[str] = None
+
+
+# ---------- 检测项目参数模板 ----------
+
+
+class ItemParamBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    params: List[Dict[str, Any]] = Field(default_factory=list, alias="params")
+
+
+class TaskResultBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    results: List[Dict[str, Any]] = Field(default_factory=list, alias="results")
+
+
+class TaskCompleteWithResultBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+    results: List[Dict[str, Any]] = Field(default_factory=list, alias="results")
+    third_party_attachments: Optional[str] = Field(None, alias="thirdPartyAttachments")
+
+
+@router.get("/qc-lab-test-items")
+async def list_qc_lab_test_items(
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100),
+    keyword: Optional[str] = None,
+    status: Optional[str] = None,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items, total = lab_svc.list_lab_items(
+            db, page=page, page_size=pageSize, keyword=keyword, status=status, tenant_no=tenant_no
+        )
+        return success_response(_page_payload(items, total, page, pageSize))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.get("/qc-lab-test-items/active")
+async def list_qc_lab_test_items_active(
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcWorkOrder:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        return success_response(lab_svc.list_active_lab_items(db, tenant_no))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.post("/qc-lab-test-items")
+async def create_qc_lab_test_item(
+    body: LabTestItemBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = lab_svc.create_lab_item(
+            db,
+            item_code=body.item_code,
+            item_name=body.item_name,
+            description=body.description,
+            status=body.status,
+            tenant_no=tenant_no,
+            order_num=body.order_num,
+            remark=body.remark,
+            user_id=_get_current_user_id(user),
+        )
+        return success_response(lab_svc._row_to_dict(obj), message="创建成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.put("/qc-lab-test-items/{item_id}")
+async def update_qc_lab_test_item(
+    item_id: str,
+    body: LabTestItemUpdateBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = lab_svc.update_lab_item(
+            db,
+            item_id,
+            item_code=body.item_code,
+            item_name=body.item_name,
+            description=body.description,
+            status=body.status,
+            order_num=body.order_num,
+            remark=body.remark,
+            user_id=_get_current_user_id(user),
+            tenant_no=tenant_no,
+        )
+        return success_response(lab_svc._row_to_dict(obj), message="更新成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.delete("/qc-lab-test-items/{item_id}")
+async def delete_qc_lab_test_item(
+    item_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        lab_svc.soft_delete_lab_item(db, item_id, tenant_no=tenant_no)
+        return success_response(None, message="删除成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+# ---------- 检测项目参数模板 ----------
+
+
+@router.get("/qc-item-params")
+async def list_item_params(
+    itemSource: str = Query(...),
+    itemId: str = Query(...),
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        return success_response(param_svc.list_params_by_item(db, itemSource, itemId, tenant_no))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.put("/qc-item-params")
+async def save_item_params(
+    body: ItemParamBody,
+    itemSource: str = Query(...),
+    itemId: str = Query(...),
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcLabTestItem:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        return success_response(
+            param_svc.save_params_for_item(db, itemSource, itemId, body.params, tenant_no, _get_current_user_id(user)),
+            message="保存成功",
+        )
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+# ---------- 质检工单 ----------
+
+
+class WorkOrderConfigureBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    has_lab: bool = Field(False, alias="hasLab")
+    has_third_party: bool = Field(False, alias="hasThirdParty")
+    lab_item_ids: List[str] = Field(default_factory=list, alias="labItemIds")
+    third_party_item_ids: List[str] = Field(default_factory=list, alias="thirdPartyItemIds")
+
+
+@router.get("/qc-work-orders")
+async def list_qc_work_orders(
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100),
+    keyword: Optional[str] = None,
+    workOrderStatus: Optional[str] = None,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcWorkOrder:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items, total = wo_svc.list_work_orders(
+            db,
+            page=page,
+            page_size=pageSize,
+            keyword=keyword,
+            work_order_status=workOrderStatus,
+            tenant_no=tenant_no,
+        )
+        return success_response(_page_payload(items, total, page, pageSize))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.get("/qc-work-orders/{work_order_id}")
+async def get_qc_work_order_detail(
+    work_order_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcWorkOrder:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        data = wo_svc.get_work_order_detail(db, work_order_id, tenant_no)
+        return success_response(data)
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.post("/qc-work-orders/{work_order_id}/configure")
+async def configure_qc_work_order(
+    work_order_id: str,
+    body: WorkOrderConfigureBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcWorkOrder:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = wo_svc.configure_work_order(
+            db,
+            work_order_id,
+            has_lab=body.has_lab,
+            has_third_party=body.has_third_party,
+            lab_item_ids=body.lab_item_ids,
+            third_party_item_ids=body.third_party_item_ids,
+            tenant_no=tenant_no,
+            user_id=_get_current_user_id(user),
+        )
+        return success_response(wo_svc._wo_to_dict(obj), message="配置成功")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+# ---------- 质检任务 ----------
+
+
+class TaskAttachmentsBody(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    third_party_attachments: str = Field("", alias="thirdPartyAttachments")
+
+
+@router.get("/qc-tasks")
+async def list_qc_tasks(
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100),
+    keyword: Optional[str] = None,
+    taskType: Optional[str] = None,
+    taskStatus: Optional[str] = None,
+    workOrderId: Optional[str] = None,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcTask:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        items, total = task_svc.list_tasks(
+            db,
+            page=page,
+            page_size=pageSize,
+            keyword=keyword,
+            task_type=taskType,
+            task_status=taskStatus,
+            work_order_id=workOrderId,
+            tenant_no=tenant_no,
+        )
+        return success_response(_page_payload(items, total, page, pageSize))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.get("/qc-tasks/{task_id}")
+async def get_qc_task_detail(
+    task_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcTask:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        data = task_svc.get_task_detail(db, task_id, tenant_no)
+        return success_response(data)
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.post("/qc-tasks/{task_id}/start")
+async def start_qc_task(
+    task_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcTask:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = task_svc.start_task(db, task_id, tenant_no=tenant_no, user_id=_get_current_user_id(user))
+        return success_response(task_svc._task_to_dict(obj), message="已开始")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.post("/qc-tasks/{task_id}/complete")
+async def complete_qc_task(
+    task_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcTask:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = task_svc.complete_task(
+            db, task_id, tenant_no=tenant_no, user_id=_get_current_user_id(user)
+        )
+        return success_response(task_svc._task_to_dict(obj), message="已完成")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.get("/qc-tasks/{task_id}/results")
+async def get_task_results(
+    task_id: str,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcTask:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        return success_response(result_svc.list_results_by_task(db, task_id, tenant_no))
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.post("/qc-tasks/{task_id}/complete-with-result")
+async def complete_task_with_result(
+    task_id: str,
+    body: TaskCompleteWithResultBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcTask:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        user_id = _get_current_user_id(user)
+        if body.results:
+            result_svc.save_results_for_task(db, task_id, body.results, tenant_no, user_id)
+        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="已完成")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))
+
+
+@router.put("/qc-tasks/{task_id}/attachments")
+async def update_qc_task_attachments(
+    task_id: str,
+    body: TaskAttachmentsBody,
+    db: Session = Depends(get_db),
+    user: dict = Depends(require_menu_code("biz:qcTask:list")),
+):
+    try:
+        tenant_no = _get_tenant_no(db, user)
+        obj = task_svc.update_task_attachments(
+            db,
+            task_id,
+            attachments=body.third_party_attachments,
+            tenant_no=tenant_no,
+            user_id=_get_current_user_id(user),
+        )
+        return success_response(task_svc._task_to_dict(obj), message="已保存")
+    except PermissionError as e:
+        return error_response(str(e), code=403)
+    except ValueError as e:
+        return error_response(str(e))
+    except Exception as e:
+        return error_response(str(e))