zkxw/backend/app/crud/sample_storage_cert_qc.py

"""样本储存证质检信息 CRUD。"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, Optional

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizSampleStorageCertQc,
)
from app.services.sample_storage_cert_qc import (
    dumps_payload,
    empty_detail,
    loads_payload,
    normalize_payload,
    resolve_sample_type_key,
)


def _row_to_dict(
    row: BizSampleStorageCertQc,
    *,
    protocol_code: str = "",
    mother_name: str = "",
    sample_type_name: str = "",
) -> Dict[str, Any]:
    return {
        "id": row.id,
        "sampleCode": row.sample_code or "",
        "labFlowCode": row.sample_code or "",
        "protocolCode": protocol_code or "",
        "motherName": mother_name or "",
        "sampleTypeName": sample_type_name or "",
        "sampleTypeKey": row.sample_type_key,
        "collectionProtocolId": row.collection_protocol_id or "",
        "qcWorkOrderId": row.qc_work_order_id or "",
        "filledFlag": str(row.filled_flag or "0") == "1",
        "payload": loads_payload(row.payload),
        "createTime": row.create_time.isoformat(sep=" ", timespec="seconds") if row.create_time else "",
        "updateTime": row.update_time.isoformat(sep=" ", timespec="seconds") if row.update_time else "",
    }


def _load_sample_context(db: Session, tenant_no: str, sample_code: str):
    code = (sample_code or "").strip()
    if not code:
        return None
    sample = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.sample_code == code,
            BizProtocolSample.del_flag == 0,
        )
        .first()
    )
    if not sample:
        return None
    type_name = ""
    if sample.sample_type_id:
        st = (
            db.query(BizCellSampleType)
            .filter(
                BizCellSampleType.id == sample.sample_type_id,
                BizCellSampleType.del_flag == 0,
            )
            .first()
        )
        if st:
            type_name = st.type_name or ""
    protocol_code = ""
    mother_name = ""
    if sample.collection_protocol_id:
        proto = (
            db.query(BizCollectionProtocol)
            .filter(
                BizCollectionProtocol.id == sample.collection_protocol_id,
                BizCollectionProtocol.tenant_no == tenant_no,
                BizCollectionProtocol.del_flag == 0,
            )
            .first()
        )
        if proto:
            protocol_code = proto.protocol_no or ""
        pc = (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id == sample.collection_protocol_id,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .first()
        )
        if pc:
            mother_name = pc.mother_name or ""
    return sample, type_name, protocol_code, mother_name


def get_detail(db: Session, tenant_no: str, sample_code: str) -> Optional[Dict[str, Any]]:
    """样本不存在返回 None；存在但未填返回 empty_detail。"""
    ctx = _load_sample_context(db, tenant_no, sample_code)
    if not ctx:
        return None
    sample, type_name, protocol_code, mother_name = ctx
    key = resolve_sample_type_key(type_name)
    row = (
        db.query(BizSampleStorageCertQc)
        .filter(
            BizSampleStorageCertQc.tenant_no == tenant_no,
            BizSampleStorageCertQc.sample_code == sample.sample_code,
            BizSampleStorageCertQc.del_flag == 0,
        )
        .first()
    )
    if not row:
        return empty_detail(
            sample_code=sample.sample_code or "",
            protocol_code=protocol_code,
            mother_name=mother_name,
            sample_type_name=type_name,
            sample_type_key=key,
            collection_protocol_id=sample.collection_protocol_id or "",
        )
    return _row_to_dict(
        row,
        protocol_code=protocol_code,
        mother_name=mother_name,
        sample_type_name=type_name,
    )


def upsert(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    sample_code: str,
    payload: Any,
    qc_work_order_id: Optional[str] = None,
) -> Dict[str, Any]:
    ctx = _load_sample_context(db, tenant_no, sample_code)
    if not ctx:
        raise ValueError("样本不存在")
    sample, type_name, protocol_code, mother_name = ctx
    key = resolve_sample_type_key(type_name)
    if not key:
        raise ValueError("该样本类型不支持填写储存证质检信息")
    clean = normalize_payload(payload)
    now = datetime.now()
    row = (
        db.query(BizSampleStorageCertQc)
        .filter(
            BizSampleStorageCertQc.tenant_no == tenant_no,
            BizSampleStorageCertQc.sample_code == sample.sample_code,
            BizSampleStorageCertQc.del_flag == 0,
        )
        .first()
    )
    if not row:
        row = BizSampleStorageCertQc(
            id=generate_uuid(),
            sample_code=sample.sample_code,
            collection_protocol_id=sample.collection_protocol_id,
            qc_work_order_id=(qc_work_order_id or None),
            sample_type_key=key,
            payload=dumps_payload(clean),
            filled_flag="1",
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(row)
    else:
        row.collection_protocol_id = sample.collection_protocol_id
        row.sample_type_key = key
        row.payload = dumps_payload(clean)
        row.filled_flag = "1"
        if qc_work_order_id:
            row.qc_work_order_id = qc_work_order_id
        row.update_by = user_id
        row.update_time = now
    db.commit()
    db.refresh(row)
    return _row_to_dict(
        row,
        protocol_code=protocol_code,
        mother_name=mother_name,
        sample_type_name=type_name,
    )