from sqlalchemy import exists, and_, or_ from db import db from typing import List, Optional from utils.list_utils import any_present from models.artist_team import ArtistTeamUser, ArtistTeam from models.user import user_label, User from models.projects import Project, ProjectTargetItem from models.project_entity_type import ProjectEntityType, ProjectEntityAddType from models.user_project import UserProject, UserProjectRoles from services.users_service import UsersService from shared.schemas import RoleSchema, ShortUserModel class PermissionsByRoleService: user_service = UsersService() UNCATEGORIZED_GROUP_ID = 4 def get_collaborator_roles_for_project(self, user_id: int, project: Project) -> Optional[List[RoleSchema]]: user_projects = self.__get_user_project_roles(user_id, project) roles = [up.role for up in user_projects] if self.__is_admin(user_id, project): roles.insert(0, UserProjectRoles.ADMIN.value) return sorted([RoleSchema(UserProjectRoles(role)) for role in roles], key=lambda x: x.id) def get_roles_for_project(self, user_id: int, project: Project) -> Optional[List[RoleSchema]]: roles = self.__get_roles(user_id, project) if roles is None: return None # if role is UserProjectRoles.EDITOR: # return RoleSchema(role, self.__get_editor_categories(user_id, project)) return [RoleSchema(role) for role in roles] def can_view_project(self, user_id: int, project: Project) -> bool: return self.__get_roles(user_id, project) is not None def can_view_campaign(self, user_id: int, project: Project) -> bool: return self.can_view_project(user_id, project) def can_view_team(self, user_id: int, project: Project) -> bool: return self.can_view_project(user_id, project) def can_access_label(self, user_id: int, label_id: int) -> bool: if self.user_service.get_user_label(user_id, label_id) is not None: return True return False def can_submit_approval_request(self, user_id: int, project: Project) -> bool: roles = self.__get_roles(user_id, project) return UserProjectRoles.APPROVER in roles def is_admin(self, user_id: int, project: Project) -> bool: roles = self.__get_roles(user_id, project) return UserProjectRoles.ADMIN in roles def is_editor_or_higher(self, user_id: int, project: Project) -> bool: roles = self.__get_roles(user_id, project) return any_present(roles, ( UserProjectRoles.ADMIN, UserProjectRoles.EDITOR, )) def is_approver_or_editor_and_higher(self, user_id: int, project: Project) -> bool: roles = self.__get_roles(user_id, project) return any_present(roles, ( UserProjectRoles.ADMIN, UserProjectRoles.EDITOR, UserProjectRoles.APPROVER, )) def __get_roles(self, user_id: int, project: Project) -> Optional[List[UserProjectRoles]]: if not project: return None artist_team_user = self.__get_artist_team_user(user_id, project) user_projects = self.__get_user_project_roles(user_id, project) user_project_roles = [up.role for up in user_projects] roles = set() if self.__is_admin(user_id, project): roles.add(UserProjectRoles.ADMIN) elif project.is_confidential: if user_project_roles: roles = {UserProjectRoles(role) for role in user_project_roles} else: roles_list = user_project_roles if self.__is_labelmate(user_id, project): roles_list.append(UserProjectRoles.LABELMATE.value) if artist_team_user: roles_list.append(artist_team_user.role) highest_access_role = min(roles_list, default=None) if highest_access_role: roles.add(UserProjectRoles(highest_access_role)) if user_project_roles and UserProjectRoles.APPROVER.value in user_project_roles: roles.add(UserProjectRoles.APPROVER) if not roles: return None return list(roles) def __is_admin(self, user_id: int, project: Project) -> bool: return db.session.query( exists().where( and_( user_label.c.label_id == project.label_id, user_label.c.user_id == user_id, User.id == user_id, User.is_admin.is_(True), ) ) ).scalar() def __is_labelmate(self, user_id: int, project: Project) -> bool: return db.session.query( exists().where(and_(user_label.c.label_id == project.label_id, user_label.c.user_id == user_id)) ).scalar() def __get_user_project_roles(self, user_id: int, project: Project) -> List[UserProject]: return ( db.session.query(UserProject) .filter(and_( UserProject.project_id == project.id, UserProject.user_id == user_id, )) .all() ) def __get_artist_team_user(self, user_id: int, project: Project) -> Optional[ArtistTeamUser]: return ( db.session.query(ArtistTeamUser) .join(Project, Project.id == project.id) .join( ArtistTeam, and_(ArtistTeamUser.artist_team_id == ArtistTeam.id, ArtistTeam.label_id == project.label_id), ) .join( ProjectTargetItem, and_( Project.id == ProjectTargetItem.project_id, ProjectTargetItem.is_deleted.is_(False), ProjectTargetItem.entity_type == ProjectEntityType.PRIMARY_ARTIST.value, ProjectTargetItem.add_type == ProjectEntityAddType.LOCKED.value, ProjectTargetItem.entity_id == ArtistTeam.artist_id, ), ) .filter(ArtistTeamUser.user_id == user_id) .one_or_none() ) def get_editors_for_project(self, project: Project): editors = ( db.session.query(User) .join(Project, Project.id == project.id) .outerjoin( ProjectTargetItem, and_( ProjectTargetItem.project_id == Project.id, ProjectTargetItem.is_deleted.is_(False), ProjectTargetItem.entity_type == ProjectEntityType.PRIMARY_ARTIST.value, ProjectTargetItem.add_type == ProjectEntityAddType.LOCKED.value, ) ) .outerjoin(ArtistTeam, ArtistTeam.artist_id == ProjectTargetItem.entity_id) .outerjoin(ArtistTeamUser, and_( ArtistTeamUser.artist_team_id == ArtistTeam.id, ArtistTeamUser.user_id == User.id )) .outerjoin( UserProject, and_( UserProject.user_id == User.id, UserProject.role == UserProjectRoles.EDITOR.value, UserProject.project_id == Project.id ) ) .filter( or_( and_(UserProject.id.isnot(None), Project.is_confidential.is_(True)), and_( Project.is_confidential.is_(False), or_( UserProject.id.isnot(None), ArtistTeamUser.id.isnot(None) ) ) ) ) .group_by(User) .order_by(User.name.asc()) .all() ) return [ShortUserModel(editor).__dict__ for editor in editors]