提交

backend/app/crud/sample_code.py

@@ -9,7 +9,36 @@ from sqlalchemy import and_, func, or_
 from sqlalchemy.orm import Session
 
 from app.core.utils import generate_uuid
-from app.models.models import BizSampleCode
+from app.models.models import BizSampleCode, SysUser
+
+
+def _resolve_issued_user_display_name(
+    db: Session,
+    user_id: Optional[str],
+    fallback_name: Optional[str] = None,
+) -> Optional[str]:
+    name = (fallback_name or "").strip()
+    if name:
+        return name
+    uid = (user_id or "").strip()
+    if not uid:
+        return None
+    u = db.query(SysUser).filter(SysUser.id == uid, SysUser.del_flag == 0).first()
+    if not u:
+        return None
+    return (u.nick_name or u.user_no or "").strip() or None
+
+
+def _parse_issued_time(value: Optional[str]) -> Optional[datetime]:
+    raw = (value or "").strip()
+    if not raw:
+        return None
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
+        try:
+            return datetime.strptime(raw, fmt)
+        except ValueError:
+            continue
+    return None
 
 
 def _row_to_dict(r: BizSampleCode) -> Dict[str, Any]:
@@ -83,21 +112,42 @@ def create_sample_code(
     db: Session,
     *,
     sample_code: str,
+    sample_code_type: Optional[str] = None,
     issued: int = 0,
     tenant_no: str = "zkxw",
     order_num: Optional[int] = None,
     remark: Optional[str] = None,
     user_id: Optional[str] = None,
+    issued_user_id: Optional[str] = None,
+    issued_user_name: Optional[str] = None,
+    issued_time: Optional[str] = None,
 ) -> BizSampleCode:
     sc = (sample_code or "").strip()
     if not sc:
         raise ValueError("样本编码不能为空")
     if sample_code_exists(db, sc, tenant_no):
         raise ValueError("样本编码已存在")
+    code_type = (sample_code_type or "").strip() or None
+    if code_type and code_type not in ("zxbm", "qdbm"):
+        raise ValueError("样本编码类型无效，仅支持 zxbm/qdbm")
+    is_issued = 1 if issued else 0
+    issue_uid: Optional[str] = None
+    issue_uname: Optional[str] = None
+    issue_time: Optional[datetime] = None
+    if is_issued:
+        issue_uid = (issued_user_id or "").strip() or None
+        if not issue_uid:
+            raise ValueError("请选择下发用户")
+        issue_uname = _resolve_issued_user_display_name(db, issue_uid, issued_user_name)
+        issue_time = _parse_issued_time(issued_time) or datetime.now()
     obj = BizSampleCode(
         id=generate_uuid(),
         sample_code=sc,
-        issued=1 if issued else 0,
+        sample_code_type=code_type,
+        issued=is_issued,
+        issued_user_id=issue_uid,
+        issued_user_name=issue_uname,
+        issued_time=issue_time,
         tenant_no=tenant_no,
         order_num=order_num if order_num is not None else 0,
         remark=(remark or "").strip() or None,
@@ -140,6 +190,10 @@ def update_sample_code(
         obj.sample_code = sc
     if issued is not None:
         obj.issued = 1 if issued else 0
+        if not obj.issued:
+            obj.issued_user_id = None
+            obj.issued_user_name = None
+            obj.issued_time = None
     if order_num is not None:
         obj.order_num = order_num
     if remark is not None:
@@ -148,8 +202,12 @@ def update_sample_code(
         obj.issued_user_id = issued_user_id or None
     if issued_user_name is not None:
         obj.issued_user_name = issued_user_name or None
+    if obj.issued == 1 and obj.issued_user_id and not (obj.issued_user_name or "").strip():
+        obj.issued_user_name = _resolve_issued_user_display_name(db, obj.issued_user_id)
     if issued_time is not None:
-        obj.issued_time = issued_time or None
+        obj.issued_time = _parse_issued_time(issued_time) if (issued_time or "").strip() else None
+    if obj.issued == 1 and obj.issued_user_id and not obj.issued_time:
+        obj.issued_time = datetime.now()
     obj.update_by = user_id
     obj.update_time = datetime.now()
     db.add(obj)