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

Merge pull request 'fix(backend): 协议作废与样本接收废弃双向联动' (#40) from codex/issue-31-discard-sync-d7063ae6ff97 into V2.0

Browse Source 
fix(backend): 协议作废与样本接收废弃双向联动

Merge PR #40 for issue #31.
This commit was merged in pull request #40.
This commit is contained in:
cykj committed 2026-11-05 03:44:15 +00:00
2 parents 65aaaae885 e2e9059c17
commit 1fb25f4fc2
8 files changed
+820 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/core/protocol_status_display.py
+1
View File
@@ -53,6 +53,7 @@ _DONATION_STATUS_LABELS = {
_VOID_TYPE_FALLBACK_LABELS = {
"khhc": "客户毁存",
"jcbhg": "检测不合格",
"sysfq": "实验室废弃",
}
# 非法来源状态视为终态，不允许再转出
backend/app/crud/collection_protocol.py
+14
View File
@@ -2642,6 +2642,7 @@ def update_collection_protocol(
source_presign_order_id: Optional[str] = None,
remark: Optional[str] = None,
commit: bool = True,
sync_reception_discard: bool = True,
) -> BizCollectionProtocol:
obj = get_collection_protocol(
db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id
@@ -2650,6 +2651,7 @@ def update_collection_protocol(
raise ValueError("协议单不存在或无权操作")
from_status = (obj.protocol_status or "").strip()
status_log_payload: Optional[Dict[str, Any]] = None
voided_now = False
if protocol_no is not None:
pn = protocol_no.strip()
if pn != (obj.protocol_no or ""):
@@ -2660,6 +2662,8 @@ def update_collection_protocol(
next_status = protocol_status.strip()
assert_protocol_status_transition(obj.protocol_status or "", next_status, kind="protocol")
obj.protocol_status = next_status
if next_status == "yzf" and from_status != "yzf":
voided_now = True
if next_status == "yzf" and void_time is not None:
if void_type is None or not str(void_type).strip():
raise ValueError("请选择作废类型")
@@ -2749,6 +2753,16 @@ def update_collection_protocol(
tenant_no=tenant_no,
user_id=user_id,
)
if voided_now and sync_reception_discard:
from app.services.protocol_reception_discard_sync import after_protocol_voided
after_protocol_voided(
db,
source="cell_storage",
protocol_id=obj.id,
tenant_no=tenant_no,
user_id=user_id,
)
if commit:
db.commit()
db.refresh(obj)
backend/app/crud/research_donation.py
+10
View File
@@ -1686,6 +1686,16 @@ def change_status(
)
db.add(obj)
if ns == "yzf" and (current or "").strip() != "yzf":
from app.services.protocol_reception_discard_sync import after_protocol_voided
after_protocol_voided(
db,
source="research_donation",
protocol_id=obj.id,
tenant_no=tenant_no,
user_id=user_id,
)
db.commit()
db.refresh(obj)
return obj
backend/app/crud/sample_reception.py
+12
View File
@@ -1033,6 +1033,12 @@ def create_sample_reception(
user_id=user_id,
create_dept_id=create_dept_id or "",
)
if conclusion == "bj":
from app.services.protocol_reception_discard_sync import after_sample_reception_saved
after_sample_reception_saved(
db, reception=obj, tenant_no=tenant_no, user_id=user_id
)
db.commit()
db.refresh(obj)
return obj
@@ -1151,6 +1157,12 @@ def update_sample_reception(
)
check_result_svc.sync_legacy_columns_from_results(db, rec, saved_rows)
db.add(rec)
if conclusion == "bj":
from app.services.protocol_reception_discard_sync import after_sample_reception_saved
after_sample_reception_saved(
db, reception=rec, tenant_no=tenant_no, user_id=user_id
)
db.commit()
db.refresh(rec)
return rec
backend/app/services/protocol_reception_discard_sync.py
+421
View File
@@ -0,0 +1,421 @@
"""市场协议作废 ↔ 质检样本接收废弃双向联动。
规则（细胞存储 / 科研捐赠）：
1. 协议变为已作废（yzf）时，协议下全部样本接收结论自动为废弃（bj）。
2. 协议下全部有效样本均已废弃/拒收（bj）时，协议自动已作废，作废类型为「实验室废弃」。
"""
from __future__ import annotations
from datetime import datetime
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.models.models import (
BizCollectionProtocol,
BizProtocolSample,
BizSampleReception,
SysDictData,
)
from app.models.research_donation import BizResearchDonation, BizResearchDonationSample
LAB_VOID_TYPE_CODE = "sysfq"
LAB_VOID_TYPE_LABEL = "实验室废弃"
VOID_TYPE_DICT = "xyzflx"
_AUTO_DISCARD_REMARK = "协议作废自动废弃"
_AUTO_VOID_DESC = "全部样本接收废弃/拒收，系统自动作废"
def ensure_lab_void_type(db: Session, *, tenant_no: str) -> str:
"""确保租户字典 xyzflx 存在「实验室废弃」，返回 dict_value。"""
tenant = (tenant_no or "").strip()
if not tenant:
raise ValueError("租户编号不能为空")
by_value = (
db.query(SysDictData)
.filter(
SysDictData.dict_type == VOID_TYPE_DICT,
SysDictData.dict_value == LAB_VOID_TYPE_CODE,
SysDictData.tenant_no == tenant,
SysDictData.del_flag == 0,
)
.first()
)
if by_value:
return LAB_VOID_TYPE_CODE
by_label = (
db.query(SysDictData)
.filter(
SysDictData.dict_type == VOID_TYPE_DICT,
SysDictData.dict_label == LAB_VOID_TYPE_LABEL,
SysDictData.tenant_no == tenant,
SysDictData.del_flag == 0,
)
.first()
)
if by_label and (by_label.dict_value or "").strip():
return (by_label.dict_value or "").strip()
now = datetime.now()
db.add(
SysDictData(
id=generate_uuid(),
dict_label=LAB_VOID_TYPE_LABEL,
dict_value=LAB_VOID_TYPE_CODE,
dict_type=VOID_TYPE_DICT,
tenant_no=tenant,
order_num=90,
remark="issue#31 实验室废弃自动作废类型",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
db.flush()
return LAB_VOID_TYPE_CODE
def _list_cell_samples(db: Session, *, protocol_id: str, tenant_no: str) -> List[BizProtocolSample]:
return (
db.query(BizProtocolSample)
.filter(
BizProtocolSample.collection_protocol_id == protocol_id,
BizProtocolSample.tenant_no == tenant_no,
BizProtocolSample.del_flag == 0,
)
.order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc())
.all()
)
def _list_donation_samples(
db: Session, *, donation_id: str, tenant_no: str
) -> List[BizResearchDonationSample]:
return (
db.query(BizResearchDonationSample)
.filter(
BizResearchDonationSample.research_donation_id == donation_id,
BizResearchDonationSample.tenant_no == tenant_no,
BizResearchDonationSample.del_flag == 0,
)
.order_by(
BizResearchDonationSample.order_num.asc(),
BizResearchDonationSample.create_time.asc(),
)
.all()
)
def _find_reception(
db: Session,
*,
tenant_no: str,
source: str,
sample_code: str,
) -> Optional[BizSampleReception]:
code = (sample_code or "").strip()
if not code:
return None
return (
db.query(BizSampleReception)
.filter(
BizSampleReception.tenant_no == tenant_no,
BizSampleReception.del_flag == 0,
BizSampleReception.reception_source == source,
BizSampleReception.sample_code == code,
)
.first()
)
def _upsert_discard_reception(
db: Session,
*,
source: str,
tenant_no: str,
user_id: Optional[str],
sample_code: str,
protocol_sample_id: str = "",
collection_protocol_id: str = "",
donation_sample_id: Optional[str] = None,
research_donation_id: Optional[str] = None,
now: Optional[datetime] = None,
) -> None:
code = (sample_code or "").strip()
if not code:
return
ts = now or datetime.now()
rec = _find_reception(db, tenant_no=tenant_no, source=source, sample_code=code)
if rec:
if (rec.accept_conclusion or "").strip() == "bj":
return
rec.accept_conclusion = "bj"
rec.update_by = user_id
rec.update_time = ts
if not (rec.remark or "").strip():
rec.remark = _AUTO_DISCARD_REMARK
db.add(rec)
return
obj = BizSampleReception(
id=generate_uuid(),
protocol_sample_id=protocol_sample_id or "",
collection_protocol_id=collection_protocol_id or "",
reception_time=ts,
handler_user_id=user_id,
handler_name=None,
receive_temperature=None,
accept_conclusion="bj",
receive_attachments=None,
reception_source=source,
donation_sample_id=donation_sample_id,
research_donation_id=research_donation_id,
sample_code=code,
lab_storage_code=None,
tenant_no=tenant_no,
remark=_AUTO_DISCARD_REMARK,
create_by=user_id,
create_dept_id="",
update_by=user_id,
create_time=ts,
update_time=ts,
del_flag=0,
check_packaging=0,
check_sealing=0,
check_no_damage=0,
check_no_leakage=0,
check_color=0,
check_tying_rope=0,
check_no_stratification=0,
)
db.add(obj)
def discard_receptions_for_voided_protocol(
db: Session,
*,
source: str,
protocol_id: str,
tenant_no: str,
user_id: Optional[str] = None,
) -> int:
"""协议已作废后，将其下样本接收全部置为废弃。返回处理样本数。"""
src = (source or "").strip()
pid = (protocol_id or "").strip()
tenant = (tenant_no or "").strip()
if src not in ("cell_storage", "research_donation") or not pid or not tenant:
return 0
now = datetime.now()
if src == "cell_storage":
samples = _list_cell_samples(db, protocol_id=pid, tenant_no=tenant)
for sample in samples:
_upsert_discard_reception(
db,
source=src,
tenant_no=tenant,
user_id=user_id,
sample_code=sample.sample_code or "",
protocol_sample_id=sample.id,
collection_protocol_id=pid,
now=now,
)
else:
samples = _list_donation_samples(db, donation_id=pid, tenant_no=tenant)
for sample in samples:
_upsert_discard_reception(
db,
source=src,
tenant_no=tenant,
user_id=user_id,
sample_code=sample.sample_code or "",
donation_sample_id=sample.id,
research_donation_id=pid,
now=now,
)
db.flush()
return len(samples)
def _all_samples_discarded(
db: Session,
*,
source: str,
protocol_id: str,
tenant_no: str,
) -> bool:
src = (source or "").strip()
if src == "cell_storage":
samples = _list_cell_samples(db, protocol_id=protocol_id, tenant_no=tenant_no)
else:
samples = _list_donation_samples(db, donation_id=protocol_id, tenant_no=tenant_no)
if not samples:
return False
for sample in samples:
code = (sample.sample_code or "").strip()
if not code:
return False
rec = _find_reception(db, tenant_no=tenant_no, source=src, sample_code=code)
if not rec or (rec.accept_conclusion or "").strip() != "bj":
return False
return True
def maybe_void_protocol_when_all_samples_discarded(
db: Session,
*,
source: str,
protocol_id: str,
tenant_no: str,
user_id: Optional[str] = None,
) -> bool:
"""若协议下样本全部废弃，则自动作废协议（作废类型=实验室废弃）。返回是否执行了作废。"""
src = (source or "").strip()
pid = (protocol_id or "").strip()
tenant = (tenant_no or "").strip()
if src not in ("cell_storage", "research_donation") or not pid or not tenant:
return False
if not _all_samples_discarded(db, source=src, protocol_id=pid, tenant_no=tenant):
return False
void_type = ensure_lab_void_type(db, tenant_no=tenant)
now = datetime.now()
if src == "cell_storage":
obj = (
db.query(BizCollectionProtocol)
.filter(
BizCollectionProtocol.id == pid,
BizCollectionProtocol.tenant_no == tenant,
BizCollectionProtocol.del_flag == 0,
)
.first()
)
if not obj:
return False
status = (obj.protocol_status or "").strip()
if status in ("yzf", "sdfq"):
return False
from app.core.protocol_status_display import assert_protocol_status_transition
from app.crud.collection_protocol import update_collection_protocol
try:
assert_protocol_status_transition(status, "yzf", kind="protocol")
except ValueError:
return False
update_collection_protocol(
db,
pid,
tenant_no=tenant,
user_id=user_id,
protocol_status="yzf",
void_time=now,
void_desc=_AUTO_VOID_DESC,
void_type=void_type,
commit=False,
sync_reception_discard=False,
)
return True
obj = (
db.query(BizResearchDonation)
.filter(
BizResearchDonation.id == pid,
BizResearchDonation.tenant_no == tenant,
BizResearchDonation.del_flag == 0,
)
.first()
)
if not obj:
return False
status = (obj.donation_status or "").strip()
if status in ("yzf", "sdfq"):
return False
# 捐赠 change_status 默认 commit；此处内联等价逻辑避免提前提交。
from app.core.protocol_status_display import assert_protocol_status_transition
from app.crud.dict_data import normalize_void_type
from app.crud.research_donation import (
_DONATION_STATUS_CHANGE_TYPE,
append_donation_status_change,
)
current = obj.donation_status or ""
try:
assert_protocol_status_transition(current, "yzf", kind="donation")
except ValueError:
return False
normalized = normalize_void_type(db, value=void_type, tenant_no=tenant)
obj.donation_status = "yzf"
obj.void_time = now
obj.void_desc = _AUTO_VOID_DESC
obj.void_type = normalized
obj.update_by = user_id
obj.update_time = now
change_type = _DONATION_STATUS_CHANGE_TYPE.get("yzf")
if change_type:
append_donation_status_change(
db,
research_donation_id=obj.id,
change_type=change_type,
from_status=current,
to_status="yzf",
change_time=now,
change_desc=_AUTO_VOID_DESC,
void_type=normalized,
tenant_no=tenant,
user_id=user_id,
)
db.add(obj)
db.flush()
return True
def after_sample_reception_saved(
db: Session,
*,
reception: BizSampleReception,
tenant_no: str,
user_id: Optional[str] = None,
) -> None:
"""样本接收保存后：若结论为废弃，尝试自动作废所属协议。"""
if (reception.accept_conclusion or "").strip() != "bj":
return
source = (reception.reception_source or "cell_storage").strip() or "cell_storage"
if source == "cell_storage":
protocol_id = (reception.collection_protocol_id or "").strip()
else:
protocol_id = (reception.research_donation_id or "").strip()
if not protocol_id:
return
maybe_void_protocol_when_all_samples_discarded(
db,
source=source,
protocol_id=protocol_id,
tenant_no=tenant_no,
user_id=user_id,
)
def after_protocol_voided(
db: Session,
*,
source: str,
protocol_id: str,
tenant_no: str,
user_id: Optional[str] = None,
) -> None:
"""协议作废后：同步废弃其下全部样本接收。"""
discard_receptions_for_voided_protocol(
db,
source=source,
protocol_id=protocol_id,
tenant_no=tenant_no,
user_id=user_id,
)
backend/tests/test_protocol_reception_discard_sync.py
+328
View File
@@ -0,0 +1,328 @@
"""协议作废 ↔ 样本接收废弃双向联动。"""
from __future__ import annotations
import unittest
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.database import Base
from app.models.models import (
BizCollectionProtocol,
BizProtocolSample,
BizProtocolStatusChange,
BizSampleReception,
SysDictData,
)
from app.models.research_donation import (
BizDonationStatusChange,
BizResearchDonation,
BizResearchDonationSample,
)
from app.services.protocol_reception_discard_sync import (
LAB_VOID_TYPE_CODE,
after_protocol_voided,
after_sample_reception_saved,
ensure_lab_void_type,
)
def _session():
engine = create_engine("sqlite://")
Base.metadata.create_all(
bind=engine,
tables=[
SysDictData.__table__,
BizCollectionProtocol.__table__,
BizProtocolSample.__table__,
BizProtocolStatusChange.__table__,
BizSampleReception.__table__,
BizResearchDonation.__table__,
BizResearchDonationSample.__table__,
BizDonationStatusChange.__table__,
],
)
return sessionmaker(bind=engine)()
def _now():
return datetime(2026, 11, 5, 12, 0, 0)
class TestProtocolReceptionDiscardSync(unittest.TestCase):
def test_ensure_lab_void_type_creates_dict_row(self):
db = _session()
code = ensure_lab_void_type(db, tenant_no="tenant-a")
self.assertEqual(code, LAB_VOID_TYPE_CODE)
row = (
db.query(SysDictData)
.filter(
SysDictData.tenant_no == "tenant-a",
SysDictData.dict_type == "xyzflx",
SysDictData.dict_value == LAB_VOID_TYPE_CODE,
)
.one()
)
self.assertEqual(row.dict_label, "实验室废弃")
# idempotent
self.assertEqual(ensure_lab_void_type(db, tenant_no="tenant-a"), LAB_VOID_TYPE_CODE)
def test_protocol_void_marks_all_sample_receptions_discarded(self):
db = _session()
now = _now()
db.add(
BizCollectionProtocol(
id="p1",
protocol_no="CP-1",
protocol_status="sxz",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
db.add(
BizProtocolSample(
id="s1",
collection_protocol_id="p1",
sample_code="ZK001",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
db.add(
BizProtocolSample(
id="s2",
collection_protocol_id="p1",
sample_code="ZK002",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
db.add(
BizSampleReception(
id="r1",
protocol_sample_id="s1",
collection_protocol_id="p1",
reception_time=now,
accept_conclusion="js",
reception_source="cell_storage",
sample_code="ZK001",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
check_packaging=0,
check_sealing=0,
check_no_damage=0,
check_no_leakage=0,
check_color=0,
check_tying_rope=0,
check_no_stratification=0,
)
)
db.commit()
after_protocol_voided(
db,
source="cell_storage",
protocol_id="p1",
tenant_no="tenant-a",
user_id="u1",
)
db.commit()
rows = (
db.query(BizSampleReception)
.filter(BizSampleReception.tenant_no == "tenant-a", BizSampleReception.del_flag == 0)
.all()
)
self.assertEqual(len(rows), 2)
by_code = {r.sample_code: r.accept_conclusion for r in rows}
self.assertEqual(by_code["ZK001"], "bj")
self.assertEqual(by_code["ZK002"], "bj")
def test_all_samples_discarded_auto_voids_protocol_with_lab_type(self):
db = _session()
now = _now()
db.add(
BizCollectionProtocol(
id="p2",
protocol_no="CP-2",
protocol_status="sxz",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
db.add(
BizProtocolSample(
id="s3",
collection_protocol_id="p2",
sample_code="ZK003",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
rec = BizSampleReception(
id="r2",
protocol_sample_id="s3",
collection_protocol_id="p2",
reception_time=now,
accept_conclusion="bj",
reception_source="cell_storage",
sample_code="ZK003",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
check_packaging=0,
check_sealing=0,
check_no_damage=0,
check_no_leakage=0,
check_color=0,
check_tying_rope=0,
check_no_stratification=0,
)
db.add(rec)
db.commit()
after_sample_reception_saved(
db, reception=rec, tenant_no="tenant-a", user_id="u1"
)
db.commit()
proto = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == "p2").one()
self.assertEqual(proto.protocol_status, "yzf")
self.assertEqual(proto.void_type, LAB_VOID_TYPE_CODE)
self.assertIn("全部样本接收", proto.void_desc or "")
def test_partial_discard_does_not_void_protocol(self):
db = _session()
now = _now()
db.add(
BizCollectionProtocol(
id="p3",
protocol_no="CP-3",
protocol_status="sxz",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
for sid, code in (("s4", "ZK004"), ("s5", "ZK005")):
db.add(
BizProtocolSample(
id=sid,
collection_protocol_id="p3",
sample_code=code,
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
rec = BizSampleReception(
id="r3",
protocol_sample_id="s4",
collection_protocol_id="p3",
reception_time=now,
accept_conclusion="bj",
reception_source="cell_storage",
sample_code="ZK004",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
check_packaging=0,
check_sealing=0,
check_no_damage=0,
check_no_leakage=0,
check_color=0,
check_tying_rope=0,
check_no_stratification=0,
)
db.add(rec)
db.commit()
after_sample_reception_saved(
db, reception=rec, tenant_no="tenant-a", user_id="u1"
)
db.commit()
proto = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == "p3").one()
self.assertEqual(proto.protocol_status, "sxz")
self.assertFalse(proto.void_type)
def test_donation_void_creates_discard_receptions(self):
db = _session()
now = _now()
db.add(
BizResearchDonation(
id="d1",
donation_no="RD-1",
donation_status="sxz",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
db.add(
BizResearchDonationSample(
id="ds1",
research_donation_id="d1",
sample_code="RD001",
tenant_no="tenant-a",
create_dept_id="",
create_time=now,
update_time=now,
del_flag=0,
)
)
db.commit()
after_protocol_voided(
db,
source="research_donation",
protocol_id="d1",
tenant_no="tenant-a",
user_id="u1",
)
db.commit()
rec = (
db.query(BizSampleReception)
.filter(
BizSampleReception.sample_code == "RD001",
BizSampleReception.reception_source == "research_donation",
)
.one()
)
self.assertEqual(rec.accept_conclusion, "bj")
self.assertEqual(rec.research_donation_id, "d1")
if __name__ == "__main__":
unittest.main()
docs/sql/migrations/2026-11-05-void-type-lab-discard.sql
+32
View File
@@ -0,0 +1,32 @@
-- Issue #31：作废类型字典补充「实验室废弃」(sysfq)
-- 为已有 xyzflx 字典的租户补齐；可重复执行。
INSERT INTO sys_dict_data (
id, dict_label, dict_value, dict_type, tenant_no, order_num, remark,
create_dept_id, create_time, update_time, del_flag
)
SELECT
MD5(CONCAT(t.tenant_no, ':xyzflx:sysfq')),
'实验室废弃',
'sysfq',
'xyzflx',
t.tenant_no,
90,
'issue#31 实验室废弃',
'',
NOW(),
NOW(),
0
FROM (
SELECT DISTINCT tenant_no
FROM sys_dict_data
WHERE dict_type = 'xyzflx' AND del_flag = 0
) t
WHERE NOT EXISTS (
SELECT 1
FROM sys_dict_data existing
WHERE existing.tenant_no = t.tenant_no
AND existing.dict_type = 'xyzflx'
AND existing.dict_value = 'sysfq'
AND existing.del_flag = 0
);
web-admin/src/utils/void-type-dict.ts
+2 -1
View File
@@ -6,7 +6,8 @@ export const VOID_TYPE_DICT = 'xyzflx'
const FALLBACK_LABELS: Record<string, string> = {
khhc: '客户毁存',
jcbhg: '检测不合格'
jcbhg: '检测不合格',
sysfq: '实验室废弃'
}
function buildFallbackOptions(): Api.Dict.DictDataOption[] {