diff --git a/backend/app/api/protocol_sample_collection.py b/backend/app/api/protocol_sample_collection.py index bd132147..c470d85a 100644 --- a/backend/app/api/protocol_sample_collection.py +++ b/backend/app/api/protocol_sample_collection.py @@ -126,7 +126,6 @@ async def quick_register_sample_info( items = [ {"sampleCode": it.sample_code, "protocolId": it.protocol_id} for it in (body.items or []) ] - protocol_ids = [it["protocolId"] for it in items] kind = (body.kind or "").strip() source = (body.reception_source or "").strip() attachments = body.attachments or "" @@ -149,7 +148,7 @@ async def quick_register_sample_info( return error_response(400, "储存证仅支持细胞存储样本") result = quick_svc.batch_handle_storage_certs( db, - protocol_ids=protocol_ids, + items=items, tenant_no=tenant_no, user=user, user_id=user_id, @@ -163,7 +162,7 @@ async def quick_register_sample_info( return error_response(400, "保险仅支持细胞存储样本") result = quick_svc.batch_handle_insurance( db, - protocol_ids=protocol_ids, + items=items, tenant_no=tenant_no, user=user, user_id=user_id, diff --git a/backend/app/crud/sample_info_quick_register.py b/backend/app/crud/sample_info_quick_register.py index 394aaf33..69103995 100644 --- a/backend/app/crud/sample_info_quick_register.py +++ b/backend/app/crud/sample_info_quick_register.py @@ -8,7 +8,8 @@ from sqlalchemy.orm import Session from app.core.data_scope import apply_data_scope from app.core.utils import generate_uuid -from app.crud import collection_protocol as cp_svc +from app.crud import sample_insurance as sample_insurance_crud +from app.crud import sample_storage_cert as sample_storage_cert_crud from app.crud.storage_release_review import _get_user_name from app.models.models import ( BizCollectionProtocol, @@ -18,7 +19,7 @@ from app.models.models import ( from app.models.research_donation import BizResearchDonation, BizResearchDonationSample from app.services.sample_info_quick_register import ( apply_quick_storage_payload, - unique_nonempty_ids, + unique_sample_items, ) @@ -66,7 +67,7 @@ def _pick_latest_review( def batch_handle_storage_certs( db: Session, *, - protocol_ids: List[str], + items: List[Dict[str, str]], tenant_no: str, user: dict, user_id: str, @@ -74,37 +75,43 @@ def batch_handle_storage_certs( handler_user_id: Optional[str], attachments: str, remark: str, + mail_cert_time: Optional[datetime] = None, + update_mail_cert_time: bool = False, ) -> Dict[str, int]: - ids = unique_nonempty_ids(protocol_ids) - if not ids: + pairs = unique_sample_items(items) + if not pairs: raise ValueError("请选择样本") success = 0 fail = 0 - for pid in ids: - if not _protocol_accessible(db, pid, tenant_no, user): + for sample_code, protocol_id in pairs: + if not protocol_id or not _protocol_accessible(db, protocol_id, tenant_no, user): fail += 1 continue try: - cp_svc.upsert_storage_cert_handle( + sample_storage_cert_crud.upsert_handle( db, - pid, - tenant_no, - user_id, + tenant_no=tenant_no, + sample_code=sample_code, + collection_protocol_id=protocol_id, + user_id=user_id, handle_time=handle_time, handler_user_id=handler_user_id, cert_image_urls=attachments, remark=remark, + mail_cert_time=mail_cert_time, + update_mail_cert_time=update_mail_cert_time, ) success += 1 except ValueError: fail += 1 + db.commit() return {"success": success, "fail": fail} def batch_handle_insurance( db: Session, *, - protocol_ids: List[str], + items: List[Dict[str, str]], tenant_no: str, user: dict, user_id: str, @@ -112,30 +119,36 @@ def batch_handle_insurance( handler_user_id: Optional[str], attachments: str, remark: str, + policy_send_time: Optional[datetime] = None, + update_policy_send_time: bool = False, ) -> Dict[str, int]: - ids = unique_nonempty_ids(protocol_ids) - if not ids: + pairs = unique_sample_items(items) + if not pairs: raise ValueError("请选择样本") success = 0 fail = 0 - for pid in ids: - if not _protocol_accessible(db, pid, tenant_no, user): + for sample_code, protocol_id in pairs: + if not protocol_id or not _protocol_accessible(db, protocol_id, tenant_no, user): fail += 1 continue try: - cp_svc.upsert_fullmoon_todo_handle( + sample_insurance_crud.upsert_handle( db, - pid, - tenant_no, - user_id, + tenant_no=tenant_no, + sample_code=sample_code, + collection_protocol_id=protocol_id, + user_id=user_id, handle_time=handle_time, handler_user_id=handler_user_id, insurance_attachments=attachments, handle_remark=remark, + policy_send_time=policy_send_time, + update_policy_send_time=update_policy_send_time, ) success += 1 except ValueError: fail += 1 + db.commit() return {"success": success, "fail": fail} @@ -161,15 +174,7 @@ def batch_quick_register_storage( attachments=attachments, remark=remark, ) - seen = set() - pairs: List[tuple] = [] - for it in items or []: - code = (it.get("sampleCode") or "").strip() - pid = (it.get("protocolId") or "").strip() - if not code or code in seen: - continue - seen.add(code) - pairs.append((code, pid)) + pairs = unique_sample_items(items) if not pairs: raise ValueError("请选择样本") diff --git a/backend/app/crud/sample_insurance.py b/backend/app/crud/sample_insurance.py new file mode 100644 index 00000000..b6bb428f --- /dev/null +++ b/backend/app/crud/sample_insurance.py @@ -0,0 +1,134 @@ +"""样本保险办理 CRUD。""" +from __future__ import annotations + +from datetime import datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.models.models import BizSampleInsurance +from app.services.sample_info_progress import apply_handle_payload, insurance_handled + + +def _fmt_dt(t: Optional[datetime]) -> str: + return t.strftime("%Y-%m-%d %H:%M:%S") if t else "" + + +def get_by_sample_code( + db: Session, tenant_no: str, sample_code: str +) -> Optional[BizSampleInsurance]: + code = (sample_code or "").strip() + if not code: + return None + return ( + db.query(BizSampleInsurance) + .filter( + BizSampleInsurance.tenant_no == tenant_no, + BizSampleInsurance.sample_code == code, + BizSampleInsurance.del_flag == 0, + ) + .first() + ) + + +def get_map_by_sample_codes( + db: Session, tenant_no: str, codes: List[str] +) -> Dict[str, BizSampleInsurance]: + cleaned = list({(c or "").strip() for c in codes if (c or "").strip()}) + if not cleaned: + return {} + rows = ( + db.query(BizSampleInsurance) + .filter( + BizSampleInsurance.tenant_no == tenant_no, + BizSampleInsurance.sample_code.in_(cleaned), + BizSampleInsurance.del_flag == 0, + ) + .all() + ) + return {row.sample_code: row for row in rows} + + +def upsert_handle( + db: Session, + *, + tenant_no: str, + sample_code: str, + collection_protocol_id: Optional[str], + user_id: Optional[str], + handle_time: Optional[datetime], + handler_user_id: Optional[str], + insurance_attachments: str = "", + handle_remark: str = "", + policy_send_time: Optional[datetime] = None, + update_policy_send_time: bool = False, +) -> BizSampleInsurance: + code = (sample_code or "").strip() + if not code: + raise ValueError("sample_code is required") + + payload = apply_handle_payload( + handle_time=handle_time, + handler_user_id=handler_user_id, + attachments=insurance_attachments or "", + remark=handle_remark or "", + ) + now = datetime.now() + row = get_by_sample_code(db, tenant_no, code) + if row is None: + row = BizSampleInsurance( + id=generate_uuid(), + sample_code=code, + collection_protocol_id=(collection_protocol_id or "").strip() or None, + tenant_no=tenant_no, + order_num=0, + todo_status="0", + create_by=user_id, + create_dept_id="", + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + db.add(row) + else: + if collection_protocol_id and not row.collection_protocol_id: + row.collection_protocol_id = (collection_protocol_id or "").strip() or None + row.update_by = user_id + row.update_time = now + + row.handle_time = payload["handle_time"] + row.handler_user_id = payload["handler_user_id"] + row.todo_status = payload["make_or_todo_status"] + row.handle_date = payload["legacy_date"] or None + row.insurance_attachments = payload["attachments"] or None + row.handle_remark = payload["remark"] or None + if update_policy_send_time: + row.policy_send_time = policy_send_time + + db.flush() + return row + + +def row_to_dict(row: BizSampleInsurance) -> Dict[str, Any]: + return { + "id": row.id, + "sampleCode": row.sample_code or "", + "collectionProtocolId": row.collection_protocol_id or "", + "todoStatus": row.todo_status or "0", + "handleDate": (row.handle_date or "").strip() if row.handle_date else "", + "handleTime": _fmt_dt(row.handle_time) if getattr(row, "handle_time", None) else "", + "handlerUserId": getattr(row, "handler_user_id", None) or "", + "handleRemark": row.handle_remark or "", + "insuranceAttachments": row.insurance_attachments or "", + "policySendTime": _fmt_dt(row.policy_send_time) + if getattr(row, "policy_send_time", None) + else "", + "insuranceHandled": insurance_handled(row), + "tenantNo": row.tenant_no, + "orderNum": row.order_num if row.order_num is not None else 0, + "remark": row.remark or "", + "createTime": _fmt_dt(row.create_time), + "updateTime": _fmt_dt(row.update_time), + } diff --git a/backend/app/crud/sample_storage_cert.py b/backend/app/crud/sample_storage_cert.py new file mode 100644 index 00000000..f3073991 --- /dev/null +++ b/backend/app/crud/sample_storage_cert.py @@ -0,0 +1,130 @@ +"""样本储存证办理 CRUD。""" +from __future__ import annotations + +from datetime import datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.models.models import BizSampleStorageCert +from app.services.sample_info_progress import apply_handle_payload, storage_cert_handled + + +def _fmt_dt(t: Optional[datetime]) -> str: + return t.strftime("%Y-%m-%d %H:%M:%S") if t else "" + + +def get_by_sample_code( + db: Session, tenant_no: str, sample_code: str +) -> Optional[BizSampleStorageCert]: + code = (sample_code or "").strip() + if not code: + return None + return ( + db.query(BizSampleStorageCert) + .filter( + BizSampleStorageCert.tenant_no == tenant_no, + BizSampleStorageCert.sample_code == code, + BizSampleStorageCert.del_flag == 0, + ) + .first() + ) + + +def get_map_by_sample_codes( + db: Session, tenant_no: str, codes: List[str] +) -> Dict[str, BizSampleStorageCert]: + cleaned = list({(c or "").strip() for c in codes if (c or "").strip()}) + if not cleaned: + return {} + rows = ( + db.query(BizSampleStorageCert) + .filter( + BizSampleStorageCert.tenant_no == tenant_no, + BizSampleStorageCert.sample_code.in_(cleaned), + BizSampleStorageCert.del_flag == 0, + ) + .all() + ) + return {row.sample_code: row for row in rows} + + +def upsert_handle( + db: Session, + *, + tenant_no: str, + sample_code: str, + collection_protocol_id: Optional[str], + user_id: Optional[str], + handle_time: Optional[datetime], + handler_user_id: Optional[str], + cert_image_urls: str = "", + remark: str = "", + mail_cert_time: Optional[datetime] = None, + update_mail_cert_time: bool = False, +) -> BizSampleStorageCert: + code = (sample_code or "").strip() + if not code: + raise ValueError("sample_code is required") + + payload = apply_handle_payload( + handle_time=handle_time, + handler_user_id=handler_user_id, + attachments=cert_image_urls or "", + remark=remark or "", + ) + now = datetime.now() + row = get_by_sample_code(db, tenant_no, code) + if row is None: + row = BizSampleStorageCert( + id=generate_uuid(), + sample_code=code, + collection_protocol_id=(collection_protocol_id or "").strip() or None, + tenant_no=tenant_no, + order_num=0, + create_by=user_id, + create_dept_id="", + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + db.add(row) + else: + if collection_protocol_id and not row.collection_protocol_id: + row.collection_protocol_id = (collection_protocol_id or "").strip() or None + row.update_by = user_id + row.update_time = now + + row.handle_time = payload["handle_time"] + row.handler_user_id = payload["handler_user_id"] + row.make_status = payload["make_or_todo_status"] + row.make_date = payload["legacy_date"] or None + row.cert_image_urls = payload["attachments"] or None + row.remark = payload["remark"] or None + if update_mail_cert_time: + row.mail_cert_time = mail_cert_time + + db.flush() + return row + + +def row_to_dict(row: BizSampleStorageCert) -> Dict[str, Any]: + return { + "id": row.id, + "sampleCode": row.sample_code or "", + "collectionProtocolId": row.collection_protocol_id or "", + "makeStatus": row.make_status or "0", + "certImageUrls": row.cert_image_urls or "", + "makeDate": (row.make_date or "").strip() if row.make_date else "", + "handleTime": _fmt_dt(row.handle_time) if getattr(row, "handle_time", None) else "", + "handlerUserId": getattr(row, "handler_user_id", None) or "", + "mailCertTime": _fmt_dt(row.mail_cert_time) if getattr(row, "mail_cert_time", None) else "", + "storageCertHandled": storage_cert_handled(row), + "tenantNo": row.tenant_no, + "orderNum": row.order_num if row.order_num is not None else 0, + "remark": row.remark or "", + "createTime": _fmt_dt(row.create_time), + "updateTime": _fmt_dt(row.update_time), + } diff --git a/backend/app/services/sample_info_quick_register.py b/backend/app/services/sample_info_quick_register.py index 6ef8e0e7..38e7b408 100644 --- a/backend/app/services/sample_info_quick_register.py +++ b/backend/app/services/sample_info_quick_register.py @@ -19,6 +19,22 @@ def unique_nonempty_ids(ids: Iterable[Optional[str]]) -> List[str]: return out +def unique_sample_items( + items: Optional[Iterable[Dict[str, str]]], +) -> List[tuple]: + """按 sampleCode 去重，保留首次出现的 (sampleCode, protocolId)。""" + seen = set() + out: List[tuple] = [] + for it in items or []: + code = (it.get("sampleCode") or "").strip() + pid = (it.get("protocolId") or "").strip() + if not code or code in seen: + continue + seen.add(code) + out.append((code, pid)) + return out + + def apply_quick_storage_payload( *, handle_time: Optional[datetime], diff --git a/backend/tests/test_sample_cert_insurance_sample_level.py b/backend/tests/test_sample_cert_insurance_sample_level.py new file mode 100644 index 00000000..8654925d --- /dev/null +++ b/backend/tests/test_sample_cert_insurance_sample_level.py @@ -0,0 +1,90 @@ +"""样本级储存证 / 保险：handled 判定与 quick-register items 去重。""" +from types import SimpleNamespace +from datetime import datetime + +from app.services.sample_info_progress import insurance_handled, storage_cert_handled +from app.services.sample_info_quick_register import unique_sample_items + + +def test_storage_cert_handled_true_when_handle_time_set(): + assert ( + storage_cert_handled( + SimpleNamespace(handle_time=object(), make_status="0", make_date=None) + ) + is True + ) + assert storage_cert_handled(None) is False + + +def test_storage_cert_handled_legacy_fields(): + assert ( + storage_cert_handled( + SimpleNamespace(handle_time=None, make_status="1", make_date="") + ) + is True + ) + assert ( + storage_cert_handled( + SimpleNamespace(handle_time=None, make_status="0", make_date="20260101") + ) + is True + ) + assert ( + storage_cert_handled( + SimpleNamespace(handle_time=None, make_status="0", make_date="") + ) + is False + ) + + +def test_insurance_handled_true_when_handle_time_set(): + assert ( + insurance_handled( + SimpleNamespace( + handle_time=datetime(2026, 10, 1, 8, 0), + todo_status="0", + handle_date="", + ) + ) + is True + ) + assert insurance_handled(None) is False + + +def test_insurance_handled_legacy_fields(): + assert ( + insurance_handled( + SimpleNamespace(handle_time=None, todo_status="1", handle_date="") + ) + is True + ) + assert ( + insurance_handled( + SimpleNamespace(handle_time=None, todo_status="0", handle_date="20260131") + ) + is True + ) + assert ( + insurance_handled( + SimpleNamespace(handle_time=None, todo_status="0", handle_date="") + ) + is False + ) + + +def test_unique_sample_items_dedupes_by_sample_code(): + pairs = unique_sample_items( + [ + {"sampleCode": "S1", "protocolId": "P1"}, + {"sampleCode": " S1 ", "protocolId": "P9"}, + {"sampleCode": "S2", "protocolId": "P2"}, + {"sampleCode": "", "protocolId": "P3"}, + {"sampleCode": "S2", "protocolId": "P2"}, + ] + ) + assert pairs == [("S1", "P1"), ("S2", "P2")] + + +def test_unique_sample_items_empty(): + assert unique_sample_items([]) == [] + assert unique_sample_items(None) == []