From e6763241d15af70166ff4c7a3c0d6f6ddabe653f Mon Sep 17 00:00:00 2001
From: zlwu <704541015@qq.com>
Date: Fri, 9 Oct 2026 12:33:50 +0800
Subject: [PATCH] feat(backend): add storage-cert QC type-key and payload
 helpers.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../app/services/sample_storage_cert_qc.py    | 78 +++++++++++++++++++
 backend/tests/test_sample_storage_cert_qc.py  | 59 ++++++++++++++
 2 files changed, 137 insertions(+)
 create mode 100644 backend/app/services/sample_storage_cert_qc.py
 create mode 100644 backend/tests/test_sample_storage_cert_qc.py

diff --git a/backend/app/services/sample_storage_cert_qc.py b/backend/app/services/sample_storage_cert_qc.py
new file mode 100644
index 00000000..a4178a5a
--- /dev/null
+++ b/backend/app/services/sample_storage_cert_qc.py
@@ -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": {},
+    }
diff --git a/backend/tests/test_sample_storage_cert_qc.py b/backend/tests/test_sample_storage_cert_qc.py
new file mode 100644
index 00000000..542b2b79
--- /dev/null
+++ b/backend/tests/test_sample_storage_cert_qc.py
@@ -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()