feat(backend): write storage cert and insurance at sample level in quick-register.

Co-authored-by: Cursor <cursoragent@cursor.com>

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,

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("请选择样本")
 

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),
+    }

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),
+    }

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],

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) == []