zkxw/backend/app/services/sample_storage_cert_qc.py

"""样本储存证质检信息 — 类型键与 payload 工具。"""
from __future__ import annotations

import json
from typing import Any, Dict, Optional

SAMPLE_TYPE_NAME_PERINATAL_HSC = "围产期造血干细胞"
SAMPLE_TYPE_NAME_UC_MSC = "脐带间充质干细胞"
SAMPLE_TYPE_KEY_PERINATAL_HSC = "perinatal_hsc"
SAMPLE_TYPE_KEY_UC_MSC = "uc_msc"

_NAME_TO_KEY = {
    SAMPLE_TYPE_NAME_PERINATAL_HSC: SAMPLE_TYPE_KEY_PERINATAL_HSC,
    SAMPLE_TYPE_NAME_UC_MSC: SAMPLE_TYPE_KEY_UC_MSC,
}


def resolve_sample_type_key(sample_type_name: Optional[str]) -> Optional[str]:
    name = (sample_type_name or "").strip()
    return _NAME_TO_KEY.get(name)


def supports_storage_cert_qc(sample_type_name: Optional[str]) -> bool:
    return resolve_sample_type_key(sample_type_name) is not None


def normalize_payload(payload: Any) -> Dict[str, Any]:
    if not isinstance(payload, dict):
        raise ValueError("payload 必须为对象")
    out: Dict[str, Any] = {}
    for k, v in payload.items():
        if not isinstance(k, str):
            continue
        key = k.strip()
        if not key:
            continue
        out[key] = v
    return out


def dumps_payload(payload: Dict[str, Any]) -> str:
    return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))


def loads_payload(raw: Optional[str]) -> Dict[str, Any]:
    if not raw:
        return {}
    try:
        data = json.loads(raw)
    except (TypeError, ValueError):
        return {}
    if not isinstance(data, dict):
        return {}
    return data


def empty_detail(
    *,
    sample_code: str,
    protocol_code: str = "",
    mother_name: str = "",
    sample_type_name: str = "",
    sample_type_key: Optional[str] = None,
    collection_protocol_id: str = "",
    qc_work_order_id: str = "",
) -> Dict[str, Any]:
    return {
        "sampleCode": sample_code,
        "labFlowCode": sample_code,
        "protocolCode": protocol_code or "",
        "motherName": mother_name or "",
        "sampleTypeName": sample_type_name or "",
        "sampleTypeKey": sample_type_key or resolve_sample_type_key(sample_type_name),
        "collectionProtocolId": collection_protocol_id or "",
        "qcWorkOrderId": qc_work_order_id or "",
        "filledFlag": False,
        "payload": {},
    }