zkxw/backend/app/crud/qc_task.py

"""质检任务 CRUD"""
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
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.crud.collection_protocol import lookup_attachments_by_urls, urls_to_attachment_list
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,
    _resolve_reception_id_for_work_order_detail,
    _sample_type_name_for_reception,
    _work_order_qc_biz_type,
    _OUTBOUND_BIZ_TYPE_LABELS,
    build_work_order_list_summaries,
)
from app.core.db_collation import eq_unicode, unicode_ci
from app.utils.sample_special import apply_special_to_item_from_samples
from app.crud import qc_task_assignee as assignee_svc
from app.crud.qc_inbound_lab_storage_sort import (
    inbound_lab_storage_order_by_clauses,
    uses_inbound_lab_storage_list_sort,
)
from app.models.models import (
    BizAttachment,
    BizCollectionProtocol,
    BizProtocolSample,
    BizQcTask,
    BizQcTaskAssignee,
    BizQcWorkOrder,
    BizQcWorkOrderLabItem,
    BizQcWorkOrderThirdItem,
    BizSampleReception,
)
from app.models.research_donation import BizResearchDonation, BizResearchDonationSample


def _dec(v: Any) -> Optional[float]:
    if v is None:
        return None
    return float(v)


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _gen_task_no(db: Session, work_order, seq: int) -> str:
    return f"{work_order.work_order_no}-{seq:02d}"


_CONCLUSION_LABELS = {"tg": "通过", "btg": "不通过"}
_PROTOCOL_STATUS_LABELS = {
    "dsx": "待生效", "sxz": "生效中", "ygq": "已过期", "yzf": "已作废", "sdfq": "属地废弃", "zky": "转科研"
}
_DONATION_STATUS_LABELS = {
    "dsx": "待生效", "sxz": "生效中", "ygq": "已过期", "yzf": "已作废", "sdfq": "属地废弃", "zckc": "转细胞存储"
}


def _conclusion_label(v: Optional[str]) -> str:
    key = (v or "").strip()
    return _CONCLUSION_LABELS.get(key, key)


def _split_attachment_urls(urls_str: Optional[str]) -> List[str]:
    if not urls_str:
        return []
    return [u.strip() for u in urls_str.split(",") if u and u.strip()]


def _task_to_dict(
    t: BizQcTask,
    work_order_no: str = "",
    db: Optional[Session] = None,
    tenant_no: Optional[str] = None,
    *,
    attachment_by_url: Optional[Dict[str, BizAttachment]] = None,
) -> Dict[str, Any]:
    tn = tenant_no or t.tenant_no
    if db:
        attachment_files = urls_to_attachment_list(
            t.third_party_attachments,
            db,
            tn,
            attachment_by_url=attachment_by_url,
        )
    else:
        attachment_files = []
    return {
        "id": t.id,
        "workOrderId": t.work_order_id,
        "batchId": getattr(t, "batch_id", None) or "",
        "orderSource": getattr(t, "order_source", "inbound") or "inbound",
        "workOrderNo": work_order_no,
        "taskNo": t.task_no,
        "taskType": t.task_type,
        "sourceLineId": t.source_line_id,
        "itemCode": t.item_code,
        "itemName": t.item_name,
        "companyId": t.company_id or "",
        "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": attachment_files,
        "completedTime": _fmt_dt(t.completed_time),
        "completedBy": t.completed_by or "",
        "remark": t.remark or "",
        "createTime": _fmt_dt(t.create_time),
        "updateTime": _fmt_dt(t.update_time),
    }


def _task_work_order_context(wo: BizQcWorkOrder, t: BizQcTask) -> Dict[str, Any]:
    biz = _work_order_qc_biz_type(wo)
    wo_type = getattr(wo, "work_order_type", None) or "inbound"
    if wo_type == "temporary":
        order_source = "temporary"
        recv_source = ""
        recv_label = ""
    elif wo_type == "outbound":
        order_source = "outbound"
        recv_source = getattr(wo, "biz_type", "") or ""
        recv_label = _OUTBOUND_BIZ_TYPE_LABELS.get(recv_source, recv_source) if recv_source else ""
    else:
        order_source = getattr(t, "order_source", "inbound") or "inbound"
        recv_source = getattr(wo, "reception_source", "cell_storage") or "cell_storage"
        recv_label = _reception_source_label(recv_source)
    return {
        "orderSource": order_source,
        "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": recv_source,
        "receptionSourceLabel": recv_label,
        "cellProductId": getattr(wo, "cell_product_id", None) or "",
        "cellProductCode": "",
        "cellProductName": "",
        "cellProductTypeName": "",
        "cellProductPassage": "",
        "cellProductSpecName": "",
        "receptionTime": "",
        "handlerName": "",
        "sampleTypeName": "",
        "workOrderStatus": wo.work_order_status or "",
    }


_WORK_ORDER_SUMMARY_COPY_KEYS = (
    "triggerSource",
    "triggerSourceLabel",
    "hasLab",
    "hasThirdParty",
    "labStorageCode",
    "protocolSampleId",
    "collectionProtocolId",
    "applicationNo",
    "applicationName",
    "projectTitle",
    "customerName",
    "channelName",
    "patientName",
    "infusionTime",
    "sortNo",
    "totalCourseCount",
    "products",
    "cellType",
    "specText",
    "productionWorkOrderId",
    "productionWorkOrderNo",
    "cellProductCode",
    "cellProductName",
    "cellProductTypeName",
    "cellProductPassage",
    "cellProductSpecName",
    "paymentSummary",
    "feeSummary",
    "bizType",
    "bizTypeLabel",
    "motherName",
    "newbornName",
    "isSpecial",
    "specialRemark",
    "unfinishedBadgeCount",
    "protocolSiblingSamples",
)


def _merge_work_order_summary_into_task(item: Dict[str, Any], summary: Dict[str, Any]) -> None:
    """将列表级工单摘要合并进任务行，供「我的质检任务」前端聚合展示。"""
    for key in _WORK_ORDER_SUMMARY_COPY_KEYS:
        if key in summary:
            item[key] = summary[key]
    if summary.get("createTime"):
        item["workOrderCreateTime"] = summary["createTime"]
    if "remark" in summary:
        item["workOrderRemark"] = summary.get("remark") or ""
    for key in (
        "receptionTime",
        "handlerName",
        "sampleTypeName",
        "protocolStatus",
        "protocolStatusLabel",
        "motherName",
        "newbornName",
        "isSpecial",
        "specialRemark",
    ):
        if not item.get(key) and summary.get(key):
            item[key] = summary[key]


def _enrich_mine_tasks_with_work_order_summaries(
    db: Session,
    rows: List[Tuple[BizQcTask, BizQcWorkOrder]],
    items: List[Dict[str, Any]],
    tenant_no: Optional[str],
) -> List[Dict[str, Any]]:
    if not rows or not items:
        return items
    summaries = build_work_order_list_summaries(
        db, [wo for _, wo in rows], tenant_no
    )
    if not summaries:
        return items
    for item in items:
        summary = summaries.get(item.get("workOrderId") or "")
        if summary:
            _merge_work_order_summary_into_task(item, summary)
    return items


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)
    cell_proto_ids = [
        rec.collection_protocol_id
        for rec in rec_map.values()
        if (rec.reception_source or "cell_storage") != "research_donation" and rec.collection_protocol_id
    ]
    donation_ids = [
        rec.research_donation_id
        for rec in rec_map.values()
        if (rec.reception_source or "") == "research_donation" and rec.research_donation_id
    ]
    proto_status_map: Dict[str, str] = {}
    proto_void_type_map: Dict[str, str] = {}
    tenant_no_for_status = ""
    if cell_proto_ids:
        for p in (
            db.query(
                BizCollectionProtocol.id,
                BizCollectionProtocol.protocol_status,
                BizCollectionProtocol.void_type,
                BizCollectionProtocol.tenant_no,
            )
            .filter(BizCollectionProtocol.id.in_(cell_proto_ids), BizCollectionProtocol.del_flag == 0)
            .all()
        ):
            proto_status_map[p.id] = p.protocol_status or ""
            proto_void_type_map[p.id] = p.void_type or ""
            if not tenant_no_for_status and (p.tenant_no or "").strip():
                tenant_no_for_status = (p.tenant_no or "").strip()
    donation_status_map: Dict[str, str] = {}
    donation_void_type_map: Dict[str, str] = {}
    if donation_ids:
        for d in (
            db.query(
                BizResearchDonation.id,
                BizResearchDonation.donation_status,
                BizResearchDonation.void_type,
                BizResearchDonation.tenant_no,
            )
            .filter(BizResearchDonation.id.in_(donation_ids), BizResearchDonation.del_flag == 0)
            .all()
        ):
            donation_status_map[d.id] = d.donation_status or ""
            donation_void_type_map[d.id] = d.void_type or ""
            if not tenant_no_for_status and (d.tenant_no or "").strip():
                tenant_no_for_status = (d.tenant_no or "").strip()
    for item in items:
        if item.get("orderSource") in ("temporary", "outbound"):
            item.setdefault("receptionSource", "")
            item.setdefault("receptionSourceLabel", "")
            continue
        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,
        )
        source = item.get("receptionSource") or (rec.reception_source if rec else "cell_storage")
        if source == "research_donation":
            _did = rec.research_donation_id if rec else ""
            _st = donation_status_map.get(_did or "", "")
            _vt = donation_void_type_map.get(_did or "", "")
            item["protocolStatus"] = _st
            item["voidType"] = _vt
            from app.core.protocol_status_display import format_protocol_status_label

            item["protocolStatusLabel"] = format_protocol_status_label(
                _st, _vt, kind="donation", db=db, tenant_no=tenant_no_for_status or None
            )
        else:
            _pid = rec.collection_protocol_id if rec else ""
            _st = proto_status_map.get(_pid or "", "")
            _vt = proto_void_type_map.get(_pid or "", "")
            item["protocolStatus"] = _st
            item["voidType"] = _vt
            from app.core.protocol_status_display import format_protocol_status_label

            item["protocolStatusLabel"] = format_protocol_status_label(
                _st, _vt, kind="protocol", db=db, tenant_no=tenant_no_for_status or None
            )
        apply_special_to_item_from_samples(
            item,
            source=source,
            protocol_sample_id=rec.protocol_sample_id or "",
            donation_sample_id=rec.donation_sample_id or "",
            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],
    *,
    attachment_by_url: Optional[Dict[str, BizAttachment]] = None,
) -> Dict[str, Any]:
    out = _task_to_dict(
        t,
        wo.work_order_no or "",
        db,
        tenant_no,
        attachment_by_url=attachment_by_url,
    )
    out.update(_task_work_order_context(wo, t))
    return out


def _attachment_map_for_task_rows(
    db: Session,
    rows: List[Tuple[BizQcTask, BizQcWorkOrder]],
    tenant_no: Optional[str],
) -> Dict[str, BizAttachment]:
    urls: List[str] = []
    for t, _wo in rows:
        urls.extend(_split_attachment_urls(t.third_party_attachments))
    if not urls:
        return {}
    return lookup_attachments_by_urls(db, urls, tenant_no)


def _task_rows_to_dicts(
    db: Session,
    rows: List[Tuple[BizQcTask, BizQcWorkOrder]],
    tenant_no: Optional[str],
) -> List[Dict[str, Any]]:
    """将任务行批量转为字典；附件元数据一次预取，避免 N+1。"""
    att_map = _attachment_map_for_task_rows(db, rows, tenant_no)
    return [
        _task_row_to_dict(t, wo, db, tenant_no, attachment_by_url=att_map)
        for t, wo in rows
    ]


def create_task_for_line(
    db: Session,
    *,
    work_order: BizQcWorkOrder,
    task_type: str,
    source_line_id: str,
    item_code: str,
    item_name: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    seq: int = 1,
    company_id: Optional[str] = None,
    company_name: Optional[str] = None,
    fee_amount: Optional[Decimal] = None,
    order_source: str = "inbound",
    batch_id: Optional[str] = None,
) -> BizQcTask:
    obj = BizQcTask(
        id=generate_uuid(),
        work_order_id=work_order.id,
        batch_id=batch_id,
        order_source=order_source,
        task_no=_gen_task_no(db, work_order, seq),
        task_type=task_type,
        source_line_id=source_line_id,
        sample_code=getattr(work_order, "sample_code", None) or None,
        item_code=item_code,
        item_name=item_name,
        company_id=company_id,
        company_name=company_name,
        fee_amount=fee_amount,
        task_status="dzx",
        tenant_no=tenant_no,
        order_num=seq,
        create_by=user_id,
        create_dept_id=create_dept_id or "",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.flush()
    return obj


def list_tasks(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    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,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = (
            db.query(BizQcTask, BizQcWorkOrder)
            .join(
                BizQcWorkOrder,
                and_(
                    BizQcWorkOrder.id == BizQcTask.work_order_id,
                    BizQcWorkOrder.del_flag == 0,
                ),
            )
            .filter(BizQcTask.del_flag == 0)
        )
        if tenant_no:
            q0 = q0.filter(BizQcTask.tenant_no == tenant_no)
        if work_order_id:
            q0 = q0.filter(BizQcTask.work_order_id == work_order_id)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizQcTask.task_no.like(like),
                    BizQcTask.item_name.like(like),
                    BizQcWorkOrder.work_order_no.like(like),
                    BizQcWorkOrder.sample_code.like(like),
                )
            )
        tt = (task_type or "").strip()
        if tt:
            q0 = q0.filter(BizQcTask.task_type == tt)
        st = (task_status or "").strip()
        if st:
            q0 = q0.filter(BizQcTask.task_status == st)
        bt = (qc_biz_type or "").strip()
        if bt:
            if bt == "temporary":
                q0 = q0.filter(
                    or_(
                        BizQcWorkOrder.qc_biz_type == "temporary",
                        BizQcWorkOrder.work_order_type == "temporary",
                    )
                )
            else:
                q0 = q0.filter(BizQcWorkOrder.qc_biz_type == bt)
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizQcTask, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizQcTask.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(
            BizQcTask.create_time.desc(),
            BizQcWorkOrder.work_order_no.desc(),
            BizQcTask.task_no.asc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = _task_rows_to_dicts(db, rows, tenant_no)
    items = _enrich_tasks_reception_fields(db, items)
    return assignee_svc.enrich_tasks_with_assignees(db, items, tenant_no), int(total)


def list_mine_tasks(
    db: Session,
    *,
    assignee_user_id: str,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    task_type: Optional[str] = None,
    task_status: Optional[str] = None,
    qc_biz_type: Optional[str] = None,
    reception_source: Optional[str] = None,
    biz_type: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = (
            db.query(BizQcTask, BizQcWorkOrder)
            .join(
                BizQcWorkOrder,
                and_(
                    BizQcWorkOrder.id == BizQcTask.work_order_id,
                    BizQcWorkOrder.del_flag == 0,
                ),
            )
            .join(
                BizQcTaskAssignee,
                and_(
                    eq_unicode(BizQcTaskAssignee.task_id, BizQcTask.id),
                    unicode_ci(BizQcTaskAssignee.user_id) == assignee_user_id,
                    BizQcTaskAssignee.del_flag == 0,
                ),
            )
            .filter(BizQcTask.del_flag == 0)
        )
        if tenant_no:
            q0 = q0.filter(BizQcTask.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizQcTask.task_no.like(like),
                    BizQcTask.item_name.like(like),
                    BizQcWorkOrder.work_order_no.like(like),
                    BizQcWorkOrder.sample_code.like(like),
                )
            )
        tt = (task_type or "").strip()
        if tt:
            q0 = q0.filter(BizQcTask.task_type == tt)
        st = (task_status or "").strip()
        if st:
            q0 = q0.filter(BizQcTask.task_status == st)
        bt = (qc_biz_type or "").strip()
        if bt:
            if bt == "temporary":
                q0 = q0.filter(
                    or_(
                        BizQcWorkOrder.qc_biz_type == "temporary",
                        BizQcWorkOrder.work_order_type == "temporary",
                    )
                )
            else:
                q0 = q0.filter(BizQcWorkOrder.qc_biz_type == bt)
        rs = (reception_source or "").strip()
        if rs:
            q0 = q0.filter(BizQcWorkOrder.reception_source == rs)
        bt_out = (biz_type or "").strip()
        if bt_out:
            q0 = q0.filter(BizQcWorkOrder.biz_type == bt_out)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizQcTask.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    q_page = _filtered()
    if uses_inbound_lab_storage_list_sort(reception_source):
        q_page = q_page.order_by(
            *inbound_lab_storage_order_by_clauses(),
            BizQcWorkOrder.work_order_no.desc(),
            BizQcTask.task_no.asc(),
        )
    else:
        q_page = q_page.order_by(
            BizQcTask.create_time.desc(),
            BizQcWorkOrder.work_order_no.desc(),
            BizQcTask.task_no.asc(),
        )
    rows = (
        q_page
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = _task_rows_to_dicts(db, rows, tenant_no)
    items = _enrich_tasks_reception_fields(db, items)
    items = _enrich_mine_tasks_with_work_order_summaries(db, rows, items, tenant_no)
    return assignee_svc.enrich_tasks_with_assignees(db, items, tenant_no), int(total)


def assert_task_assignee(
    db: Session, task_id: str, user_id: str, tenant_no: str
) -> None:
    if not assignee_svc.is_user_assignee(db, task_id, user_id, tenant_no):
        raise PermissionError("您不是该任务的负责人，无权操作")


def list_tasks_by_work_order(db: Session, work_order_id: str, tenant_no: str) -> List[Dict[str, Any]]:
    wo_id = (work_order_id or "").strip()
    if not wo_id:
        return []
    return list_tasks_by_work_order_ids(db, [wo_id], tenant_no).get(wo_id, [])


def list_tasks_by_work_order_ids(
    db: Session, work_order_ids: List[str], tenant_no: str
) -> Dict[str, List[Dict[str, Any]]]:
    ids = list({(x or "").strip() for x in work_order_ids if (x or "").strip()})
    if not ids or not tenant_no:
        return {}
    ids = ids[:100]
    rows = (
        db.query(BizQcTask, BizQcWorkOrder)
        .join(BizQcWorkOrder, BizQcWorkOrder.id == BizQcTask.work_order_id)
        .filter(
            BizQcTask.work_order_id.in_(ids),
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
            BizQcWorkOrder.del_flag == 0,
        )
        .order_by(BizQcTask.work_order_id.asc(), BizQcTask.order_num.asc())
        .all()
    )
    items = _task_rows_to_dicts(db, rows, tenant_no)
    items = _enrich_tasks_reception_fields(db, items)
    items = assignee_svc.enrich_tasks_with_assignees(db, items, tenant_no)
    grouped: Dict[str, List[Dict[str, Any]]] = {wo_id: [] for wo_id in ids}
    for item in items:
        wo_id = (item.get("workOrderId") or "").strip()
        if wo_id in grouped:
            grouped[wo_id].append(item)
    return grouped


def get_task(
    db: Session, row_id: str, tenant_no: Optional[str] = None
) -> Optional[BizQcTask]:
    q = db.query(BizQcTask).filter(BizQcTask.id == row_id, BizQcTask.del_flag == 0)
    if tenant_no:
        q = q.filter(BizQcTask.tenant_no == tenant_no)
    return q.first()


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,
            BizQcWorkOrder.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("任务不存在")
    t, wo = row
    out = _task_row_to_dict(t, wo, db, tenant_no)
    out = _enrich_tasks_reception_fields(db, [out])[0]
    out = assignee_svc.enrich_tasks_with_assignees(db, [out], tenant_no)[0]
    from app.crud.qc_work_order import _enrich_work_order_qc_targets

    out = _enrich_work_order_qc_targets(db, tenant_no, [out])[0]
    if (out.get("qcBizType") or "") != "temporary":
        resolved_reception_id = _resolve_reception_id_for_work_order_detail(db, wo, tenant_no)
        reception_detail = (
            sample_reception_svc.get_reception_by_id(db, resolved_reception_id, tenant_no)
            if resolved_reception_id
            else None
        )
        if reception_detail:
            out["reception"] = reception_detail
    return out


def start_task(db: Session, row_id: str, *, tenant_no: str, user_id: Optional[str]) -> BizQcTask:
    t = get_task(db, row_id, tenant_no=tenant_no)
    if not t:
        raise ValueError("任务不存在")
    if t.task_type != "sysz":
        raise ValueError("仅实验室自检任务可开始")
    if t.task_status != "dzx":
        raise ValueError("当前状态不可开始")
    t.task_status = "jxz"
    t.update_by = user_id
    t.update_time = datetime.now()
    db.add(t)
    db.commit()
    db.refresh(t)
    return t


def _set_task_remark(t: BizQcTask, remark: Optional[str]) -> None:
    t.remark = (remark or "").strip() or None


def _require_dsfjc_attachments(t: BizQcTask, *, for_status: str = "ywc") -> None:
    """第三方检测且任务状态为已完成时，须至少有一个检测材料附件。"""
    if (t.task_type or "").strip() != "dsfjc":
        return
    if (for_status or "").strip() != "ywc":
        return
    if not (t.third_party_attachments or "").strip():
        raise ValueError("第三方检测须上传检测材料附件")


def complete_task(
    db: Session,
    row_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str],
    conclusion: Optional[str] = None,
    remark: Optional[str] = None,
    commit: bool = True,
) -> 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":
        raise ValueError("任务类型无效")
    _require_dsfjc_attachments(t)
    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
    t.update_time = datetime.now()
    db.add(t)
    if commit:
        db.commit()
        db.refresh(t)
        from app.crud import qc_work_order as wo_svc

        wo_svc.try_complete_work_order(db, t.work_order_id, tenant_no)
    else:
        db.flush()
    return t


_QC_TASK_RESULT_STATUSES = frozenset({"dzx", "jxz", "ywc"})


def resolve_task_result_conclusion(
    conclusion: Optional[str],
    target_status: str,
) -> str:
    """Normalize conclusion for apply_task_result.

    Returns '' | 'tg' | 'btg'. Empty allowed only when target is dzx/jxz.
    """
    concl = (conclusion or "").strip()
    if concl and concl not in ("tg", "btg"):
        raise ValueError("检测结论无效")
    if target_status == "ywc" and concl not in ("tg", "btg"):
        raise ValueError("请选择检测结论（通过/不通过）")
    return concl


def apply_task_result(
    db: Session,
    row_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str],
    conclusion: Optional[str] = None,
    remark: Optional[str] = None,
    material_usages: Optional[List[Dict[str, Any]]] = None,
    task_status: Optional[str] = None,
    commit: bool = True,
) -> BizQcTask:
    """保存检测结果与结论，并按指定任务状态落库（默认已完成）。"""
    from app.crud import qc_material_usage as mu_svc
    from app.crud import qc_work_order as wo_svc

    t = get_task(db, row_id, tenant_no=tenant_no)
    if not t:
        raise ValueError("任务不存在")
    target = (task_status or "").strip() or "ywc"
    if target not in _QC_TASK_RESULT_STATUSES:
        raise ValueError("任务状态无效")
    concl = resolve_task_result_conclusion(conclusion, target)
    _require_dsfjc_attachments(t, for_status=target)
    if material_usages is not None:
        mu_svc.deduct_for_task(
            db,
            row_id,
            task_no=t.task_no or "",
            material_usages=material_usages,
            tenant_no=tenant_no,
            user_id=user_id,
        )
    was_done = (t.task_status or "").strip() == "ywc"
    t.conclusion = concl or None
    _set_task_remark(t, remark if remark is not None else "")
    t.task_status = target
    t.update_by = user_id
    t.update_time = datetime.now()
    if target == "ywc":
        t.completed_time = datetime.now()
        t.completed_by = user_id
    else:
        t.completed_time = None
        t.completed_by = None
    db.add(t)
    if commit:
        db.commit()
        db.refresh(t)
        if target == "ywc":
            wo_svc.try_complete_work_order(db, t.work_order_id, tenant_no)
        elif was_done:
            wo_svc.try_reopen_work_order(db, t.work_order_id, tenant_no)
    else:
        db.flush()
    return t


def update_task_attachments(
    db: Session,
    row_id: str,
    *,
    attachments: str,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
) -> BizQcTask:
    t = get_task(db, row_id, tenant_no=tenant_no)
    if not t:
        raise ValueError("任务不存在")
    t.third_party_attachments = (attachments or "").strip() or None
    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()
    db.add(t)
    if commit:
        db.commit()
        db.refresh(t)
    else:
        db.flush()
    return t


def create_qc_task_manual(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    task_type: str = "sysz",
    item_code: str = "",
    item_name: str = "",
    company_id: Optional[str] = None,
    company_name: Optional[str] = None,
    fee_amount: Optional[float] = None,
    sample_code: str = "",
    work_order_id: str = "",
    remark: str = "",
) -> BizQcTask:
    """手动创建质检任务（可独立于工单）"""
    # 生成任务号：ZJRW + 日期 + 序号
    day = datetime.now().strftime("%Y%m%d")
    like = f"ZJRW{day}%"
    cnt = db.query(BizQcTask).filter(
        BizQcTask.tenant_no == tenant_no,
        BizQcTask.task_no.like(like),
    ).count()
    task_no = f"ZJRW{day}{cnt + 1:04d}"

    task = BizQcTask(
        id=generate_uuid(),
        work_order_id=work_order_id or "",
        task_no=task_no,
        task_type=task_type,
        source_line_id="",
        sample_code=sample_code,
        item_code=item_code,
        item_name=item_name,
        company_id=company_id,
        company_name=company_name,
        fee_amount=fee_amount,
        task_status="dzx",
        tenant_no=tenant_no,
        remark=remark,
        create_by=user_id,
        create_dept_id=create_dept_id,
    )
    db.add(task)
    db.flush()
    return task


def _soft_delete_source_line_for_task(
    db: Session,
    task: BizQcTask,
    *,
    tenant_no: str,
    user_id: Optional[str],
    now: datetime,
) -> None:
    """删除任务时同步软删批次下对应 lab/third 明细行，避免配置弹窗仍勾选该项目。"""
    sid = (task.source_line_id or "").strip()
    if not sid:
        return
    task_type = (task.task_type or "").strip()
    line = None
    if task_type == "sysz":
        line = (
            db.query(BizQcWorkOrderLabItem)
            .filter(
                BizQcWorkOrderLabItem.id == sid,
                BizQcWorkOrderLabItem.tenant_no == tenant_no,
                BizQcWorkOrderLabItem.del_flag == 0,
            )
            .first()
        )
    elif task_type == "dsfjc":
        line = (
            db.query(BizQcWorkOrderThirdItem)
            .filter(
                BizQcWorkOrderThirdItem.id == sid,
                BizQcWorkOrderThirdItem.tenant_no == tenant_no,
                BizQcWorkOrderThirdItem.del_flag == 0,
            )
            .first()
        )
    else:
        line = (
            db.query(BizQcWorkOrderLabItem)
            .filter(
                BizQcWorkOrderLabItem.id == sid,
                BizQcWorkOrderLabItem.tenant_no == tenant_no,
                BizQcWorkOrderLabItem.del_flag == 0,
            )
            .first()
        )
        if not line:
            line = (
                db.query(BizQcWorkOrderThirdItem)
                .filter(
                    BizQcWorkOrderThirdItem.id == sid,
                    BizQcWorkOrderThirdItem.tenant_no == tenant_no,
                    BizQcWorkOrderThirdItem.del_flag == 0,
                )
                .first()
            )
    if not line:
        return
    line.del_flag = 1
    line.update_by = user_id
    line.update_time = now
    db.add(line)


def _refresh_work_order_item_flags(
    db: Session,
    wo: BizQcWorkOrder,
    *,
    tenant_no: str,
    user_id: Optional[str],
    now: datetime,
) -> None:
    """按未删除明细行回写工单 has_lab / has_third_party / 第三方费用合计。"""
    lab_count = (
        db.query(BizQcWorkOrderLabItem)
        .filter(
            BizQcWorkOrderLabItem.work_order_id == wo.id,
            BizQcWorkOrderLabItem.tenant_no == tenant_no,
            BizQcWorkOrderLabItem.del_flag == 0,
        )
        .count()
    )
    third_rows = (
        db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.work_order_id == wo.id,
            BizQcWorkOrderThirdItem.tenant_no == tenant_no,
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .all()
    )
    wo.has_lab = "1" if lab_count else "0"
    wo.has_third_party = "1" if third_rows else "0"
    if third_rows:
        fee_total = sum((r.fee_amount or Decimal("0")) for r in third_rows)
        wo.third_party_fee_total = fee_total
    else:
        wo.third_party_fee_total = None
    wo.update_by = user_id
    wo.update_time = now
    db.add(wo)


def delete_qc_task(db: Session, task_id: str, tenant_no: str, user_id: Optional[str]) -> None:
    """软删除质检任务、检测结果，并同步解除批次中的项目关联。"""
    task = db.query(BizQcTask).filter(
        and_(
            BizQcTask.id == task_id,
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
        )
    ).first()
    if not task:
        raise ValueError("任务不存在")

    work_order_id = task.work_order_id
    now = datetime.now()

    # 软删除任务结果
    from app.models.qc_param import BizQcTaskResult

    results = db.query(BizQcTaskResult).filter(
        and_(BizQcTaskResult.task_id == task_id, BizQcTaskResult.del_flag == 0)
    ).all()
    for r in results:
        r.del_flag = 1
        r.update_by = user_id
        r.update_time = now

    # 同步软删批次明细行（配置弹窗勾选来源）
    _soft_delete_source_line_for_task(
        db, task, tenant_no=tenant_no, user_id=user_id, now=now
    )

    # 软删除任务负责人
    assignee_svc.soft_delete_assignees_by_task_ids(db, [task_id], tenant_no)

    # 软删除任务本体
    task.del_flag = 1
    task.update_by = user_id
    task.update_time = now

    db.flush()

    from app.crud.qc_work_order import get_work_order, try_complete_work_order

    wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
    if wo:
        _refresh_work_order_item_flags(
            db, wo, tenant_no=tenant_no, user_id=user_id, now=now
        )
        if (wo.work_order_status or "").strip() == "ywc":
            pending = (
                db.query(BizQcTask)
                .filter(
                    BizQcTask.work_order_id == work_order_id,
                    BizQcTask.tenant_no == tenant_no,
                    BizQcTask.del_flag == 0,
                    BizQcTask.task_status != "ywc",
                )
                .count()
            )
            if pending:
                wo.work_order_status = "ypz"
                wo.update_by = user_id
                wo.update_time = now
                db.add(wo)

    db.commit()
    try_complete_work_order(db, work_order_id, tenant_no)