shanxia2.0/backend/app/crud/quotation.py

"""报价单CRUD"""
from decimal import Decimal
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy import and_, func

from app.models.models import BizQuotation, BizQuotationItem, BizQuotationHis, BizQuotationItemHis
from app.crud.base import CRUDBase
from app.core.utils import generate_id


class QuotationCRUD(CRUDBase[BizQuotation]):
    """报价单CRUD操作"""

    def get_by_inquiry(self, db: Session, inquiry_id: str) -> List[BizQuotation]:
        """根据询价单获取报价列表"""
        return db.query(BizQuotation).filter(
            and_(
                BizQuotation.inquiry_id == inquiry_id,
                BizQuotation.is_del == 0
            )
        ).all()

    def get_by_supplier(self, db: Session, supplier_id: str) -> List[BizQuotation]:
        """根据供应商获取报价列表"""
        return db.query(BizQuotation).filter(
            and_(
                BizQuotation.supplier_id == supplier_id,
                BizQuotation.is_del == 0
            )
        ).order_by(BizQuotation.create_at.desc()).all()

    def get_by_inquiry_and_supplier(
        self,
        db: Session,
        inquiry_id: str,
        supplier_id: str
    ) -> Optional[BizQuotation]:
        """根据询价单和供应商获取报价"""
        return db.query(BizQuotation).filter(
            and_(
                BizQuotation.inquiry_id == inquiry_id,
                BizQuotation.supplier_id == supplier_id,
                BizQuotation.is_del == 0
            )
        ).first()

    def get_latest_by_inquiry_and_supplier(
        self,
        db: Session,
        inquiry_id: str,
        supplier_id: str
    ) -> Optional[BizQuotation]:
        """获取某供应商在某询价单下最新一次报价"""
        return db.query(BizQuotation).filter(
            and_(
                BizQuotation.inquiry_id == inquiry_id,
                BizQuotation.supplier_id == supplier_id,
                BizQuotation.is_del == 0
            )
        ).order_by(BizQuotation.create_at.desc()).first()

    def get_multi_with_filters(
        self,
        db: Session,
        inquiry_id: str = None,
        supplier_id: str = None,
        region_id: str = None,
        status: str = None,
        skip: int = 0,
        limit: int = 10
    ) -> tuple[List[BizQuotation], int]:
        """多条件查询报价单"""
        query = db.query(BizQuotation).filter(BizQuotation.is_del == 0)

        if inquiry_id:
            query = query.filter(BizQuotation.inquiry_id == inquiry_id)
        if supplier_id:
            query = query.filter(BizQuotation.supplier_id == supplier_id)
        if region_id:
            query = query.filter(BizQuotation.region_id == region_id)
        if status:
            query = query.filter(BizQuotation.status == status)

        total = query.count()
        quotations = query.order_by(BizQuotation.create_at.desc()).offset(skip).limit(limit).all()
        return quotations, total

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

        his_data = {
            "id": original.id,
            "inquiry_id": original.inquiry_id,
            "supplier_id": original.supplier_id,
            "region_id": original.region_id,
            "status": original.status,
            "total_amount": original.total_amount,
            "submit_time": original.submit_time,
            "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 = BizQuotationHis(**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,
            "inquiry_id": original.inquiry_id,
            "supplier_id": original.supplier_id,
            "region_id": original.region_id,
            "status": original.status,
            "total_amount": original.total_amount,
            "submit_time": original.submit_time,
            "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 = BizQuotationHis(**his_data)
        db.add(his)

        return self.delete(db, id)


class QuotationItemCRUD(CRUDBase[BizQuotationItem]):
    """报价单项CRUD操作"""

    def get_by_quotation(self, db: Session, quotation_id: str) -> List[BizQuotationItem]:
        """获取报价单的所有报价项"""
        return db.query(BizQuotationItem).filter(
            and_(
                BizQuotationItem.quotation_id == quotation_id,
                BizQuotationItem.is_del == 0
            )
        ).all()

    def get_multi_by_quotations(
        self,
        db: Session,
        quotation_ids: List[str]
    ) -> List[BizQuotationItem]:
        """根据多个报价单ID获取报价项"""
        return db.query(BizQuotationItem).filter(
            and_(
                BizQuotationItem.quotation_id.in_(quotation_ids),
                BizQuotationItem.is_del == 0
            )
        ).all()

    def create_items(
        self,
        db: Session,
        quotation_id: str,
        items: List[dict],
        creator: dict = None
    ) -> List[BizQuotationItem]:
        """批量创建报价项"""
        result = []
        for item in items:
            quotation_item = BizQuotationItem(
                id=generate_id(),
                quotation_id=quotation_id,
                product_id=item.get("product_id"),
                unit_id=item.get("unit_id"),
                price=Decimal(str(item.get("price", 0))),
                create_by=creator.get("user_id") if creator else None,
                create_name=creator.get("real_name") if creator else None
            )
            db.add(quotation_item)
            result.append(quotation_item)
        db.commit()
        for item in result:
            db.refresh(item)
        return result

    def delete_by_quotation(self, db: Session, quotation_id: str) -> bool:
        """删除报价单的所有报价项"""
        items = self.get_by_quotation(db, quotation_id)
        for item in items:
            item.is_del = 1
        db.commit()
        return True


quotation_crud = QuotationCRUD(BizQuotation)
quotation_item_crud = QuotationItemCRUD(BizQuotationItem)