mtsd/backend/app/crud/base.py

"""CRUD基础类"""
from typing import TypeVar, Generic, Type, Optional, List, Any
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
import re

from app.core.utils import generate_id
from app.core.database import Base
from sqlalchemy import func

ModelType = TypeVar("ModelType", bound=Base)


def camel_to_snake(name: str) -> str:
    """驼峰命名转蛇形命名"""
    s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
    return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()


def get_model_attr(obj, key: str):
    """获取模型的属性名，支持驼峰转蛇形"""
    # 先尝试直接匹配
    if hasattr(obj, key):
        return key
    # 尝试驼峰转蛇形
    snake_key = camel_to_snake(key)
    if hasattr(obj, snake_key):
        return snake_key
    # 如果都没有，返回None
    return None


class CRUDBase(Generic[ModelType]):
    """CRUD基类"""

    def __init__(self, model: Type[ModelType]):
        self.model = model

    def get(self, db: Session, id: str) -> Optional[ModelType]:
        """根据ID获取单条记录"""
        return db.query(self.model).filter(self.model.id == id).first()

    def get_multi(
        self,
        db: Session,
        skip: int = 0,
        limit: int = 100,
        filters: List[Any] = None,
        order_by: str = None,
        order_type: str = "desc"
    ) -> List[ModelType]:
        """获取多条记录"""
        query = db.query(self.model)
        if filters:
            query = query.filter(*filters)
        if order_by:
            order_column = getattr(self.model, order_by, None)
            if order_column:
                if order_type == "asc":
                    query = query.order_by(order_column.asc())
                else:
                    query = query.order_by(order_column.desc())
        else:
            query = query.order_by(self.model.create_at.desc())
        return query.offset(skip).limit(limit).all()

    def get_all(self, db: Session, filters: List[Any] = None) -> List[ModelType]:
        """获取所有记录"""
        query = db.query(self.model)
        if filters:
            query = query.filter(*filters)
        return query.all()

    def create(self, db: Session, obj_in: dict, creator: dict = None) -> ModelType:
        """创建记录"""
        obj = self.model()
        for key, value in obj_in.items():
            attr_name = get_model_attr(obj, key)
            if attr_name:
                setattr(obj, attr_name, value)
        # 设置ID
        if not obj.id:
            obj.id = generate_id()
        # 设置创建人信息
        if creator:
            if not obj.create_by and "user_id" in creator:
                obj.create_by = creator["user_id"]
            if not obj.create_name and "real_name" in creator:
                obj.create_name = creator["real_name"]
        db.add(obj)
        db.commit()
        db.refresh(obj)
        return obj

    def update(self, db: Session, id: str, obj_in: dict, updater: dict = None) -> Optional[ModelType]:
        """更新记录"""
        obj = self.get(db, id)
        if obj:
            for key, value in obj_in.items():
                attr_name = get_model_attr(obj, key)
                if attr_name and key not in ["id", "create_at", "create_by", "create_name"]:
                    setattr(obj, attr_name, value)
            # 设置更新人信息
            if updater:
                if "user_id" in updater:
                    obj.update_by = updater["user_id"]
                if "real_name" in updater:
                    obj.update_name = updater.get("real_name", obj.create_name)
            obj.update_at = func.now()
            db.commit()
            db.refresh(obj)
        return obj

    def delete(self, db: Session, id: str) -> bool:
        """删除记录（软删除）"""
        obj = self.get(db, id)
        if obj:
            if hasattr(obj, "is_del"):
                obj.is_del = 1
            else:
                db.delete(obj)
            db.commit()
            return True
        return False

    def hard_delete(self, db: Session, id: str) -> bool:
        """物理删除"""
        obj = self.get(db, id)
        if obj:
            db.delete(obj)
            db.commit()
            return True
        return False

    def count(self, db: Session, filters: List[Any] = None) -> int:
        """统计数量"""
        query = db.query(self.model)
        if filters:
            query = query.filter(*filters)
        return query.count()

    def exists(self, db: Session, filters: List[Any] = None) -> bool:
        """检查记录是否存在"""
        query = db.query(self.model)
        if filters:
            query = query.filter(*filters)
        return query.first() is not None


class HistoryCRUD(CRUDBase[ModelType]):
    """历史表CRUD类"""

    def create_history(self, db: Session, original_data: dict) -> ModelType:
        """创建历史记录"""
        obj = self.model()
        # 复制原始数据到历史表
        for key, value in original_data.items():
            if hasattr(obj, key):
                setattr(obj, key, value)
        # 设置历史记录创建时间
        if hasattr(obj, "his_create_at"):
            obj.his_create_at = func.now()
        db.add(obj)
        db.commit()
        db.refresh(obj)
        return obj