Maas-Model-platform/backend/app/services/user_local_model_permission_service.py

"""
用户本地模型权限服务

处理用户对本地模型的权限管理逻辑
"""
from typing import List, Dict, Optional
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError

from app.models.user_local_model_permission import UserLocalModelPermission
from app.models.model import ModelNew


class UserLocalModelPermissionService:
    """
    用户本地模型权限服务类
    """
    
    def __init__(self, db: Session):
        self.db = db
    
    def get_user_model_permissions(self, user_id: str) -> List[Dict]:
        """
        获取用户的所有本地模型权限
        
        Args:
            user_id: 用户ID
            
        Returns:
            权限列表，包含模型信息和权限状态
        """
        try:
            # 获取所有本地模型
            local_models = self.db.query(ModelNew).filter(ModelNew.is_local == True).all()
            
            # 获取用户的权限设置
            permissions = self.db.query(UserLocalModelPermission).filter(
                UserLocalModelPermission.user_id == user_id
            ).all()
            
            # 构建权限字典
            permission_dict = {p.model_id: p.has_access for p in permissions}
            
            # 构建结果列表
            result = []
            for model in local_models:
                result.append({
                    "model_id": model.id,
                    "model_name": model.display_name,
                    "model_title": model.model_code,
                    "base_url": model.base_url,
                    "has_access": permission_dict.get(model.id, False)
                })
            
            return result
        except Exception as e:
            # 如果发生错误，返回空列表
            return []
    
    async def update_user_model_permission(self, user_id: str, model_id: int, has_access: bool) -> bool:
        """
        更新用户对本地模型的权限
        
        Args:
            user_id: 用户ID
            model_id: 模型ID
            has_access: 是否有权限访问
            
        Returns:
            是否更新成功
        """
        # 检查模型是否存在且是本地模型
        model = self.db.query(ModelNew).filter(
            ModelNew.id == model_id,
            ModelNew.is_local == True
        ).first()
        
        if not model:
            return False
        
        # 查找现有权限记录
        permission = self.db.query(UserLocalModelPermission).filter(
            UserLocalModelPermission.user_id == user_id,
            UserLocalModelPermission.model_id == model_id
        ).first()
        
        if permission:
            # 更新现有记录
            permission.has_access = has_access
        else:
            # 创建新记录
            permission = UserLocalModelPermission(
                user_id=user_id,
                model_id=model_id,
                has_access=has_access
            )
            self.db.add(permission)
        
        try:
            self.db.commit()
            # 删除相关缓存
            from app.services.cache_service import CacheService
            await CacheService.delete_user_permission(user_id, model_id)
            await CacheService.delete_user_local_models(user_id)
            return True
        except IntegrityError:
            self.db.rollback()
            return False
    
    async def update_user_all_model_permissions(self, user_id: str, has_access: bool) -> bool:
        """
        更新用户对所有本地模型的权限
        
        Args:
            user_id: 用户ID
            has_access: 是否有权限访问
            
        Returns:
            是否更新成功
        """
        # 获取所有本地模型
        local_models = self.db.query(ModelNew).filter(ModelNew.is_local == True).all()
        
        try:
            for model in local_models:
                # 查找现有权限记录
                permission = self.db.query(UserLocalModelPermission).filter(
                    UserLocalModelPermission.user_id == user_id,
                    UserLocalModelPermission.model_id == model.id
                ).first()
                
                if permission:
                    # 更新现有记录
                    permission.has_access = has_access
                else:
                    # 创建新记录
                    permission = UserLocalModelPermission(
                        user_id=user_id,
                        model_id=model.id,
                        has_access=has_access
                    )
                    self.db.add(permission)
            
            self.db.commit()
            # 删除相关缓存
            from app.services.cache_service import CacheService
            await CacheService.delete_user_local_models(user_id)
            # 也可以删除每个模型的权限缓存，但为了性能考虑，这里只删除用户本地模型列表缓存
            return True
        except Exception:
            self.db.rollback()
            return False
    
    async def check_user_model_access(self, user_id: str, model_id: int) -> bool:
        """
        检查用户是否有权限访问指定本地模型
        
        Args:
            user_id: 用户ID
            model_id: 模型ID
            
        Returns:
            是否有权限访问
        """
        # 检查模型是否存在且是本地模型
        model = self.db.query(ModelNew).filter(
            ModelNew.id == model_id,
            ModelNew.is_local == True
        ).first()
        
        if not model:
            return False
        
        # 检查本地模型是否启用
        from app.services.system_config_manager import get_config_bool
        if get_config_bool("enable_local_models", True):
            # 如果本地模型启用，所有用户都有权限访问所有本地模型
            return True
        
        # 从缓存获取权限
        from app.services.cache_service import CacheService
        has_access = await CacheService.get_user_permission(user_id, model_id)
        
        if has_access is not None:
            return has_access
        
        # 从数据库获取
        permission = self.db.query(UserLocalModelPermission).filter(
            UserLocalModelPermission.user_id == user_id,
            UserLocalModelPermission.model_id == model_id
        ).first()
        
        has_access = permission.has_access if permission else False
        # 缓存权限信息
        await CacheService.set_user_permission(user_id, model_id, has_access)
        
        return has_access