cykj/zkxw
1
0 Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

fix(lab-sample): 新旧实验室样本台账补录双向联动

Browse Source 
新模块读取旧台账补录并映射字段；旧台账保存时同步到新补充表与生产传染四项，新模块保存时回写旧补录。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-11-12 17:19:19 +08:00
1 parent 2aecb69cc4
commit c0aba421da
3 files changed
+339 -2
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/crud/lab_sample_process.py
+11
View File
@@ -179,6 +179,17 @@ def upsert_supplement(
del_flag=0,
)
db.add(row)
# 同步到新模块补充表/生产接收，保证实验室样本处理数据联动
from app.services.lab_sample_processing import sync_legacy_supplement_to_processing
sync_legacy_supplement_to_processing(
db,
tenant_no=tenant_no,
lab_storage_code=code,
sample_type_name=type_name,
payload=clean_payload,
operator_id=user_id,
)
db.commit()
db.refresh(row)
return get_supplement(db, tenant_no=tenant_no, lab_storage_code=code) or {}
backend/app/services/lab_sample_processing.py
+280 -2
View File
@@ -206,6 +206,240 @@ def _pick_extra(
return {}
# 旧台账补录字段名 → 新模块字段名（同名键保持不变）
_LEGACY_SUPPLEMENT_TO_NEW: Dict[str, str] = {
"receiveCheckText": "receiveCheck",
"infectiousFourTestLabel": "infectiousFourCard",
"preOpNucleatedCount": "preOpTnc",
"preEnrichNucleatedCount": "preEnrichTnc",
"postEnrichNucleatedCount": "postEnrichTnc",
"cd34Count1e4": "cd34Count",
"transportPackingList": "hasPackingList",
"cordBloodReceiveConfirm": "hasReceiveConfirm",
"remarkProcess": "processRemark",
"motherHbsag": "hbsag",
"motherAntiHcv": "antiHcv",
"motherAntiTp": "antiTp",
"motherAntiHiv": "antiHiv",
"motherCmvIgm": "cmvIgm",
}
_NEW_TO_LEGACY_SUPPLEMENT: Dict[str, str] = {
v: k for k, v in _LEGACY_SUPPLEMENT_TO_NEW.items()
}
# MSC 等仍使用 mother* 键名，双向都保留
for _same in (
"labMaterialCode",
"storageStatus",
"discardReason",
"hasMaternalBlood",
"remeasureMaternalBlood",
"infectiousFourExternal",
"endotoxinTest",
"bacteriaTest",
"fungusTest",
"mycoplasmaTest",
"motherHbsag",
"motherHbvDna",
"motherAntiHcv",
"motherHcvRna",
"motherAntiTp",
"motherCmvIgm",
"motherAntiHiv",
"motherHiv1Rna",
"storageDateP2",
"storageTotalTubes",
"availableCellCount1e7",
):
_LEGACY_SUPPLEMENT_TO_NEW.setdefault(_same, _same)
_NEW_TO_LEGACY_SUPPLEMENT.setdefault(_same, _same)
def _is_blank(val: Any) -> bool:
return val is None or (isinstance(val, str) and not val.strip())
def _map_legacy_supplement(payload: Optional[Dict[str, Any]]) -> Dict[str, Any]:
out: Dict[str, Any] = {}
for key, value in (payload or {}).items():
if _is_blank(value):
continue
new_key = _LEGACY_SUPPLEMENT_TO_NEW.get(str(key), str(key))
out[new_key] = value
# MSC 场景保留 mother* 原键，避免被 UBMC 别名覆盖后丢失
if str(key).startswith("mother") and new_key != str(key):
out[str(key)] = value
return out
def _map_new_to_legacy_supplement(payload: Optional[Dict[str, Any]]) -> Dict[str, Any]:
out: Dict[str, Any] = {}
for key, value in (payload or {}).items():
if _is_blank(value):
continue
legacy_key = _NEW_TO_LEGACY_SUPPLEMENT.get(str(key), str(key))
out[legacy_key] = value
return out
def _merge_extra_with_legacy(
extra: Optional[Dict[str, Any]], legacy: Optional[Dict[str, Any]]
) -> Dict[str, Any]:
"""新补充表优先；旧台账补录填空（含字段名映射）。"""
mapped = _map_legacy_supplement(legacy)
merged = dict(mapped)
for key, value in (extra or {}).items():
if not _is_blank(value):
merged[key] = value
return merged
def _load_legacy_supplements(
db: Session, tenant_no: str, codes: List[str]
) -> Dict[str, Dict[str, Any]]:
from app.models.models import BizLabSampleProcessSupplement
from app.services import lab_sample_process as legacy_ledger
uniq = [c for c in {(x or "").strip() for x in codes} if c]
if not uniq:
return {}
rows = (
db.query(BizLabSampleProcessSupplement)
.filter(
BizLabSampleProcessSupplement.tenant_no == tenant_no,
BizLabSampleProcessSupplement.del_flag == 0,
BizLabSampleProcessSupplement.lab_storage_code.in_(uniq),
)
.all()
)
out: Dict[str, Dict[str, Any]] = {}
for row in rows:
out[row.lab_storage_code] = legacy_ledger.parse_payload(row.payload)
return out
def sync_legacy_supplement_to_processing(
db: Session,
*,
tenant_no: str,
lab_storage_code: str,
sample_type_name: Optional[str] = None,
payload: Optional[Dict[str, Any]] = None,
operator_id: str = "legacy-sync",
) -> None:
"""旧台账补录保存后，同步到新模块补充表，并把传染四项写回生产接收。"""
code = (lab_storage_code or "").strip()
if not code:
return
mapped = _map_legacy_supplement(payload)
if not mapped:
return
dataset = resolve_dataset_by_type_name(sample_type_name or "", code)
if not dataset:
if code.upper().startswith("UBMC"):
dataset = DATASET_UBMC
elif code.upper().startswith(("BC", "UC")):
dataset = DATASET_MSC
else:
dataset = DATASET_HSC
upsert_extra_payload(
db,
tenant_no=tenant_no,
dataset=dataset,
sample_code=code,
payload=mapped,
operator_id=operator_id,
replace=False,
)
infectious = mapped.get("infectiousFourCard")
if infectious not in (None, ""):
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()
)
if reception:
production = (
db.query(BizSampleProductionReception)
.filter(
BizSampleProductionReception.tenant_no == tenant_no,
BizSampleProductionReception.reception_id == reception.id,
BizSampleProductionReception.del_flag == 0,
)
.order_by(BizSampleProductionReception.process_time.desc())
.first()
)
if production is not None:
production.infectious_four_test_result = _infectious_code(infectious)
production.update_by = operator_id
production.update_time = datetime.now()
def sync_processing_to_legacy_supplement(
db: Session,
*,
tenant_no: str,
lab_storage_code: str,
sample_type_name: Optional[str] = None,
payload: Optional[Dict[str, Any]] = None,
operator_id: str = "processing-sync",
) -> None:
"""新模块保存后，回写旧台账补录，保持两边可见字段一致。"""
from app.models.models import BizLabSampleProcessSupplement
from app.services import lab_sample_process as legacy_ledger
code = (lab_storage_code or "").strip()
mapped = _map_new_to_legacy_supplement(payload)
if not code or not mapped:
return
now = datetime.now()
row = (
db.query(BizLabSampleProcessSupplement)
.filter(
BizLabSampleProcessSupplement.tenant_no == tenant_no,
BizLabSampleProcessSupplement.lab_storage_code == code,
BizLabSampleProcessSupplement.del_flag == 0,
)
.first()
)
current = legacy_ledger.parse_payload(row.payload) if row else {}
current.update(mapped)
src = "research_donation" if code.upper().startswith("UBMC") else "cell_storage"
if row is None:
db.add(
BizLabSampleProcessSupplement(
id=_new_id(),
lab_storage_code=code,
reception_source=src,
sample_type_name=(sample_type_name or None),
payload=legacy_ledger.dumps_payload(current),
tenant_no=tenant_no,
create_by=operator_id,
create_dept_id="",
create_time=now,
update_by=operator_id,
update_time=now,
del_flag=0,
)
)
else:
if sample_type_name:
row.sample_type_name = sample_type_name
row.payload = legacy_ledger.dumps_payload(current)
row.update_by = operator_id
row.update_time = now
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
@@ -384,13 +618,24 @@ def list_rows(
)
check_flags_by_reception = _check_flag_map(db, tenant_no, reception_ids_for_flags)
legacy_codes: List[str] = []
for sample, *_ in filtered:
legacy_codes.append(sample.sample_code or "")
reception = receptions_by_sample_id.get(sample.id) or receptions_by_code.get(sample.sample_code or "")
if reception and reception.lab_storage_code:
legacy_codes.append(reception.lab_storage_code)
legacy_supplements = _load_legacy_supplements(db, tenant_no, legacy_codes)
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
lab_code = (reception.lab_storage_code if reception else None) or ""
extra = _pick_extra(extras, lab_code, code)
extra = _merge_extra_with_legacy(
_pick_extra(extras, lab_code, code),
legacy_supplements.get(lab_code) or legacy_supplements.get(code or ""),
)
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(
@@ -466,12 +711,26 @@ def list_rows(
)
check_flags_by_reception.update(ubmc_flags)
ubmc_legacy = _load_legacy_supplements(
db,
tenant_no,
[
(r.lab_storage_code or r.sample_code or "")
for r in ubmc_recs
],
)
legacy_supplements.update(ubmc_legacy)
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 = _pick_extra(extras, code, reception.sample_code)
extra = _merge_extra_with_legacy(
_pick_extra(extras, code, reception.sample_code),
legacy_supplements.get(code)
or legacy_supplements.get((reception.sample_code or "").strip()),
)
mother = mother_by_donation.get(str(reception.research_donation_id or ""))
fake_sample = SimpleNamespace(
id=reception.donation_sample_id or reception.id,
@@ -596,6 +855,12 @@ def _build_row(
# 仅源系统没有的列走补充表；业务源表字段不被补充表覆盖
val = extra.get(f.key)
# 源表为空时，允许用新/旧补充表填空（旧台账联动）
if f.source != SOURCE_EXTRA and _is_blank(val):
fallback = extra.get(f.key)
if not _is_blank(fallback):
val = fallback
if isinstance(val, datetime):
val = _fmt_dt(val)
elif isinstance(val, Decimal):
@@ -894,6 +1159,19 @@ def update_row(
operator_id=operator_id,
replace=False,
)
legacy_code = (
(reception.lab_storage_code if reception else None)
or (sample.sample_code if sample else None)
or code
)
sync_processing_to_legacy_supplement(
db,
tenant_no=tenant_no,
lab_storage_code=str(legacy_code),
sample_type_name=None,
payload=extra_payload,
operator_id=operator_id,
)
db.commit()
rows, _ = list_rows(db, tenant_no=tenant_no, dataset=dataset, keyword=code, page=1, page_size=20)
backend/tests/test_lab_sample_processing.py
+48
View File
@@ -133,6 +133,54 @@ class LabSampleProcessingFieldsTest(unittest.TestCase):
self.assertEqual(_infectious_code("正常"), "zc")
self.assertEqual(_infectious_code("不正常"), "bzc")
def test_legacy_supplement_fills_empty_source_fields(self):
from app.services.lab_sample_processing import _build_row, _merge_extra_with_legacy
from types import SimpleNamespace
extra = _merge_extra_with_legacy(
{},
{"infectiousFourTestLabel": "正常", "labMaterialCode": "202611006"},
)
sample = SimpleNamespace(id="S1", sample_code="UBMC202611006", collection_protocol_id=None)
reception = SimpleNamespace(
id="R1",
reception_time=None,
receive_temperature=None,
handler_name=None,
remark=None,
lab_storage_code="UBMC202611006",
)
production = SimpleNamespace(
id="PR1",
original_blood_weight_g=None,
hes_volume_ml=None,
blood_type=None,
infectious_four_test_result=None,
enriched_blood_weight_g=None,
cryopreservation_volume_ml=None,
handler_name=None,
remark=None,
original_blood_volume_ml=None,
nucleated_cell_count_1e8=None,
inoculation_amount=None,
cryopreservation_time=None,
storage_total_tubes=None,
available_cell_count_1e7=None,
)
row = _build_row(
dataset=DATASET_UBMC,
sample=sample,
type_name="脐血单个核细胞",
party=None,
coll=None,
hospital=None,
reception=reception,
production=production,
extra=extra,
)
self.assertEqual(row["infectiousFourCard"], "正常")
self.assertEqual(row["labMaterialCode"], "202611006")
if __name__ == "__main__":
unittest.main()