zkxw/backend/app/services/lab_sample_processing_fields.py

"""实验室样本处理数据 — 字段目录与部门归属。"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple

DATASET_HSC = "hsc"
DATASET_UBMC = "ubmc"
DATASET_MSC = "msc"
DATASETS = (DATASET_HSC, DATASET_UBMC, DATASET_MSC)

DEPT_MARKET = "market"
DEPT_PRODUCTION = "production"
DEPT_QC = "qc"
DEPTS = (DEPT_MARKET, DEPT_PRODUCTION, DEPT_QC)

PERM_LIST = "lab:sample-processing:list"
PERM_MARKET_EDIT = "lab:sample-processing:market:edit"
PERM_PRODUCTION_EDIT = "lab:sample-processing:production:edit"
PERM_QC_EDIT = "lab:sample-processing:qc:edit"
PERM_MANAGER = "lab:sample-processing:manager"

DEPT_EDIT_PERMS = {
    DEPT_MARKET: PERM_MARKET_EDIT,
    DEPT_PRODUCTION: PERM_PRODUCTION_EDIT,
    DEPT_QC: PERM_QC_EDIT,
}

SOURCE_PROTOCOL = "protocol"
SOURCE_RECEPTION = "reception"
SOURCE_PRODUCTION = "production_reception"
SOURCE_EXTRA = "extra"


@dataclass(frozen=True)
class FieldDef:
    key: str
    label: str
    dept: str
    source: str
    attr: Optional[str] = None  # ORM attr on source object when applicable


def _f(key: str, label: str, dept: str, source: str, attr: Optional[str] = None) -> FieldDef:
    return FieldDef(key=key, label=label, dept=dept, source=source, attr=attr or key)


HSC_FIELDS: Tuple[FieldDef, ...] = (
    _f("sampleCode", "细胞编码", DEPT_MARKET, SOURCE_PROTOCOL, "sample_code"),
    _f("motherName", "准妈妈姓名", DEPT_MARKET, SOURCE_PROTOCOL, "mother_name"),
    _f("collectTime", "采集时间", DEPT_MARKET, SOURCE_PROTOCOL, "sample_collect_time"),
    _f("collectHospital", "采集医院", DEPT_MARKET, SOURCE_PROTOCOL, "hospital_name"),
    _f("receptionTime", "接收时间", DEPT_PRODUCTION, SOURCE_RECEPTION, "reception_time"),
    _f("receiveTemperature", "样品温度到达/℃", DEPT_PRODUCTION, SOURCE_RECEPTION, "receive_temperature"),
    _f("receiveCheck", "收样检查", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("hasMaternalBlood", "是否有母血", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("remeasureMaternalBlood", "是否重采母血", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("hasConsent", "知情同意书或协议书", DEPT_MARKET, SOURCE_EXTRA),
    _f("hasPackingList", "脐血运输装箱清单", DEPT_MARKET, SOURCE_EXTRA),
    _f("hasReceiveConfirm", "脐带血接收确认单", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("hasHealthSurvey", "孕妇健康调查表", DEPT_MARKET, SOURCE_EXTRA),
    _f("labelClear", "标识清晰", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("receptionHandler", "接收登记人", DEPT_PRODUCTION, SOURCE_RECEPTION, "handler_name"),
    _f("receptionRemark", "备注", DEPT_PRODUCTION, SOURCE_RECEPTION, "remark"),
    _f("originalBloodWeightG", "原血重量/g", DEPT_PRODUCTION, SOURCE_PRODUCTION, "original_blood_weight_g"),
    _f("hesVolumeMl", "HES用量/ml", DEPT_PRODUCTION, SOURCE_PRODUCTION, "hes_volume_ml"),
    _f("bloodType", "血型", DEPT_PRODUCTION, SOURCE_PRODUCTION, "blood_type"),
    _f("infectiousFourCard", "传染四项检测（检测卡）", DEPT_QC, SOURCE_PRODUCTION, "infectious_four_test_result"),
    _f("infectiousFourExternal", "传染四项检测（外检）", DEPT_QC, SOURCE_EXTRA),
    _f("preEnrichTnc", "富集前有核细胞数/109/L", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("enrichedBloodWeightG", "富集后血量/g", DEPT_PRODUCTION, SOURCE_PRODUCTION, "enriched_blood_weight_g"),
    _f("postEnrichTnc", "富集后有核细胞数/109/L", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("cryoVolumeMl", "冻存液用量/ml", DEPT_PRODUCTION, SOURCE_PRODUCTION, "cryopreservation_volume_ml"),
    _f("cd34Count", "CD34+细胞计数/*10E4", DEPT_QC, SOURCE_EXTRA),
    _f("operatorName", "操作人", DEPT_PRODUCTION, SOURCE_PRODUCTION, "handler_name"),
    _f("processRemark", "备注", DEPT_PRODUCTION, SOURCE_PRODUCTION, "remark"),
)

UBMC_FIELDS: Tuple[FieldDef, ...] = (
    _f("sampleCode", "细胞编码", DEPT_MARKET, SOURCE_PROTOCOL, "sample_code"),
    _f("motherName", "准妈妈姓名", DEPT_MARKET, SOURCE_PROTOCOL, "mother_name"),
    _f("labMaterialCode", "实验室物料编码", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("labStorageCode", "入库编码", DEPT_PRODUCTION, SOURCE_RECEPTION, "lab_storage_code"),
    _f("storageStatus", "储存状态", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("discardReason", "废弃原因", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("receptionTime", "接收时间", DEPT_PRODUCTION, SOURCE_RECEPTION, "reception_time"),
    _f("receiveTemperature", "样品温度到达/℃", DEPT_PRODUCTION, SOURCE_RECEPTION, "receive_temperature"),
    _f("receiveCheck", "收样检查", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("receptionHandler", "接收登记人", DEPT_PRODUCTION, SOURCE_RECEPTION, "handler_name"),
    _f("receptionRemark", "备注", DEPT_PRODUCTION, SOURCE_RECEPTION, "remark"),
    _f("originalBloodVolumeMl", "原血体积/mL", DEPT_PRODUCTION, SOURCE_PRODUCTION, "original_blood_volume_ml"),
    _f("nucleatedCellCount1e8", "有核细胞数/108", DEPT_PRODUCTION, SOURCE_PRODUCTION, "nucleated_cell_count_1e8"),
    _f("preOpTnc", "操作前有核细胞数/109/L", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("infectiousFourCard", "传染四项检测（检测卡）", DEPT_QC, SOURCE_PRODUCTION, "infectious_four_test_result"),
    _f("hbsag", "乙型肝炎病毒表面抗原(HBsAg)", DEPT_QC, SOURCE_EXTRA),
    _f("antiHcv", "丙型肝炎病毒抗体(抗HCV)", DEPT_QC, SOURCE_EXTRA),
    _f("antiTp", "梅毒螺旋体特异性抗体(抗TP)", DEPT_QC, SOURCE_EXTRA),
    _f("antiHiv", "人类免疫缺陷病毒抗体(抗HIV-1/2)", DEPT_QC, SOURCE_EXTRA),
    _f("cmvIgm", "巨细胞病毒IgM 抗体 (CMV-IgM)", DEPT_QC, SOURCE_EXTRA),
    _f("operatorName", "操作人", DEPT_PRODUCTION, SOURCE_PRODUCTION, "handler_name"),
    _f("processRemark", "备注", DEPT_PRODUCTION, SOURCE_PRODUCTION, "remark"),
)

MSC_FIELDS: Tuple[FieldDef, ...] = (
    _f("sampleCode", "实验室流转编码", DEPT_MARKET, SOURCE_PROTOCOL, "sample_code"),
    _f("motherName", "母亲姓名", DEPT_MARKET, SOURCE_PROTOCOL, "mother_name"),
    _f("collectTime", "脐带采集时间", DEPT_MARKET, SOURCE_PROTOCOL, "sample_collect_time"),
    _f("transportTempC", "运输温度/℃", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("receptionTime", "脐带接收时间", DEPT_PRODUCTION, SOURCE_RECEPTION, "reception_time"),
    _f("receptionHandler", "接收登记人", DEPT_PRODUCTION, SOURCE_RECEPTION, "handler_name"),
    _f("inoculationAmount", "接种量", DEPT_PRODUCTION, SOURCE_PRODUCTION, "inoculation_amount"),
    _f("storageStatus", "储存状态", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("cryoTime", "冻存时间", DEPT_PRODUCTION, SOURCE_PRODUCTION, "cryopreservation_time"),
    _f("storageTotalTubes", "储存总量（管）", DEPT_PRODUCTION, SOURCE_PRODUCTION, "storage_total_tubes"),
    _f("availableCellCount1e7", "可提供细胞数量/*10E7", DEPT_PRODUCTION, SOURCE_PRODUCTION, "available_cell_count_1e7"),
    _f("endotoxinTest", "内毒素检测", DEPT_QC, SOURCE_EXTRA),
    _f("bacteriaTest", "细菌检测", DEPT_QC, SOURCE_EXTRA),
    _f("fungusTest", "真菌检测", DEPT_QC, SOURCE_EXTRA),
    _f("mycoplasmaTest", "支原体检测", DEPT_QC, SOURCE_EXTRA),
    _f("operatorName", "操作人", DEPT_PRODUCTION, SOURCE_PRODUCTION, "handler_name"),
    _f("motherHbsag", "乙型肝炎病毒表面抗原(HBsAg)", DEPT_QC, SOURCE_EXTRA),
    _f("motherHbvDna", "乙型肝炎病毒核酸(HBV DNA)", DEPT_QC, SOURCE_EXTRA),
    _f("motherAntiHcv", "丙型肝炎病毒抗体(抗HCV)", DEPT_QC, SOURCE_EXTRA),
    _f("motherHcvRna", "丙型肝炎病毒核酸(HCV RNA)", DEPT_QC, SOURCE_EXTRA),
    _f("motherAntiTp", "梅毒螺旋体特异性抗体(抗TP)", DEPT_QC, SOURCE_EXTRA),
    _f("motherCmvIgm", "巨细胞病毒lgM 抗体 (CMV lgM)", DEPT_QC, SOURCE_EXTRA),
    _f("motherAntiHiv", "人类免疫缺陷病毒抗体(抗HIV-1/2)", DEPT_QC, SOURCE_EXTRA),
    _f("motherHiv1Rna", "人类免疫缺陷病毒核酸(1型) (HIV-1RNA)", DEPT_QC, SOURCE_EXTRA),
    _f("storageDateP2", "储存日期(P2冻存日期）", DEPT_PRODUCTION, SOURCE_EXTRA),
    _f("processRemark", "备注", DEPT_PRODUCTION, SOURCE_PRODUCTION, "remark"),
)

FIELDS_BY_DATASET: Dict[str, Tuple[FieldDef, ...]] = {
    DATASET_HSC: HSC_FIELDS,
    DATASET_UBMC: UBMC_FIELDS,
    DATASET_MSC: MSC_FIELDS,
}

DATASET_LABELS = {
    DATASET_HSC: "造血干细胞数据",
    DATASET_UBMC: "单个核数据",
    DATASET_MSC: "脐带间充质干细胞数据",
}


def fields_for(dataset: str) -> Tuple[FieldDef, ...]:
    return FIELDS_BY_DATASET.get(dataset) or ()


def field_map(dataset: str) -> Dict[str, FieldDef]:
    return {f.key: f for f in fields_for(dataset)}


def editable_depts_for_user(*, is_manager: bool, dept_flags: Dict[str, bool]) -> set:
    if is_manager:
        return set(DEPTS)
    return {d for d, ok in dept_flags.items() if ok}


def resolve_dataset_by_type_name(name: Optional[str], sample_code: Optional[str] = None) -> Optional[str]:
    text = (name or "").strip()
    code = (sample_code or "").strip().upper()
    if "单个核" in text or "脐血单个核" in text or code.startswith("UBMC"):
        return DATASET_UBMC
    if "间充质" in text or code.startswith("BC") or code.startswith("UC"):
        return DATASET_MSC
    if "造血" in text or code.startswith("B"):
        return DATASET_HSC
    return None


def schema_payload(dataset: str) -> List[Dict[str, str]]:
    return [
        {"key": f.key, "label": f.label, "dept": f.dept, "source": f.source}
        for f in fields_for(dataset)
    ]