From b303afb272e02ecc9b4487e5145072e5180b2443 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Wed, 20 May 2026 00:56:24 +0800 Subject: [PATCH] feat(crud): add cell product CRUD operations --- backend/app/crud/cell_product.py | 207 +++++++++++++++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 backend/app/crud/cell_product.py diff --git a/backend/app/crud/cell_product.py b/backend/app/crud/cell_product.py new file mode 100644 index 000000000..83c8a71cf --- /dev/null +++ b/backend/app/crud/cell_product.py @@ -0,0 +1,207 @@ +"""细胞产品 CRUD（biz_cell_product）""" +from __future__ import annotations + +import math +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple + +from sqlalchemy import and_, func, or_ +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.models.models import BizCellProduct + + +def _row_to_dict(c: BizCellProduct) -> Dict[str, Any]: + return { + "id": c.id, + "productCode": c.product_code, + "productName": c.product_name, + "productTypeId": c.product_type_id or "", + "specId": c.spec_id or "", + "batchNo": c.batch_no or "", + "storageSlotId": c.storage_slot_id or "", + "productStatus": c.product_status or "", + "sourceType": c.source_type or "", + "productionTaskId": c.production_task_id or "", + "quantity": float(c.quantity) if c.quantity is not None else None, + "unit": c.unit or "", + "expiryDate": c.expiry_date.strftime("%Y-%m-%d %H:%M:%S") if c.expiry_date else "", + "productImages": c.product_images or "", + "tenantNo": c.tenant_no, + "orderNum": c.order_num if c.order_num is not None else 0, + "remark": c.remark or "", + "createBy": c.create_by or "", + "createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "", + "updateBy": c.update_by or "", + "updateTime": c.update_time.strftime("%Y-%m-%d %H:%M:%S") if c.update_time else "", + } + + +def list_cell_products( + db: Session, + *, + tenant_no: str, + page: int = 1, + page_size: int = 20, + product_code: Optional[str] = None, + product_name: Optional[str] = None, + product_type_id: Optional[str] = None, + product_status: Optional[str] = None, +) -> Dict[str, Any]: + def _filtered(): + q0 = db.query(BizCellProduct).filter( + BizCellProduct.del_flag == 0, + BizCellProduct.tenant_no == tenant_no, + ) + pc = (product_code or "").strip() + if pc: + q0 = q0.filter(BizCellProduct.product_code.like(f"%{pc}%")) + pn = (product_name or "").strip() + if pn: + q0 = q0.filter(BizCellProduct.product_name.like(f"%{pn}%")) + if product_type_id: + q0 = q0.filter(BizCellProduct.product_type_id == product_type_id) + if product_status: + q0 = q0.filter(BizCellProduct.product_status == product_status) + return q0 + + total = _filtered().order_by(None).with_entities(func.count(BizCellProduct.id)).scalar() or 0 + page = max(1, page) + page_size = min(max(1, page_size), 100) + rows = ( + _filtered() + .order_by(BizCellProduct.create_time.desc()) + .offset((page - 1) * page_size) + .limit(page_size) + .all() + ) + return {"items": [_row_to_dict(x) for x in rows], "total": int(total)} + + +def get_cell_product(db: Session, id: str, tenant_no: str) -> Optional[BizCellProduct]: + return ( + db.query(BizCellProduct) + .filter( + and_( + BizCellProduct.id == id, + BizCellProduct.tenant_no == tenant_no, + BizCellProduct.del_flag == 0, + ) + ) + .first() + ) + + +def create_cell_product( + db: Session, + *, + tenant_no: str, + user_id: Optional[str] = None, + create_dept_id: Optional[str] = None, + data: Any = None, +) -> BizCellProduct: + code = (data.product_code or "").strip() + if not code: + raise ValueError("产品编码不能为空") + + existing = db.query(BizCellProduct).filter( + BizCellProduct.tenant_no == tenant_no, + BizCellProduct.product_code == code, + BizCellProduct.del_flag == 0, + ).first() + if existing: + raise ValueError(f"产品编码 {code} 已存在") + + obj = BizCellProduct( + id=generate_uuid(), + product_code=code, + product_name=(data.product_name or "").strip(), + product_type_id=(data.product_type_id or "").strip() or None, + spec_id=(data.spec_id or "").strip() or None, + batch_no=(data.batch_no or "").strip() or None, + storage_slot_id=(data.storage_slot_id or "").strip() or None, + product_status=data.product_status or "zk", + source_type=data.source_type or "manual", + quantity=data.quantity, + unit=(data.unit or "").strip() or None, + expiry_date=data.expiry_date, + product_images=(data.product_images or "").strip() or None, + tenant_no=tenant_no, + order_num=0, + remark=(data.remark or "").strip() or None, + create_by=user_id, + create_dept_id=create_dept_id or "", + create_time=datetime.now(), + update_by=user_id, + update_time=datetime.now(), + del_flag=0, + ) + db.add(obj) + db.commit() + db.refresh(obj) + return obj + + +def update_cell_product( + db: Session, + id: str, + tenant_no: str, + user_id: Optional[str] = None, + data: Any = None, +) -> BizCellProduct: + obj = get_cell_product(db, id, tenant_no) + if not obj: + raise ValueError("细胞产品不存在") + + if data.product_name is not None: + obj.product_name = (data.product_name or "").strip() + if data.product_type_id is not None: + obj.product_type_id = (data.product_type_id or "").strip() or None + if data.spec_id is not None: + obj.spec_id = (data.spec_id or "").strip() or None + if data.batch_no is not None: + obj.batch_no = (data.batch_no or "").strip() or None + if data.storage_slot_id is not None: + obj.storage_slot_id = (data.storage_slot_id or "").strip() or None + if data.product_status is not None: + obj.product_status = data.product_status + if data.quantity is not None: + obj.quantity = data.quantity + if data.unit is not None: + obj.unit = (data.unit or "").strip() or None + if data.expiry_date is not None: + obj.expiry_date = data.expiry_date + if data.product_images is not None: + obj.product_images = (data.product_images or "").strip() or None + if data.remark is not None: + obj.remark = (data.remark or "").strip() or None + + obj.update_by = user_id + obj.update_time = datetime.now() + db.add(obj) + db.commit() + db.refresh(obj) + return obj + + +def delete_cell_product( + db: Session, + id: str, + tenant_no: str, + user_id: Optional[str] = None, +) -> None: + obj = get_cell_product(db, id, tenant_no) + if not obj: + raise ValueError("细胞产品不存在") + obj.del_flag = 1 + obj.update_by = user_id + obj.update_time = datetime.now() + db.add(obj) + db.commit() + + +def total_pages(total: int, page_size: int) -> int: + if total <= 0: + return 0 + return int(math.ceil(total / float(page_size)))