diff --git a/backend/app/crud/qc_work_order.py b/backend/app/crud/qc_work_order.py index 884954d7..5b17b3a2 100644 --- a/backend/app/crud/qc_work_order.py +++ b/backend/app/crud/qc_work_order.py @@ -23,6 +23,7 @@ from app.crud.qc_inbound_lab_storage_sort import ( from app.crud.collection_protocol import urls_to_attachment_list from app.utils.sample_special import apply_special_to_item_from_samples from app.crud import sample_reception as sample_reception_svc +from app.crud.temp_qc_target import resolve_temp_qc_target from app.crud.sample_reception import _donation_sample_id_col from app.crud import sample_reception_check_result as check_result_svc from app.crud.cell_product import _enrich_product_dicts as _cp_enrich, _row_to_dict as _cp_row_to_dict, get_cell_product diff --git a/backend/app/crud/temp_qc_target.py b/backend/app/crud/temp_qc_target.py new file mode 100644 index 00000000..74b9db82 --- /dev/null +++ b/backend/app/crud/temp_qc_target.py @@ -0,0 +1,22 @@ +"""临时质检对象:系统产品 / 外部编号互斥归一化。""" + +from typing import Optional, Tuple + + +def resolve_temp_qc_target( + *, + cell_product_id: Optional[str] = None, + external_target_code: Optional[str] = None, +) -> Tuple[Optional[str], Optional[str]]: + """临时质检对象互斥归一化。返回 (cell_product_id|None, external_target_code|None)。""" + cp = (cell_product_id or "").strip() or None + ext = (external_target_code or "").strip() or None + if cp and ext: + raise ValueError("质检对象不能同时选择系统产品并填写外部编号") + if not cp and not ext: + raise ValueError("请选择质检对象或填写外部编号") + if ext and len(ext) > 64: + raise ValueError("外部编号最长 64 个字符") + if ext: + return None, ext + return cp, None diff --git a/backend/app/models/models.py b/backend/app/models/models.py index 4523b93a..c7480bae 100644 --- a/backend/app/models/models.py +++ b/backend/app/models/models.py @@ -1685,6 +1685,7 @@ class BizQcWorkOrder(Base): infusion_schedule_id = Column(String(64), nullable=True, comment="中试输注计划ID(出库质检关联)") biz_type = Column(String(20), nullable=True, comment="业务类型: pilot_service=中试服务, clinical_application=临床应用") cell_product_id = Column(String(64), nullable=True, comment="质检对象-细胞产品ID") + external_target_code = Column(String(64), nullable=True, comment="外部质检对象编号(手写)") production_work_order_id = Column(String(64), nullable=True, comment="关联生产工单ID(生产过程触发时)") production_task_id = Column(String(64), nullable=True, comment="关联生产任务ID(可选)") production_step_output_id = Column(String(64), nullable=True, comment="关联生产任务步骤产出物ID(生产过程触发时)") diff --git a/backend/tests/test_temp_qc_target_resolve.py b/backend/tests/test_temp_qc_target_resolve.py new file mode 100644 index 00000000..029ab794 --- /dev/null +++ b/backend/tests/test_temp_qc_target_resolve.py @@ -0,0 +1,44 @@ +"""临时质检对象:系统产品 / 外部编号互斥归一化。""" + +import unittest + +from app.crud.temp_qc_target import resolve_temp_qc_target + + +class TestResolveTempQcTarget(unittest.TestCase): + def test_cell_product_only(self): + cp, ext = resolve_temp_qc_target( + cell_product_id=" pid-1 ", external_target_code=" " + ) + self.assertEqual(cp, "pid-1") + self.assertIsNone(ext) + + def test_external_only(self): + cp, ext = resolve_temp_qc_target( + cell_product_id="", external_target_code=" EXT-1 " + ) + self.assertIsNone(cp) + self.assertEqual(ext, "EXT-1") + + def test_both_raises(self): + with self.assertRaises(ValueError) as ctx: + resolve_temp_qc_target( + cell_product_id="pid", external_target_code="EXT" + ) + self.assertIn("不能同时", str(ctx.exception)) + + def test_neither_raises(self): + with self.assertRaises(ValueError) as ctx: + resolve_temp_qc_target(cell_product_id="", external_target_code="") + self.assertIn("质检对象", str(ctx.exception)) + + def test_external_too_long(self): + with self.assertRaises(ValueError) as ctx: + resolve_temp_qc_target( + cell_product_id="", external_target_code="x" * 65 + ) + self.assertIn("64", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main()