zkxw/backend/app/services/protocol_sample_list_summary.py

"""协议/捐赠列表卡片：按样本编码批量取接收结论与门禁质检结论（轻量字段）。"""

from __future__ import annotations

from collections import defaultdict
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session

from app.models.models import BizQcTask, BizQcWorkOrder, BizSampleReception
from app.services.qc_work_order_conclusion import (
    build_qc_conclusion_fields,
    work_order_qc_conclusion_label,
)

_ACCEPT_CONCLUSION_LABELS = {"js": "接收", "bj": "拒收"}
_VALID_QC_CONCLUSIONS = frozenset({"tg", "btg"})


def build_sample_item_reception_qc_fields(
    reception: Optional[Dict[str, Any]],
    qc: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
    """合并接收/质检结论摘要为列表 sampleItems 字段。"""
    conclusion = ""
    remark = ""
    if reception:
        conclusion = (reception.get("acceptConclusion") or "").strip()
        remark = (reception.get("remark") or "").strip()
    qc_code = ""
    qc_label = ""
    if qc:
        qc_code = (qc.get("qcConclusion") or "").strip()
        qc_label = (qc.get("qcConclusionLabel") or "").strip()
        if qc_code and not qc_label:
            qc_label = work_order_qc_conclusion_label(qc_code)
    return {
        "receptionAcceptConclusion": conclusion,
        "receptionAcceptConclusionLabel": _ACCEPT_CONCLUSION_LABELS.get(conclusion, ""),
        "receptionRemark": remark,
        "qcConclusion": qc_code,
        "qcConclusionLabel": qc_label,
    }


def get_list_reception_summary_map_by_sample_codes(
    db: Session,
    tenant_no: str,
    sample_codes: List[str],
    *,
    reception_source: str,
) -> Dict[str, Dict[str, str]]:
    """按租户 + 样本编码 + 来源返回接收摘要（acceptConclusion / remark），同码取首条。"""
    codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()})
    if not tenant_no or not codes or not (reception_source or "").strip():
        return {}
    rows = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
            BizSampleReception.reception_source == reception_source,
            BizSampleReception.sample_code.in_(codes),
        )
        .all()
    )
    out: Dict[str, Dict[str, str]] = {}
    for rec in rows:
        code = (rec.sample_code or "").strip()
        if not code or code in out:
            continue
        out[code] = {
            "acceptConclusion": (rec.accept_conclusion or "").strip(),
            "remark": (rec.remark or "").strip(),
        }
    return out


def get_list_qc_conclusion_map_by_sample_codes(
    db: Session,
    tenant_no: str,
    sample_codes: List[str],
    *,
    reception_source: str,
) -> Dict[str, Dict[str, str]]:
    """按租户 + 样本编码 + 来源返回门禁质检结论（优先工单落库，否则由任务派生）。"""
    codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()})
    if not tenant_no or not codes or not (reception_source or "").strip():
        return {}
    rows = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.reception_source == reception_source,
            BizQcWorkOrder.trigger_source == "reception",
            BizQcWorkOrder.sample_code.in_(codes),
        )
        .all()
    )
    code_to_wo: Dict[str, BizQcWorkOrder] = {}
    for wo in rows:
        code = (wo.sample_code or "").strip()
        if code and code not in code_to_wo:
            code_to_wo[code] = wo

    need_derive_ids: List[str] = []
    for wo in code_to_wo.values():
        stored = (getattr(wo, "conclusion", None) or "").strip()
        if stored not in _VALID_QC_CONCLUSIONS:
            wid = (wo.id or "").strip()
            if wid:
                need_derive_ids.append(wid)

    conclusions_by_wo: Dict[str, List[Optional[str]]] = defaultdict(list)
    if need_derive_ids:
        task_rows = (
            db.query(BizQcTask.work_order_id, BizQcTask.conclusion)
            .filter(
                BizQcTask.del_flag == 0,
                BizQcTask.work_order_id.in_(need_derive_ids),
            )
            .all()
        )
        for wo_id, conc in task_rows:
            wid = (wo_id or "").strip()
            if wid:
                conclusions_by_wo[wid].append(conc)

    out: Dict[str, Dict[str, str]] = {}
    for code, wo in code_to_wo.items():
        stored = (getattr(wo, "conclusion", None) or "").strip()
        if stored in _VALID_QC_CONCLUSIONS:
            out[code] = {
                "qcConclusion": stored,
                "qcConclusionLabel": work_order_qc_conclusion_label(stored),
            }
            continue
        wid = (wo.id or "").strip()
        fields = build_qc_conclusion_fields(conclusions_by_wo.get(wid, []))
        out[code] = {
            "qcConclusion": fields["qcConclusion"],
            "qcConclusionLabel": fields["qcConclusionLabel"],
        }
    return out


# 兼容旧调用名
def get_list_qc_status_map_by_sample_codes(
    db: Session,
    tenant_no: str,
    sample_codes: List[str],
    *,
    reception_source: str,
) -> Dict[str, Dict[str, str]]:
    return get_list_qc_conclusion_map_by_sample_codes(
        db, tenant_no, sample_codes, reception_source=reception_source
    )