zlwuandCursor
ac8e471aca
Let customer-service quick register write storage, certificate, and insurance separately without extra menu or data-scope checks. List freight by document data scope like the web admin, allow protocol staff outside the collection region, and pad OA datetimes to seconds so PC date pickers can show historical minute values. Co-authored-by: Cursor <cursoragent@cursor.com>
405 lines
16 KiB
Python
405 lines
16 KiB
Python
"""采集协议样本采集信息 API"""
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
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_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=["采集协议样本采集信息"])
|
|
|
|
_SAMPLE_INFO_MENU = require_any_menu_code(
|
|
"cell-storage:sample-info:list",
|
|
)
|
|
|
|
_QUICK_REGISTER_MENU = require_any_menu_code(
|
|
"cell-storage:sample-info:list",
|
|
"research-donation:sample-info:list",
|
|
)
|
|
|
|
|
|
def _get_tenant_no(db: Session, user: dict) -> str:
|
|
user_id = user.get("user_id")
|
|
if not user_id:
|
|
raise PermissionError("用户身份信息缺失，无法确定租户")
|
|
sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
|
|
if not sys_user or not sys_user.tenant_no:
|
|
raise PermissionError("用户租户信息缺失")
|
|
return sys_user.tenant_no
|
|
|
|
|
|
def _page_payload(items: list, total: int, page: int, page_size: int) -> dict:
|
|
return {
|
|
"list": items,
|
|
"total": total,
|
|
"page": page,
|
|
"pageSize": page_size,
|
|
"totalPages": total_pages(total, page_size),
|
|
}
|
|
|
|
|
|
@router.get("/protocol-sample-collections/list")
|
|
async def list_protocol_sample_collections(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100),
|
|
keyword: Optional[str] = None,
|
|
sampleStatus: Optional[str] = Query(None, alias="sampleStatus"),
|
|
collectionFlag: Optional[str] = Query(None, alias="collectionFlag"),
|
|
receptionFlag: Optional[str] = Query(None, alias="receptionFlag"),
|
|
qcFlag: Optional[str] = Query(None, alias="qcFlag"),
|
|
storageFlag: Optional[str] = Query(None, alias="storageFlag"),
|
|
storageCertHandled: Optional[str] = Query(None, alias="storageCertHandled"),
|
|
insuranceHandled: Optional[str] = Query(None, alias="insuranceHandled"),
|
|
db: Session = Depends(get_db),
|
|
user: dict = Depends(_SAMPLE_INFO_MENU),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items, total = svc.list_protocol_sample_collections(
|
|
db,
|
|
page=page,
|
|
page_size=pageSize,
|
|
tenant_no=tenant_no,
|
|
keyword=keyword,
|
|
sample_status=sampleStatus,
|
|
collection_flag=collectionFlag,
|
|
reception_flag=receptionFlag,
|
|
qc_flag=qcFlag,
|
|
storage_flag=storageFlag,
|
|
storage_cert_handled=storageCertHandled,
|
|
insurance_handled=insuranceHandled,
|
|
user_info=user,
|
|
)
|
|
return success_response(_page_payload(items, total, page, pageSize))
|
|
except PermissionError as e:
|
|
return error_response(403, str(e))
|
|
except Exception as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
def _get_current_user_id(user: dict) -> str:
|
|
user_id = user.get("user_id")
|
|
if not user_id:
|
|
raise PermissionError("用户身份信息缺失")
|
|
return str(user_id)
|
|
|
|
|
|
class QuickRegisterItem(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
sample_code: str = Field(..., alias="sampleCode")
|
|
protocol_id: str = Field(..., alias="protocolId")
|
|
|
|
|
|
class QuickRegisterBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
kind: str
|
|
reception_source: str = Field(..., alias="receptionSource")
|
|
items: List[QuickRegisterItem]
|
|
handle_time: Optional[str] = Field(None, alias="handleTime")
|
|
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")
|
|
storage_handle_time: Optional[str] = Field(None, alias="storageHandleTime")
|
|
storage_handler_user_id: Optional[str] = Field(None, alias="storageHandlerUserId")
|
|
storage_attachments: Optional[str] = Field(None, alias="storageAttachments")
|
|
storage_remark: Optional[str] = Field(None, alias="storageRemark")
|
|
cert_handle_time: Optional[str] = Field(None, alias="certHandleTime")
|
|
cert_handler_user_id: Optional[str] = Field(None, alias="certHandlerUserId")
|
|
cert_attachments: Optional[str] = Field(None, alias="certAttachments")
|
|
cert_remark: Optional[str] = Field(None, alias="certRemark")
|
|
insurance_handle_time: Optional[str] = Field(None, alias="insuranceHandleTime")
|
|
insurance_handler_user_id: Optional[str] = Field(None, alias="insuranceHandlerUserId")
|
|
insurance_attachments: Optional[str] = Field(None, alias="insuranceAttachments")
|
|
insurance_remark: Optional[str] = Field(None, alias="insuranceRemark")
|
|
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")
|
|
async def quick_register_sample_info(
|
|
body: QuickRegisterBody,
|
|
db: Session = Depends(get_db),
|
|
user: dict = Depends(_QUICK_REGISTER_MENU),
|
|
):
|
|
kind = (body.kind or "").strip()
|
|
try:
|
|
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
|
|
items = [
|
|
{"sampleCode": it.sample_code, "protocolId": it.protocol_id} for it in (body.items or [])
|
|
]
|
|
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,
|
|
reception_source=source,
|
|
tenant_no=tenant_no,
|
|
user_id=user_id,
|
|
handle_time=ht,
|
|
handler_user_id=body.handler_user_id,
|
|
attachments=attachments,
|
|
remark=remark,
|
|
)
|
|
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,
|
|
tenant_no=tenant_no,
|
|
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,
|
|
)
|
|
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,
|
|
tenant_no=tenant_no,
|
|
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,
|
|
)
|
|
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_id=user_id,
|
|
incoming=cs_incoming,
|
|
)
|
|
elif kind == "all":
|
|
if source != "cell_storage":
|
|
return error_response(400, "登记全部仅支持细胞存储样本")
|
|
split = any(
|
|
value is not None
|
|
for value in (
|
|
body.storage_handle_time,
|
|
body.storage_handler_user_id,
|
|
body.storage_attachments,
|
|
body.storage_remark,
|
|
body.cert_handle_time,
|
|
body.insurance_handle_time,
|
|
body.cert_attachments,
|
|
body.insurance_attachments,
|
|
body.cert_remark,
|
|
body.insurance_remark,
|
|
body.cert_handler_user_id,
|
|
body.insurance_handler_user_id,
|
|
)
|
|
)
|
|
if split:
|
|
storage_ht = (
|
|
parse_datetime(body.storage_handle_time)
|
|
if (body.storage_handle_time or "").strip()
|
|
else None
|
|
)
|
|
storage_handler = body.storage_handler_user_id
|
|
storage_atts = body.storage_attachments or ""
|
|
storage_remark = body.storage_remark or ""
|
|
cert_ht = (
|
|
parse_datetime(body.cert_handle_time)
|
|
if (body.cert_handle_time or "").strip()
|
|
else None
|
|
)
|
|
ins_ht = (
|
|
parse_datetime(body.insurance_handle_time)
|
|
if (body.insurance_handle_time or "").strip()
|
|
else None
|
|
)
|
|
cert_handler = body.cert_handler_user_id
|
|
ins_handler = body.insurance_handler_user_id
|
|
cert_atts = body.cert_attachments or ""
|
|
ins_atts = body.insurance_attachments or ""
|
|
cert_remark = body.cert_remark or ""
|
|
ins_remark = body.insurance_remark or ""
|
|
else:
|
|
storage_ht = None
|
|
storage_handler = None
|
|
storage_atts = ""
|
|
storage_remark = ""
|
|
cert_ht = ins_ht = ht
|
|
cert_handler = ins_handler = body.handler_user_id
|
|
cert_atts = ins_atts = attachments
|
|
cert_remark = ins_remark = remark
|
|
storage_side = bool(
|
|
storage_ht is not None
|
|
or (storage_atts or "").strip()
|
|
or (storage_remark or "").strip()
|
|
)
|
|
cert_side = bool(
|
|
cert_ht is not None
|
|
or update_mail
|
|
or (cert_atts or "").strip()
|
|
or (cert_remark or "").strip()
|
|
)
|
|
ins_side = bool(
|
|
ins_ht is not None
|
|
or update_policy
|
|
or (ins_atts or "").strip()
|
|
or (ins_remark or "").strip()
|
|
)
|
|
if storage_side and storage_ht is None:
|
|
return error_response(400, "请填写入库办理时间")
|
|
if cert_side and cert_ht is None:
|
|
return error_response(400, "请填写储存证办理时间")
|
|
if ins_side and ins_ht is None:
|
|
return error_response(400, "请填写保险办理时间")
|
|
wrote_any = False
|
|
success = 0
|
|
fail = 0
|
|
if storage_ht is not None:
|
|
storage_r = quick_svc.batch_quick_register_storage(
|
|
db,
|
|
items=items,
|
|
reception_source=source,
|
|
tenant_no=tenant_no,
|
|
user_id=user_id,
|
|
handle_time=storage_ht,
|
|
handler_user_id=storage_handler,
|
|
attachments=storage_atts,
|
|
remark=storage_remark,
|
|
do_commit=False,
|
|
)
|
|
success = max(success, storage_r.get("success", 0))
|
|
fail = max(fail, storage_r.get("fail", 0))
|
|
wrote_any = True
|
|
if cert_ht is not None:
|
|
cert_r = quick_svc.batch_handle_storage_certs(
|
|
db,
|
|
items=items,
|
|
tenant_no=tenant_no,
|
|
user_id=user_id,
|
|
handle_time=cert_ht,
|
|
handler_user_id=cert_handler,
|
|
attachments=cert_atts,
|
|
remark=cert_remark,
|
|
mail_cert_time=mail_t,
|
|
update_mail_cert_time=update_mail,
|
|
do_commit=False,
|
|
)
|
|
success = max(success, cert_r.get("success", 0))
|
|
fail = max(fail, cert_r.get("fail", 0))
|
|
wrote_any = True
|
|
if ins_ht is not None:
|
|
ins_r = quick_svc.batch_handle_insurance(
|
|
db,
|
|
items=items,
|
|
tenant_no=tenant_no,
|
|
user_id=user_id,
|
|
handle_time=ins_ht,
|
|
handler_user_id=ins_handler,
|
|
attachments=ins_atts,
|
|
remark=ins_remark,
|
|
policy_send_time=policy_t,
|
|
update_policy_send_time=update_policy,
|
|
do_commit=False,
|
|
)
|
|
success = max(success, ins_r.get("success", 0))
|
|
fail = max(fail, ins_r.get("fail", 0))
|
|
wrote_any = True
|
|
if has_any_customer_service_payload(cs_incoming):
|
|
cs_r = quick_svc.batch_register_customer_service(
|
|
db,
|
|
items=items,
|
|
tenant_no=tenant_no,
|
|
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))
|
|
|