feat(backend): add storage-cert QC type-key and payload helpers.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-10-09 12:33:50 +08:00
1 parent e521d781e3
commit e6763241d1
2 files changed
+137

No files matched your search

@@ -0,0 +1,78 @@
"""样本储存证质检信息 — 类型键与 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": {},
}
@@ -0,0 +1,59 @@
"""储存证质检信息:类型键与 payload 规范化。"""
import unittest
from app.services.sample_storage_cert_qc import (
SAMPLE_TYPE_KEY_PERINATAL_HSC,
SAMPLE_TYPE_KEY_UC_MSC,
empty_detail,
normalize_payload,
resolve_sample_type_key,
supports_storage_cert_qc,
)
class TestResolveSampleTypeKey(unittest.TestCase):
def test_hsc(self):
self.assertEqual(resolve_sample_type_key("围产期造血干细胞"), SAMPLE_TYPE_KEY_PERINATAL_HSC)
def test_msc(self):
self.assertEqual(resolve_sample_type_key("脐带间充质干细胞"), SAMPLE_TYPE_KEY_UC_MSC)
def test_other(self):
self.assertIsNone(resolve_sample_type_key("胎盘亚全能干细胞"))
self.assertIsNone(resolve_sample_type_key(""))
def test_supports(self):
self.assertTrue(supports_storage_cert_qc("围产期造血干细胞"))
self.assertFalse(supports_storage_cert_qc("其他"))
class TestNormalizePayload(unittest.TestCase):
def test_rejects_non_object(self):
with self.assertRaises(ValueError):
normalize_payload([])
with self.assertRaises(ValueError):
normalize_payload("x")
def test_keeps_dict_stringifies_keys(self):
out = normalize_payload({"viability_pct": 90.4, 1: "bad"})
self.assertEqual(out["viability_pct"], 90.4)
self.assertNotIn(1, out)
class TestEmptyDetail(unittest.TestCase):
def test_shape(self):
d = empty_detail(
sample_code="B1",
protocol_code="P1",
mother_name="张三",
sample_type_name="围产期造血干细胞",
sample_type_key="perinatal_hsc",
)
self.assertFalse(d["filledFlag"])
self.assertEqual(d["payload"], {})
self.assertEqual(d["sampleCode"], "B1")
self.assertEqual(d["labFlowCode"], "B1")
if __name__ == "__main__":
unittest.main()