feat(backend): extend sample-info quick-register for sms/followup/fulfillment/all.

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

backend/app/api/protocol_sample_collection.py

@@ -8,11 +8,12 @@ from sqlalchemy.orm import Session
 from app.core.database import get_db
 from app.core.menu_guard import require_any_menu_code
 from app.core.response import error_response, success_response
-from app.core.utils import parse_datetime
+from app.core.utils import parse_date, parse_datetime
 from app.crud import protocol_sample_collection as svc
 from app.crud import sample_info_quick_register as quick_svc
 from app.crud.collection_protocol import total_pages
 from app.models.models import SysUser
+from app.services.sample_customer_service_register import has_any_customer_service_payload
 
 router = APIRouter(prefix="/web/admin", tags=["采集协议样本采集信息"])
 
@@ -109,6 +110,46 @@ class QuickRegisterBody(BaseModel):
     handler_user_id: Optional[str] = Field(None, alias="handlerUserId")
     attachments: Optional[str] = None
     remark: Optional[str] = None
+    mail_cert_time: Optional[str] = Field(None, alias="mailCertTime")
+    policy_send_time: Optional[str] = Field(None, alias="policySendTime")
+    sms_send_date: Optional[str] = Field(None, alias="smsSendDate")
+    sms_send_status: Optional[str] = Field(None, alias="smsSendStatus")
+    microbe_followup_date: Optional[str] = Field(None, alias="microbeFollowupDate")
+    microbe_followup_status: Optional[str] = Field(None, alias="microbeFollowupStatus")
+    fullmoon_followup_date: Optional[str] = Field(None, alias="fullmoonFollowupDate")
+    fullmoon_followup_status: Optional[str] = Field(None, alias="fullmoonFollowupStatus")
+    household_registered: Optional[str] = Field(None, alias="householdRegistered")
+    einvoice_send_time: Optional[str] = Field(None, alias="einvoiceSendTime")
+    effective_address: Optional[str] = Field(None, alias="effectiveAddress")
+
+
+def _optional_dt(raw: Optional[str]):
+    if raw is None:
+        return None, False
+    text = str(raw).strip()
+    if not text:
+        return None, False
+    return parse_datetime(text), True
+
+
+def _customer_service_incoming_from_body(body: QuickRegisterBody) -> dict:
+    return {
+        "sms_send_date": parse_date(body.sms_send_date) if body.sms_send_date else None,
+        "sms_send_status": body.sms_send_status,
+        "microbe_followup_date": parse_date(body.microbe_followup_date)
+        if body.microbe_followup_date
+        else None,
+        "microbe_followup_status": body.microbe_followup_status,
+        "fullmoon_followup_date": parse_date(body.fullmoon_followup_date)
+        if body.fullmoon_followup_date
+        else None,
+        "fullmoon_followup_status": body.fullmoon_followup_status,
+        "household_registered": body.household_registered,
+        "einvoice_send_time": parse_datetime(body.einvoice_send_time)
+        if body.einvoice_send_time
+        else None,
+        "effective_address": body.effective_address,
+    }
 
 
 @router.post("/sample-info/quick-register")
@@ -121,8 +162,6 @@ async def quick_register_sample_info(
         tenant_no = _get_tenant_no(db, user)
         user_id = _get_current_user_id(user)
         ht = parse_datetime(body.handle_time) if body.handle_time else None
-        if ht is None:
-            return error_response(400, "请填写办理时间")
         items = [
             {"sampleCode": it.sample_code, "protocolId": it.protocol_id} for it in (body.items or [])
         ]
@@ -130,7 +169,13 @@ async def quick_register_sample_info(
         source = (body.reception_source or "").strip()
         attachments = body.attachments or ""
         remark = body.remark or ""
+        mail_t, update_mail = _optional_dt(body.mail_cert_time)
+        policy_t, update_policy = _optional_dt(body.policy_send_time)
+        cs_incoming = _customer_service_incoming_from_body(body)
+
         if kind == "storage":
+            if ht is None:
+                return error_response(400, "请填写办理时间")
             result = quick_svc.batch_quick_register_storage(
                 db,
                 items=items,
@@ -146,6 +191,8 @@ async def quick_register_sample_info(
         elif kind == "storage_cert":
             if source != "cell_storage":
                 return error_response(400, "储存证仅支持细胞存储样本")
+            if ht is None:
+                return error_response(400, "请填写办理时间")
             result = quick_svc.batch_handle_storage_certs(
                 db,
                 items=items,
@@ -156,10 +203,14 @@ async def quick_register_sample_info(
                 handler_user_id=body.handler_user_id,
                 attachments=attachments,
                 remark=remark,
+                mail_cert_time=mail_t,
+                update_mail_cert_time=update_mail,
             )
         elif kind == "insurance":
             if source != "cell_storage":
                 return error_response(400, "保险仅支持细胞存储样本")
+            if ht is None:
+                return error_response(400, "请填写办理时间")
             result = quick_svc.batch_handle_insurance(
                 db,
                 items=items,
@@ -170,14 +221,88 @@ async def quick_register_sample_info(
                 handler_user_id=body.handler_user_id,
                 attachments=attachments,
                 remark=remark,
+                policy_send_time=policy_t,
+                update_policy_send_time=update_policy,
             )
+        elif kind in ("sms", "followup", "fulfillment"):
+            if source != "cell_storage":
+                return error_response(400, "客户服务登记仅支持细胞存储样本")
+            result = quick_svc.batch_register_customer_service(
+                db,
+                items=items,
+                tenant_no=tenant_no,
+                user=user,
+                user_id=user_id,
+                incoming=cs_incoming,
+            )
+        elif kind == "all":
+            if source != "cell_storage":
+                return error_response(400, "登记全部仅支持细胞存储样本")
+            wrote_any = False
+            success = 0
+            fail = 0
+            if ht is not None:
+                cert_r = quick_svc.batch_handle_storage_certs(
+                    db,
+                    items=items,
+                    tenant_no=tenant_no,
+                    user=user,
+                    user_id=user_id,
+                    handle_time=ht,
+                    handler_user_id=body.handler_user_id,
+                    attachments=attachments,
+                    remark=remark,
+                    mail_cert_time=mail_t,
+                    update_mail_cert_time=update_mail,
+                    do_commit=False,
+                )
+                ins_r = quick_svc.batch_handle_insurance(
+                    db,
+                    items=items,
+                    tenant_no=tenant_no,
+                    user=user,
+                    user_id=user_id,
+                    handle_time=ht,
+                    handler_user_id=body.handler_user_id,
+                    attachments=attachments,
+                    remark=remark,
+                    policy_send_time=policy_t,
+                    update_policy_send_time=update_policy,
+                    do_commit=False,
+                )
+                success = max(cert_r.get("success", 0), ins_r.get("success", 0))
+                fail = max(cert_r.get("fail", 0), ins_r.get("fail", 0))
+                wrote_any = True
+            elif update_mail or update_policy:
+                # 仅补时间线字段时仍要求办理时间走证险主路径；此处拒绝避免半写
+                return error_response(400, "登记储存证/保险请填写办理时间")
+            if has_any_customer_service_payload(cs_incoming):
+                cs_r = quick_svc.batch_register_customer_service(
+                    db,
+                    items=items,
+                    tenant_no=tenant_no,
+                    user=user,
+                    user_id=user_id,
+                    incoming=cs_incoming,
+                    do_commit=False,
+                )
+                success = max(success, cs_r.get("success", 0))
+                fail = max(fail, cs_r.get("fail", 0))
+                wrote_any = True
+            if not wrote_any:
+                return error_response(400, "请至少填写一项登记信息")
+            db.commit()
+            result = {"success": success, "fail": fail}
         else:
             return error_response(400, "登记类型无效")
         return success_response(result, message="登记完成")
     except PermissionError as e:
+        db.rollback()
         return error_response(403, str(e))
     except ValueError as e:
+        db.rollback()
         return error_response(400, str(e))
     except Exception as e:
+        db.rollback()
         return error_response(400, str(e))
 

backend/app/crud/sample_info_quick_register.py

@@ -8,9 +8,11 @@ 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 sample_customer_service as sample_customer_service_crud
 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.services.sample_customer_service_register import has_any_customer_service_payload
 from app.models.models import (
     BizCollectionProtocol,
     BizProtocolSample,
@@ -77,6 +79,7 @@ def batch_handle_storage_certs(
     remark: str,
     mail_cert_time: Optional[datetime] = None,
     update_mail_cert_time: bool = False,
+    do_commit: bool = True,
 ) -> Dict[str, int]:
     pairs = unique_sample_items(items)
     if not pairs:
@@ -104,6 +107,7 @@ def batch_handle_storage_certs(
             success += 1
         except ValueError:
             fail += 1
+    if do_commit:
         db.commit()
     return {"success": success, "fail": fail}
 
@@ -121,6 +125,7 @@ def batch_handle_insurance(
     remark: str,
     policy_send_time: Optional[datetime] = None,
     update_policy_send_time: bool = False,
+    do_commit: bool = True,
 ) -> Dict[str, int]:
     pairs = unique_sample_items(items)
     if not pairs:
@@ -148,6 +153,45 @@ def batch_handle_insurance(
             success += 1
         except ValueError:
             fail += 1
+    if do_commit:
+        db.commit()
+    return {"success": success, "fail": fail}
+
+
+def batch_register_customer_service(
+    db: Session,
+    *,
+    items: List[Dict[str, str]],
+    tenant_no: str,
+    user: dict,
+    user_id: str,
+    incoming: Dict[str, Any],
+    do_commit: bool = True,
+) -> Dict[str, int]:
+    if not has_any_customer_service_payload(incoming):
+        raise ValueError("请至少填写一项客户服务信息")
+    pairs = unique_sample_items(items)
+    if not pairs:
+        raise ValueError("请选择样本")
+    success = 0
+    fail = 0
+    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:
+            sample_customer_service_crud.upsert_by_sample_code(
+                db,
+                tenant_no=tenant_no,
+                sample_code=sample_code,
+                collection_protocol_id=protocol_id,
+                user_id=user_id,
+                incoming=incoming,
+            )
+            success += 1
+        except ValueError:
+            fail += 1
+    if do_commit:
         db.commit()
     return {"success": success, "fail": fail}
 

backend/app/services/sample_customer_service_register.py

@@ -48,3 +48,14 @@ def merge_customer_service_fields(
             continue
         merged[key] = value
     return merged
+
+
+def has_any_customer_service_payload(incoming: Optional[Dict[str, Any]]) -> bool:
+    """True if any known customer-service field has a non-empty value."""
+    incoming = incoming or {}
+    for key in CUSTOMER_SERVICE_FIELD_KEYS:
+        if key not in incoming:
+            continue
+        if not _is_empty_merge_value(incoming.get(key)):
+            return True
+    return False

backend/tests/test_sample_customer_service_register.py

@@ -2,7 +2,10 @@
 import unittest
 from datetime import date, datetime
 
-from app.services.sample_customer_service_register import merge_customer_service_fields
+from app.services.sample_customer_service_register import (
+    has_any_customer_service_payload,
+    merge_customer_service_fields,
+)
 
 
 class TestMergeCustomerServiceFields(unittest.TestCase):
@@ -52,5 +55,16 @@ class TestMergeCustomerServiceFields(unittest.TestCase):
         self.assertEqual(merged["einvoice_send_time"], dt)
 
 
+class TestHasAnyCustomerServicePayload(unittest.TestCase):
+    def test_empty_false(self):
+        self.assertFalse(has_any_customer_service_payload({}))
+        self.assertFalse(has_any_customer_service_payload({"sms_send_status": ""}))
+        self.assertFalse(has_any_customer_service_payload({"sms_send_status": "  "}))
+
+    def test_nonempty_true(self):
+        self.assertTrue(has_any_customer_service_payload({"sms_send_status": "已发"}))
+        self.assertTrue(has_any_customer_service_payload({"effective_address": "合肥"}))
+
+
 if __name__ == "__main__":
     unittest.main()