zcglxt/app/crud/asset.py

"""
资产CRUD操作
"""
from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy import and_, or_, func
from sqlalchemy.orm import Session
from app.models.asset import Asset, AssetStatusHistory
from app.schemas.asset import AssetCreate, AssetUpdate


class AssetCRUD:
    """资产CRUD操作类"""

    def get(self, db: Session, id: int) -> Optional[Asset]:
        """根据ID获取资产"""
        return db.query(Asset).filter(
            and_(
                Asset.id == id,
                Asset.deleted_at.is_(None)
            )
        ).first()

    def get_by_code(self, db: Session, code: str) -> Optional[Asset]:
        """根据编码获取资产"""
        return db.query(Asset).filter(
            and_(
                Asset.asset_code == code,
                Asset.deleted_at.is_(None)
            )
        ).first()

    def get_by_serial_number(self, db: Session, serial_number: str) -> Optional[Asset]:
        """根据序列号获取资产"""
        return db.query(Asset).filter(
            and_(
                Asset.serial_number == serial_number,
                Asset.deleted_at.is_(None)
            )
        ).first()

    def get_multi(
        self,
        db: Session,
        skip: int = 0,
        limit: int = 20,
        keyword: Optional[str] = None,
        device_type_id: Optional[int] = None,
        organization_id: Optional[int] = None,
        status: Optional[str] = None,
        purchase_date_start: Optional[Any] = None,
        purchase_date_end: Optional[Any] = None
    ) -> Tuple[List[Asset], int]:
        """获取资产列表"""
        query = db.query(Asset).filter(Asset.deleted_at.is_(None))

        # 关键词搜索
        if keyword:
            query = query.filter(
                or_(
                    Asset.asset_code.ilike(f"%{keyword}%"),
                    Asset.asset_name.ilike(f"%{keyword}%"),
                    Asset.model.ilike(f"%{keyword}%"),
                    Asset.serial_number.ilike(f"%{keyword}%")
                )
            )

        # 筛选条件
        if device_type_id:
            query = query.filter(Asset.device_type_id == device_type_id)
        if organization_id:
            query = query.filter(Asset.organization_id == organization_id)
        if status:
            query = query.filter(Asset.status == status)
        if purchase_date_start:
            query = query.filter(Asset.purchase_date >= purchase_date_start)
        if purchase_date_end:
            query = query.filter(Asset.purchase_date <= purchase_date_end)

        # 排序
        query = query.order_by(Asset.id.desc())

        # 总数
        total = query.count()

        # 分页
        items = query.offset(skip).limit(limit).all()

        return items, total

    def create(
        self,
        db: Session,
        obj_in: AssetCreate,
        asset_code: str,
        creator_id: Optional[int] = None
    ) -> Asset:
        """创建资产"""
        # 计算保修到期日期
        warranty_expire_date = None
        if obj_in.purchase_date and obj_in.warranty_period:
            from datetime import timedelta
            warranty_expire_date = obj_in.purchase_date + timedelta(days=obj_in.warranty_period * 30)

        db_obj = Asset(
            **obj_in.model_dump(),
            asset_code=asset_code,
            status="pending",
            warranty_expire_date=warranty_expire_date,
            created_by=creator_id
        )
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def update(
        self,
        db: Session,
        db_obj: Asset,
        obj_in: AssetUpdate,
        updater_id: Optional[int] = None
    ) -> Asset:
        """更新资产"""
        obj_data = obj_in.model_dump(exclude_unset=True)

        # 重新计算保修到期日期
        if "purchase_date" in obj_data or "warranty_period" in obj_data:
            purchase_date = obj_data.get("purchase_date", db_obj.purchase_date)
            warranty_period = obj_data.get("warranty_period", db_obj.warranty_period)

            if purchase_date and warranty_period:
                from datetime import timedelta
                obj_data["warranty_expire_date"] = purchase_date + timedelta(days=warranty_period * 30)

        for field, value in obj_data.items():
            setattr(db_obj, field, value)

        db_obj.updated_by = updater_id
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def delete(self, db: Session, id: int, deleter_id: Optional[int] = None) -> bool:
        """删除资产（软删除）"""
        obj = self.get(db, id)
        if not obj:
            return False

        obj.deleted_at = func.now()
        obj.deleted_by = deleter_id
        db.add(obj)
        db.commit()
        return True

    def get_by_ids(self, db: Session, ids: List[int]) -> List[Asset]:
        """根据ID列表获取资产"""
        return db.query(Asset).filter(
            and_(
                Asset.id.in_(ids),
                Asset.deleted_at.is_(None)
            )
        ).all()

    def update_status(
        self,
        db: Session,
        asset_id: int,
        new_status: str,
        updater_id: Optional[int] = None
    ) -> Optional[Asset]:
        """更新资产状态"""
        obj = self.get(db, asset_id)
        if not obj:
            return None

        obj.status = new_status
        obj.updated_by = updater_id
        db.add(obj)
        db.commit()
        db.refresh(obj)
        return obj

    def search_by_dynamic_field(
        self,
        db: Session,
        field_name: str,
        field_value: Any,
        skip: int = 0,
        limit: int = 20
    ) -> Tuple[List[Asset], int]:
        """
        根据动态字段搜索资产

        使用JSONB的@>操作符进行高效查询
        """
        query = db.query(Asset).filter(
            and_(
                Asset.deleted_at.is_(None),
                Asset.dynamic_attributes.has_key(field_name)
            )
        )

        # 根据值类型进行不同的查询
        if isinstance(field_value, str):
            query = query.filter(Asset.dynamic_attributes[field_name].astext == field_value)
        else:
            query = query.filter(Asset.dynamic_attributes[field_name] == field_value)

        total = query.count()
        items = query.offset(skip).limit(limit).all()

        return items, total


class AssetStatusHistoryCRUD:
    """资产状态历史CRUD操作类"""

    def create(
        self,
        db: Session,
        asset_id: int,
        old_status: Optional[str],
        new_status: str,
        operation_type: str,
        operator_id: int,
        operator_name: Optional[str] = None,
        organization_id: Optional[int] = None,
        remark: Optional[str] = None,
        extra_data: Optional[Dict[str, Any]] = None
    ) -> AssetStatusHistory:
        """创建状态历史记录"""
        db_obj = AssetStatusHistory(
            asset_id=asset_id,
            old_status=old_status,
            new_status=new_status,
            operation_type=operation_type,
            operator_id=operator_id,
            operator_name=operator_name,
            organization_id=organization_id,
            remark=remark,
            extra_data=extra_data
        )
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def get_by_asset(
        self,
        db: Session,
        asset_id: int,
        skip: int = 0,
        limit: int = 50
    ) -> List[AssetStatusHistory]:
        """获取资产的状态历史"""
        return db.query(AssetStatusHistory).filter(
            AssetStatusHistory.asset_id == asset_id
        ).order_by(
            AssetStatusHistory.created_at.desc()
        ).offset(skip).limit(limit).all()


# 创建全局实例
asset = AssetCRUD()
asset_status_history = AssetStatusHistoryCRUD()