sqxp/backend/app/crud/address.py

"""地址CRUD"""
from typing import Optional, List, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import and_

from app.models.models import BasAddress, BasAddressHis
from app.crud.base import CRUDBase


class AddressCRUD(CRUDBase[BasAddress]):
    """地址CRUD操作"""

    def get_by_user(self, db: Session, user_id: str) -> List[BasAddress]:
        """获取用户的所有地址"""
        return db.query(BasAddress).filter(
            and_(BasAddress.user_id == user_id, BasAddress.is_del == 0)
        ).order_by(BasAddress.is_default.desc(), BasAddress.create_at.desc()).all()

    def get_by_user_paged(
        self, db: Session, user_id: str, skip: int = 0, limit: int = 20
    ) -> Tuple[List[BasAddress], int]:
        """分页获取用户地址（默认优先、再按创建时间倒序）"""
        q = db.query(BasAddress).filter(
            and_(BasAddress.user_id == user_id, BasAddress.is_del == 0)
        )
        total = q.count()
        rows = (
            q.order_by(BasAddress.is_default.desc(), BasAddress.create_at.desc())
            .offset(skip)
            .limit(limit)
            .all()
        )
        return rows, total

    def get_by_user_and_type(self, db: Session, user_id: str, address_type: str) -> List[BasAddress]:
        """获取用户指定类型的地址"""
        return db.query(BasAddress).filter(
            and_(BasAddress.user_id == user_id, BasAddress.address_type == address_type, BasAddress.is_del == 0)
        ).order_by(BasAddress.is_default.desc(), BasAddress.create_at.desc()).all()

    def get_default(self, db: Session, user_id: str, address_type: str = None) -> Optional[BasAddress]:
        """获取用户默认地址"""
        query = db.query(BasAddress).filter(
            and_(BasAddress.user_id == user_id, BasAddress.is_default == True, BasAddress.is_del == 0)
        )
        if address_type:
            query = query.filter(BasAddress.address_type == address_type)
        return query.first()

    def set_default(self, db: Session, user_id: str, address_id: str) -> bool:
        """设置默认地址"""
        # 先取消所有默认地址
        db.query(BasAddress).filter(
            and_(BasAddress.user_id == user_id, BasAddress.is_default == True)
        ).update({"is_default": False})
        
        # 设置新的默认地址
        address = self.get(db, address_id)
        if address and address.user_id == user_id:
            address.is_default = True
            db.commit()
            return True
        return False

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

        his_data = {
            "id": original.id,
            "user_id": original.user_id,
            "address_type": original.address_type,
            "contact_person": original.contact_person,
            "contact_phone": original.contact_phone,
            "province": original.province,
            "city": original.city,
            "district": original.district,
            "detail_address": original.detail_address,
            "is_default": original.is_default,
            "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 = BasAddressHis(**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,
            "user_id": original.user_id,
            "address_type": original.address_type,
            "contact_person": original.contact_person,
            "contact_phone": original.contact_phone,
            "province": original.province,
            "city": original.city,
            "district": original.district,
            "detail_address": original.detail_address,
            "is_default": original.is_default,
            "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 = BasAddressHis(**his_data)
        db.add(his)

        return self.delete(db, id)


address_crud = AddressCRUD(BasAddress)