mtsd/backend/app/crud/product.py

"""商品CRUD"""
from typing import Optional, List
from sqlalchemy.orm import Session, defer
from sqlalchemy import and_, func, or_

from app.models.models import BasProduct, BasProductHis
from app.crud.base import CRUDBase


class ProductCRUD(CRUDBase[BasProduct]):
    """商品CRUD操作"""

    def get_by_name(self, db: Session, name: str) -> Optional[BasProduct]:
        """根据名称获取商品"""
        return db.query(BasProduct).filter(
            and_(BasProduct.name == name, BasProduct.is_del == 0)
        ).first()

    def get_active_all(self, db: Session) -> List[BasProduct]:
        """获取所有启用的商品"""
        return db.query(BasProduct).filter(
            and_(BasProduct.is_del == 0, BasProduct.is_active == True)
        ).order_by(BasProduct.create_at.desc()).all()

    def get_multi_by_category(self, db: Session, category_id: str) -> List[BasProduct]:
        """根据分类获取商品"""
        return db.query(BasProduct).filter(
            and_(BasProduct.category_id == category_id, BasProduct.is_del == 0, BasProduct.is_active == True)
        ).order_by(BasProduct.create_at.desc()).all()

    def get_multi_with_filters(
        self,
        db: Session,
        category_id: str = None,
        keyword: str = None,
        is_active: bool = None,
        skip: int = 0,
        limit: int = 10,
        *,
        defer_detail_content: bool = False,
    ) -> tuple[List[BasProduct], int]:
        """多条件查询商品"""
        filters = [BasProduct.is_del == 0]
        if category_id:
            filters.append(BasProduct.category_id == category_id)
        if is_active is not None:
            filters.append(BasProduct.is_active == is_active)
        if keyword:
            filters.append(
                or_(
                    BasProduct.name.like(f"%{keyword}%"),
                    BasProduct.description.like(f"%{keyword}%"),
                )
            )

        total = (
            db.query(func.count(BasProduct.id))
            .filter(*filters)
            .order_by(None)
            .scalar()
            or 0
        )

        query = db.query(BasProduct).filter(*filters)
        if defer_detail_content:
            query = query.options(defer(BasProduct.detail_content))
        products = query.order_by(BasProduct.create_at.desc()).offset(skip).limit(limit).all()
        return products, total

    def update_with_history(self, db: Session, id: str, obj_in: dict, updater: dict = None) -> Optional[BasProduct]:
        """更新商品并记录历史"""
        original = self.get(db, id)
        if not original:
            return None

        his_data = {
            "id": original.id,
            "name": original.name,
            "category_id": original.category_id,
            "unit_id": original.unit_id,
            "measure_value": getattr(original, "measure_value", None),
            "description": original.description,
            "image_url": original.image_url,
            "is_active": original.is_active,
            "create_at": original.create_at,
            "create_by": original.create_by,
            "create_name": original.create_name,
            "update_at": original.update_at,
            "update_by": original.update_by,
        }
        his = BasProductHis(**his_data)
        db.add(his)

        return self.update(db, id, obj_in, updater)

    def soft_delete_with_history(self, db: Session, id: str) -> bool:
        """删除商品并记录历史"""
        original = self.get(db, id)
        if not original:
            return False

        his_data = {
            "id": original.id,
            "name": original.name,
            "category_id": original.category_id,
            "unit_id": original.unit_id,
            "measure_value": getattr(original, "measure_value", None),
            "description": original.description,
            "image_url": original.image_url,
            "is_active": original.is_active,
            "create_at": original.create_at,
            "create_by": original.create_by,
            "create_name": original.create_name,
            "update_at": original.update_at,
            "update_by": original.update_by,
        }
        his = BasProductHis(**his_data)
        db.add(his)

        return self.delete(db, id)


product_crud = ProductCRUD(BasProduct)