shanxia2.0/backend/app/api/qc.py

"""质量检测 — 第三方/实验室主数据、质检工单、质检任务 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_material as mat_svc
from app.crud import qc_material_usage as mu_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)
    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")
    material_usages: Optional[List[Dict[str, Any]]] = Field(None, alias="materialUsages")


@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,
    qcBizType: 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,
            qc_biz_type=qcBizType,
            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)
        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,
            )
        # 物料扣减（可选，仅当传了 materialUsages 时）
        if body.material_usages:
            task_obj_pre = task_svc.get_task(db, task_id, tenant_no=tenant_no)
            task_no = task_obj_pre.task_no if task_obj_pre else ""
            mu_svc.deduct_for_task(
                db,
                task_id,
                task_no=task_no,
                material_usages=body.material_usages,
                tenant_no=tenant_no,
                user_id=user_id,
            )
        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:
        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))


# ---------- 质检物料库 ----------


class QcMaterialBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    material_code: str = Field(..., alias="materialCode")
    material_name: str = Field(..., alias="materialName")
    spec_model: Optional[str] = Field(None, alias="specModel")
    unit: Optional[str] = None
    warning_qty: Optional[float] = Field(0, alias="warningQty")
    status: str = "1"
    remark: Optional[str] = None


class QcMaterialUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    material_code: Optional[str] = Field(None, alias="materialCode")
    material_name: Optional[str] = Field(None, alias="materialName")
    spec_model: Optional[str] = Field(None, alias="specModel")
    unit: Optional[str] = None
    warning_qty: Optional[float] = Field(None, alias="warningQty")
    status: Optional[str] = None
    remark: Optional[str] = None


class StockInBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    qty: float = Field(..., gt=0)
    remark: Optional[str] = None


@router.get("/qc-materials")
async def list_qc_materials(
    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:qcMaterial:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = mat_svc.list_materials(
            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-materials/options")
async def qc_material_options(
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcMaterial:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        return success_response(mat_svc.options_list(db, tenant_no=tenant_no))
    except Exception as e:
        return error_response(str(e))


@router.post("/qc-materials")
async def create_qc_material(
    body: QcMaterialBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcMaterial:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        obj = mat_svc.create_material(db, body.model_dump(by_alias=True, exclude_none=True), tenant_no=tenant_no, user_id=user_id)
        return success_response(obj, message="创建成功")
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


@router.put("/qc-materials/{material_id}")
async def update_qc_material(
    material_id: str,
    body: QcMaterialUpdateBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcMaterial:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        obj = mat_svc.update_material(db, material_id, body.model_dump(by_alias=True, exclude_none=True), tenant_no=tenant_no, user_id=user_id)
        return success_response(obj, message="更新成功")
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


@router.delete("/qc-materials/{material_id}")
async def delete_qc_material(
    material_id: str,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcMaterial:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        mat_svc.soft_delete_material(db, material_id, tenant_no=tenant_no, user_id=user_id)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


@router.post("/qc-materials/{material_id}/stock-in")
async def stock_in_qc_material(
    material_id: str,
    body: StockInBody,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcMaterial:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_current_user_id(user)
        obj = mat_svc.stock_in(db, material_id, qty=body.qty, remark=body.remark, tenant_no=tenant_no, user_id=user_id)
        return success_response(obj, message="入库成功")
    except ValueError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(str(e))


# ---------- 物料变动记录 ----------


@router.get("/qc-material-usages")
async def list_qc_material_usages(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    materialId: Optional[str] = None,
    changeType: Optional[str] = None,
    startTime: Optional[str] = None,
    endTime: Optional[str] = None,
    db: Session = Depends(get_db),
    user: dict = Depends(require_menu_code("biz:qcMaterialUsage:list")),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items, total = mu_svc.list_usages(
            db, tenant_no, page=page, page_size=pageSize,
            material_id=materialId, change_type=changeType,
            start_time=startTime, end_time=endTime,
        )
        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))