diff --git a/backend/app/services/lab_sample_processing.py b/backend/app/services/lab_sample_processing.py index fccd5fca..66c8d6f0 100644 --- a/backend/app/services/lab_sample_processing.py +++ b/backend/app/services/lab_sample_processing.py @@ -19,11 +19,13 @@ from sqlalchemy.orm import Session from app.models.models import ( BizCellSampleType, BizCollectionHospital, + BizCollectionProtocol, BizProtocolCollectionInfo, BizProtocolPartyCustomer, BizProtocolSample, BizSampleProductionReception, BizSampleReception, + BizSampleReceptionCheckResult, ) from app.services.lab_sample_processing_fields import ( DATASET_HSC, @@ -112,10 +114,98 @@ def _code_prefix_filters(dataset: str) -> List[str]: def _match_keyword(row: Dict[str, Any], kw: str) -> bool: if not kw: return True - text = f"{row.get('sampleCode') or ''} {row.get('motherName') or ''}" + text = ( + f"{row.get('sampleCode') or ''} " + f"{row.get('protocolSampleCode') or ''} " + f"{row.get('motherName') or ''}" + ) return kw.lower() in text.lower() +def _infectious_label(code: Optional[str]) -> Optional[str]: + c = ("" if code is None else str(code)).strip() + if not c: + return None + if c == "zc": + return "正常" + if c == "bzc": + return "不正常" + return c + + +def _infectious_code(label: Optional[Any]) -> Optional[str]: + if label is None: + return None + s = str(label).strip() + if not s: + return None + if s == "正常": + return "zc" + if s == "不正常": + return "bzc" + return s + + +def _yes_no_from_bool(flag: Optional[bool]) -> Optional[str]: + if flag is None: + return None + return "是" if flag else "否" + + +def _derive_receive_check(flags: Dict[str, bool]) -> Optional[str]: + if not flags: + return None + skip = {"知情同意书", "健康调查表", "样本信息正确，标识清晰"} + preferred = [ + "针路密闭", + "容器无破损", + "外观正常", + "样本无破损", + "针卡关紧", + "两端系紧", + "温度4-25", + "采集至到库时间≤24小时", + ] + parts = [name for name in preferred if flags.get(name)] + if not parts: + parts = [name for name, checked in flags.items() if checked and name not in skip] + return "、".join(parts) if parts else None + + +def _check_flag_map( + db: Session, tenant_no: str, reception_ids: List[str] +) -> Dict[str, Dict[str, bool]]: + ids = [i for i in reception_ids if i] + if not ids: + return {} + rows = ( + db.query(BizSampleReceptionCheckResult) + .filter( + BizSampleReceptionCheckResult.tenant_no == tenant_no, + BizSampleReceptionCheckResult.del_flag == 0, + BizSampleReceptionCheckResult.reception_id.in_(ids), + ) + .all() + ) + out: Dict[str, Dict[str, bool]] = {} + for r in rows: + bucket = out.setdefault(r.reception_id, {}) + name = (r.item_name or "").strip() + if name: + bucket[name] = bool(r.checked) + return out + + +def _pick_extra( + extras: Dict[str, Any], *codes: Optional[str] +) -> Dict[str, Any]: + for code in codes: + key = (code or "").strip() + if key and key in extras: + return _parse_extra(extras[key].payload_json) + return {} + + def _match_date_range(row: Dict[str, Any], date_from: Optional[str], date_to: Optional[str]) -> bool: if not date_from and not date_to: return True @@ -266,13 +356,44 @@ def list_rows( ): productions.setdefault(p.reception_id, p) + protocol_ids = sorted( + { + s.collection_protocol_id + for s, *_ in filtered + if getattr(s, "collection_protocol_id", None) + } + ) + protocols: Dict[str, BizCollectionProtocol] = {} + if protocol_ids: + for p in ( + db.query(BizCollectionProtocol) + .filter( + BizCollectionProtocol.tenant_no == tenant_no, + BizCollectionProtocol.del_flag == 0, + BizCollectionProtocol.id.in_(protocol_ids), + ) + .all() + ): + protocols[p.id] = p + + reception_ids_for_flags = list( + { + *(r.id for r in receptions_by_sample_id.values() if r and r.id), + *(r.id for r in receptions_by_code.values() if r and r.id), + } + ) + check_flags_by_reception = _check_flag_map(db, tenant_no, reception_ids_for_flags) + built_map: Dict[str, Dict[str, Any]] = {} for sample, stype, party, coll, hospital in filtered: code = sample.sample_code reception = receptions_by_sample_id.get(sample.id) or receptions_by_code.get(code) production = productions.get(reception.id) if reception else None - extra = _parse_extra(extras[code].payload_json if code in extras else None) - built_map[code] = _build_row( + lab_code = (reception.lab_storage_code if reception else None) or "" + extra = _pick_extra(extras, lab_code, code) + protocol = protocols.get(sample.collection_protocol_id) if sample.collection_protocol_id else None + check_flags = check_flags_by_reception.get(reception.id, {}) if reception else {} + row = _build_row( dataset=dataset, sample=sample, type_name=(stype.type_name if stype else None), @@ -282,7 +403,10 @@ def list_rows( reception=reception, production=production, extra=extra, + protocol=protocol, + check_flags=check_flags, ) + built_map[row.get("sampleCode") or code] = row # 单个核：业务已发生数据主要在科研捐献接收（lab_storage_code=UBMC*），无细胞储存协议样本时仍需展示 if dataset == DATASET_UBMC: @@ -335,16 +459,23 @@ def list_rows( ): productions.setdefault(p.reception_id, p) + ubmc_flags = _check_flag_map( + db, + tenant_no, + [r.id for r in ubmc_recs if r.id and r.id not in check_flags_by_reception], + ) + check_flags_by_reception.update(ubmc_flags) + for reception in ubmc_recs: code = (reception.lab_storage_code or reception.sample_code or "").strip() if not code or code in built_map: continue production = productions.get(reception.id) - extra = _parse_extra(extras[code].payload_json if code in extras else None) + extra = _pick_extra(extras, code, reception.sample_code) mother = mother_by_donation.get(str(reception.research_donation_id or "")) fake_sample = SimpleNamespace( id=reception.donation_sample_id or reception.id, - sample_code=code, + sample_code=reception.sample_code or code, collection_protocol_id=None, ) fake_party = SimpleNamespace(mother_name=mother) if mother else None @@ -358,6 +489,8 @@ def list_rows( reception=reception, production=production, extra=extra, + protocol=None, + check_flags=check_flags_by_reception.get(reception.id, {}), ) kw = (keyword or "").strip() @@ -390,9 +523,23 @@ def _build_row( reception: Optional[BizSampleReception], production: Optional[BizSampleProductionReception], extra: Dict[str, Any], + protocol: Optional[BizCollectionProtocol] = None, + check_flags: Optional[Dict[str, bool]] = None, ) -> Dict[str, Any]: + flags = check_flags or {} + protocol_code = ( + (sample.sample_code if sample else None) + or extra.get("protocolSampleCode") + or "" + ) + lab_code = ((reception.lab_storage_code if reception else None) or "").strip() + # 台账「细胞编码」展示实验室入库编码（B000…），协议样本号保留检索 + display_code = lab_code if lab_code and len(lab_code) > 1 else ( + protocol_code or extra.get("sampleCode") or "" + ) + protocol_vals = { - "sample_code": sample.sample_code if sample else extra.get("sampleCode"), + "sample_code": display_code, "mother_name": party.mother_name if party else None, "sample_collect_time": coll.sample_collect_time if coll else None, "hospital_name": hospital.hospital_name if hospital else None, @@ -425,15 +572,11 @@ def _build_row( "available_cell_count_1e7": production.available_cell_count_1e7, } - sample_code = ( - (sample.sample_code if sample else None) - or extra.get("sampleCode") - or "" - ) data: Dict[str, Any] = { - "id": (sample.id if sample else f"extra:{dataset}:{sample_code}"), + "id": (sample.id if sample else f"extra:{dataset}:{display_code}"), "dataset": dataset, - "sampleCode": sample_code, + "sampleCode": display_code, + "protocolSampleCode": protocol_code or None, "sampleTypeName": type_name, "protocolSampleId": sample.id if sample else None, "collectionProtocolId": sample.collection_protocol_id if sample else None, @@ -458,8 +601,43 @@ def _build_row( elif isinstance(val, Decimal): val = _fmt_num(val) data[f.key] = val - if not data.get("sampleCode"): - data["sampleCode"] = sample_code + + data["sampleCode"] = display_code + + # 从协议采集信息 / 附件 / 收样检查项推导补充列（补充表有值时不覆盖） + if not data.get("hasMaternalBlood"): + if coll is not None: + data["hasMaternalBlood"] = "是" if coll.maternal_blood_collect_time else "否" + + if not data.get("hasConsent"): + if protocol and (protocol.consent_form_urls or "").strip(): + data["hasConsent"] = "是" + else: + flagged = _yes_no_from_bool(flags.get("知情同意书")) + if flagged is not None: + data["hasConsent"] = flagged + elif protocol is not None: + data["hasConsent"] = "否" + + if not data.get("hasHealthSurvey"): + if protocol and (protocol.health_survey_form_urls or "").strip(): + data["hasHealthSurvey"] = "是" + else: + flagged = _yes_no_from_bool(flags.get("健康调查表")) + if flagged is not None: + data["hasHealthSurvey"] = flagged + elif protocol is not None: + data["hasHealthSurvey"] = "否" + + if not data.get("labelClear"): + data["labelClear"] = _yes_no_from_bool(flags.get("样本信息正确，标识清晰")) + + if not data.get("receiveCheck"): + data["receiveCheck"] = _derive_receive_check(flags) + + if data.get("infectiousFourCard") is not None: + data["infectiousFourCard"] = _infectious_label(data.get("infectiousFourCard")) + return data @@ -549,6 +727,19 @@ def update_row( if not accepted: raise ValueError("没有可保存的字段") + reception = ( + db.query(BizSampleReception) + .filter( + BizSampleReception.tenant_no == tenant_no, + BizSampleReception.del_flag == 0, + or_( + BizSampleReception.lab_storage_code == code, + BizSampleReception.sample_code == code, + ), + ) + .order_by(BizSampleReception.reception_time.desc()) + .first() + ) sample = ( db.query(BizProtocolSample) .filter( @@ -558,16 +749,30 @@ def update_row( ) .first() ) - reception = ( - db.query(BizSampleReception) - .filter( - BizSampleReception.tenant_no == tenant_no, - BizSampleReception.sample_code == code, - BizSampleReception.del_flag == 0, + if sample is None and reception is not None and reception.protocol_sample_id: + sample = ( + db.query(BizProtocolSample) + .filter( + BizProtocolSample.tenant_no == tenant_no, + BizProtocolSample.id == reception.protocol_sample_id, + BizProtocolSample.del_flag == 0, + ) + .first() + ) + if reception is None and sample is not None: + reception = ( + db.query(BizSampleReception) + .filter( + BizSampleReception.tenant_no == tenant_no, + BizSampleReception.del_flag == 0, + or_( + BizSampleReception.protocol_sample_id == sample.id, + BizSampleReception.sample_code == sample.sample_code, + ), + ) + .order_by(BizSampleReception.reception_time.desc()) + .first() ) - .order_by(BizSampleReception.reception_time.desc()) - .first() - ) production = None if reception: production = ( @@ -661,9 +866,12 @@ def update_row( elif f.attr in numeric_attrs: production_val = None if value in (None, "") else Decimal(str(value)) setattr(production, f.attr, production_val) + elif f.attr == "infectious_four_test_result": + production.infectious_four_test_result = ( + None if value in (None, "") else _infectious_code(value) + ) elif f.attr in { "blood_type", - "infectious_four_test_result", "handler_name", "remark", "inoculation_amount", @@ -690,7 +898,7 @@ def update_row( rows, _ = list_rows(db, tenant_no=tenant_no, dataset=dataset, keyword=code, page=1, page_size=20) for row in rows: - if row.get("sampleCode") == code: + if row.get("sampleCode") == code or row.get("protocolSampleCode") == code: return row raise LookupError("保存成功但读取失败") diff --git a/backend/tests/test_lab_sample_processing.py b/backend/tests/test_lab_sample_processing.py index 07ee9fb0..786ddc89 100644 --- a/backend/tests/test_lab_sample_processing.py +++ b/backend/tests/test_lab_sample_processing.py @@ -49,13 +49,21 @@ class LabSampleProcessingFieldsTest(unittest.TestCase): self.assertEqual(fmap["originalBloodWeightG"].dept, DEPT_PRODUCTION) def test_source_fields_not_overridden_by_extra(self): + from datetime import datetime from app.services.lab_sample_processing import _build_row from types import SimpleNamespace sample = SimpleNamespace(id="S1", sample_code="ZKZX34000865", collection_protocol_id="P1") party = SimpleNamespace(mother_name="真实妈妈") - coll = SimpleNamespace(sample_collect_time=None) + coll = SimpleNamespace( + sample_collect_time=None, + maternal_blood_collect_time=datetime(2026, 11, 4, 5, 20), + ) hospital = SimpleNamespace(hospital_name="真实医院") + protocol = SimpleNamespace( + consent_form_urls="protocol-consent/a", + health_survey_form_urls="", + ) reception = SimpleNamespace( id="R1", reception_time=None, @@ -69,7 +77,7 @@ class LabSampleProcessingFieldsTest(unittest.TestCase): original_blood_weight_g=72.08, hes_volume_ml=None, blood_type="B", - infectious_four_test_result="正常", + infectious_four_test_result="zc", enriched_blood_weight_g=None, cryopreservation_volume_ml=None, handler_name="操作人", @@ -95,13 +103,36 @@ class LabSampleProcessingFieldsTest(unittest.TestCase): "originalBloodWeightG": 1, "cd34Count": "11.4", }, + protocol=protocol, + check_flags={ + "健康调查表": True, + "样本信息正确，标识清晰": True, + "针路密闭": True, + "容器无破损": True, + }, ) + self.assertEqual(row["sampleCode"], "B0000103") + self.assertEqual(row["protocolSampleCode"], "ZKZX34000865") self.assertEqual(row["motherName"], "真实妈妈") self.assertEqual(row["originalBloodWeightG"], 72.08) self.assertEqual(row["bloodType"], "B") self.assertEqual(row["cd34Count"], "11.4") + self.assertEqual(row["hasMaternalBlood"], "是") + self.assertEqual(row["hasConsent"], "是") + self.assertEqual(row["hasHealthSurvey"], "是") + self.assertEqual(row["labelClear"], "是") + self.assertEqual(row["infectiousFourCard"], "正常") + self.assertIn("针路密闭", row["receiveCheck"] or "") self.assertFalse(row["fromExtraOnly"]) + def test_infectious_code_roundtrip(self): + from app.services.lab_sample_processing import _infectious_code, _infectious_label + + self.assertEqual(_infectious_label("zc"), "正常") + self.assertEqual(_infectious_label("bzc"), "不正常") + self.assertEqual(_infectious_code("正常"), "zc") + self.assertEqual(_infectious_code("不正常"), "bzc") + if __name__ == "__main__": unittest.main()