zkxw/backend/app/services/sample_storage_cert_qc.py

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

import json
from typing import Any, Dict, Iterable, List, Optional, Set

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

ACL_SCOPE_QC = "qc"
ACL_SCOPE_PRODUCTION = "production"
ACL_SCOPES = frozenset({ACL_SCOPE_QC, ACL_SCOPE_PRODUCTION})
ACL_CONFIG_MENU = "cell-storage:sample-info:storage-cert-qc-acl"

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

# 与 web-admin STORAGE_CERT_QC_GROUPS 对齐的合法 payload 键（两类型并集）。
ALL_STORAGE_CERT_QC_FIELD_KEYS = frozenset(
    {
        "collection_time",
        "transport_temp_c",
        "receive_time",
        "cryo_volume_ml",
        "tnc_1e8",
        "cd34_count_1e6",
        "cd34_cd45dim_1e5",
        "viability_pct",
        "bacteria_test",
        "hbsag",
        "anti_hcv",
        "anti_tp",
        "anti_hiv",
        "cmv_igm",
        "mother_hbsag",
        "mother_hbv_dna",
        "mother_anti_hcv",
        "mother_hcv_rna",
        "mother_anti_tp",
        "mother_cmv_igm",
        "mother_anti_hiv",
        "mother_hiv1_rna",
        "hla_typing",
        "storage_date",
        "storage_years",
        "result_comment",
        "report_issue_date",
        "storage_tube_count",
        "available_cell_type",
        "available_cell_count_1e7",
        "available_viability_pct",
        "endotoxin_test",
        "fungus_test",
        "mycoplasma_test",
        "available_cell_morphology",
        "available_immunophenotype",
        "result_judgment",
        "storage_date_p2",
        "freeze_method",
        "storage_temp",
    }
)

# 脐带间充质储存证不收录母亲血液三项核酸检测。
UC_MSC_EXCLUDED_PAYLOAD_KEYS = frozenset(
    {
        "mother_hbv_dna",
        "mother_hcv_rna",
        "mother_hiv1_rna",
    }
)


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 legal_field_keys_for_type(sample_type_key: Optional[str]) -> Set[str]:
    keys = set(ALL_STORAGE_CERT_QC_FIELD_KEYS)
    if sample_type_key == SAMPLE_TYPE_KEY_UC_MSC:
        keys -= UC_MSC_EXCLUDED_PAYLOAD_KEYS
    return keys


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 sanitize_payload_for_type(
    payload: Dict[str, Any],
    sample_type_key: Optional[str],
) -> Dict[str, Any]:
    """按样本类型去掉不应展示/落库的键。"""
    if sample_type_key != SAMPLE_TYPE_KEY_UC_MSC:
        return payload
    return {key: value for key, value in payload.items() if key not in UC_MSC_EXCLUDED_PAYLOAD_KEYS}


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 dumps_field_keys(keys: Iterable[str]) -> str:
    cleaned = sorted({str(k).strip() for k in keys if str(k).strip()})
    return json.dumps(cleaned, ensure_ascii=False, separators=(",", ":"))


def loads_field_keys(raw: Optional[str]) -> List[str]:
    if not raw:
        return []
    try:
        data = json.loads(raw)
    except (TypeError, ValueError):
        return []
    if not isinstance(data, list):
        return []
    out: List[str] = []
    seen = set()
    for item in data:
        key = str(item or "").strip()
        if not key or key in seen:
            continue
        seen.add(key)
        out.append(key)
    return out


def _payload_value_text(value: Any) -> str:
    if value is None:
        return ""
    return str(value).strip()


def merge_payload_with_acl(
    *,
    existing: Dict[str, Any],
    incoming: Dict[str, Any],
    editable_keys: Iterable[str],
    sample_type_key: Optional[str],
) -> Dict[str, Any]:
    """仅合并有权字段；无权字段若相对原值变化则拒绝。"""
    allowed = {str(k).strip() for k in editable_keys if str(k).strip()}
    if not allowed:
        raise PermissionError("无字段编辑权限")
    clean_incoming = sanitize_payload_for_type(normalize_payload(incoming), sample_type_key)
    base = sanitize_payload_for_type(dict(existing or {}), sample_type_key)
    for key, value in clean_incoming.items():
        if key in allowed:
            continue
        if _payload_value_text(base.get(key)) != _payload_value_text(value):
            raise PermissionError(f"无权修改字段: {key}")
    merged = dict(base)
    for key in allowed:
        if key in clean_incoming:
            merged[key] = clean_incoming[key]
    return sanitize_payload_for_type(merged, sample_type_key)


def empty_detail(
    *,
    sample_code: str,
    lab_flow_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": lab_flow_code or "",
        "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": {},
    }