sqxp/backend/app/crud/product.py

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

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
    ) -> tuple[List[BasProduct], int]:
        """多条件查询商品"""
        query = db.query(BasProduct).filter(BasProduct.is_del == 0)
        
        if category_id:
            query = query.filter(BasProduct.category_id == category_id)
        if is_active is not None:
            query = query.filter(BasProduct.is_active == is_active)
        if keyword:
            query = query.filter(
                (BasProduct.name.like(f"%{keyword}%")) |
                (BasProduct.description.like(f"%{keyword}%"))
            )
        
        total = query.count()
        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)