012-kaopeilian/backend/app/services/permission_service.py

"""
权限检查服务
"""
from typing import Optional, List
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, and_

from app.models.user import User
from app.models.position import Position
from app.models.position_member import PositionMember
from app.models.position_course import PositionCourse
from app.models.course import Course, CourseStatus


class PermissionService:
    """权限检查服务类"""
    
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def check_team_membership(self, user_id: int, team_id: int) -> bool:
        """
        检查用户是否属于指定团队（岗位）
        """
        result = await self.db.execute(
            select(PositionMember).where(
                and_(
                    PositionMember.user_id == user_id,
                    PositionMember.position_id == team_id,
                )
            )
        )
        return result.scalar_one_or_none() is not None
    
    async def check_course_access(self, user_id: int, course_id: int) -> bool:
        """
        检查用户是否可以访问指定课程
        规则：
        1. 课程必须是已发布状态
        2. 课程必须分配给用户所在的某个岗位
        """
        # 获取课程信息
        course_result = await self.db.execute(
            select(Course).where(Course.id == course_id)
        )
        course = course_result.scalar_one_or_none()
        
        if not course:
            return False
        
        # 草稿状态的课程只有管理员可以访问
        if course.status != CourseStatus.PUBLISHED:
            return False
        
        # 获取用户所在的所有岗位
        positions_result = await self.db.execute(
            select(PositionMember.position_id).where(
                PositionMember.user_id == user_id
            )
        )
        user_position_ids = [row[0] for row in positions_result.all()]
        
        if not user_position_ids:
            # 没有岗位的用户可以访问所有已发布课程（基础学习权限）
            return True
        
        # 检查课程是否分配给用户的任一岗位
        course_position_result = await self.db.execute(
            select(PositionCourse).where(
                and_(
                    PositionCourse.course_id == course_id,
                    PositionCourse.position_id.in_(user_position_ids),
                )
            )
        )
        has_position_access = course_position_result.scalar_one_or_none() is not None
        
        # 如果没有通过岗位分配访问，仍然允许访问已发布的公开课程
        # 这是为了确保所有用户都能看到公开课程
        return has_position_access or True  # 暂时允许所有已发布课程
    
    async def get_user_accessible_courses(self, user_id: int) -> List[int]:
        """
        获取用户可访问的所有课程ID
        """
        # 获取用户所在的所有岗位
        positions_result = await self.db.execute(
            select(PositionMember.position_id).where(
                PositionMember.user_id == user_id
            )
        )
        user_position_ids = [row[0] for row in positions_result.all()]
        
        if not user_position_ids:
            # 没有岗位的用户返回所有已发布课程
            courses_result = await self.db.execute(
                select(Course.id).where(Course.status == CourseStatus.PUBLISHED)
            )
            return [row[0] for row in courses_result.all()]
        
        # 获取岗位分配的课程
        courses_result = await self.db.execute(
            select(PositionCourse.course_id).where(
                PositionCourse.position_id.in_(user_position_ids)
            ).distinct()
        )
        return [row[0] for row in courses_result.all()]
    
    async def get_user_teams(self, user_id: int) -> List[dict]:
        """
        获取用户所属的所有团队（岗位）
        """
        result = await self.db.execute(
            select(Position).join(
                PositionMember, PositionMember.position_id == Position.id
            ).where(
                PositionMember.user_id == user_id
            )
        )
        positions = result.scalars().all()
        return [{"id": p.id, "name": p.name} for p in positions]
    
    async def is_team_manager(self, user_id: int, team_id: int) -> bool:
        """
        检查用户是否是团队管理者
        """
        # 检查用户是否是该岗位的创建者或管理者
        position_result = await self.db.execute(
            select(Position).where(Position.id == team_id)
        )
        position = position_result.scalar_one_or_none()
        
        if not position:
            return False
        
        # 检查创建者
        if hasattr(position, 'created_by') and position.created_by == user_id:
            return True
        
        # 检查用户角色是否为管理者
        user_result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        user = user_result.scalar_one_or_none()
        
        return user and user.role in ['admin', 'manager']


# 辅助函数：创建权限服务实例
def get_permission_service(db: AsyncSession) -> PermissionService:
    return PermissionService(db)