from typing import List from flask import g import utils.session_utils as session_utils from artists.repositories.artist_team_repository import ArtistTeamRepository from db import db from projects.schemas import ( TeamCountersResponseSchema, UserSearchResponseSchema, UserSearchRequestSchema, TeamMembersUpdateRequestSchema, TeamMemberParams, ProjectSharingResponseSchema, TeamMemberSchema, AddTeamMemberRequestSchema, UpdateDeleteTeamMemberRequestSchema, ) from shared.schemas import TeamMemberResponseSchema from campaigns.repositories.campaigns_repository import CampaignsRepository from services.labels_repository import LabelsRepository from projects.repositories.projects_repository import ProjectsRepository from models.user_project import UserProjectRoles, UserProject from models.projects import Project from services.permission_role_service import PermissionsByRoleService from services.users_repository import UsersRepository from utils.exceptions import NotFound, UnprocessableEntity, BadRequest from services.project_history.history_service import HistoryService from services.project_history.schemas import HistoryTeamMemberSchema from utils.list_utils import diff from utils.session_utils import has_session_changes class ProjectTeamService: projects_repository = ProjectsRepository() campaigns_repository = CampaignsRepository() labels_repository = LabelsRepository() permissions_by_role_service = PermissionsByRoleService() users_repository = UsersRepository() artist_team_repository = ArtistTeamRepository() history_service = HistoryService() def search_users(self, project_id: int, params: UserSearchRequestSchema) -> List[UserSearchResponseSchema]: users = self.projects_repository.search_users(project_id, search=params.freeText) return [UserSearchResponseSchema(user) for user in users] def get_project_team_members(self, project_id: int) -> List[TeamMemberResponseSchema]: project = self.projects_repository.get_project_by_id(project_id) team_members = self.projects_repository.get_team_members_by_project(project) return [ TeamMemberResponseSchema( user=team_member, roles=self.permissions_by_role_service.get_collaborator_roles_for_project( user_id=team_member.id, project=project, ), user_labels=self.labels_repository.get_all_labels_for_user(user_id=team_member.id), ) for team_member in team_members ] def get_project_team_counters(self, project_id: int) -> TeamCountersResponseSchema: user_projects_count = self.projects_repository.get_unique_project_users_count(project_id) return TeamCountersResponseSchema(usersCount=user_projects_count) def update_project_team_members( self, project_id: int, params: TeamMembersUpdateRequestSchema ) -> List[TeamMemberResponseSchema]: self.__validate_update_project_team_members(params, project_id) project = self.projects_repository.get_project_by_id(project_id) old_values = self.__history_project_team_members(project) for team_member in params.added: self.__add_team_member(project_id, team_member) for team_member in params.updated: self.__update_team_member(project_id, team_member) for team_member in params.deleted: self.__delete_team_member(project_id, team_member) has_changes = has_session_changes() session_utils.session_flush() team_members = self.get_project_team_members(project_id) if has_changes and (len(params.added) or len(params.deleted)): self.__log_project_users_change(project_id, old_values, team_members) project.touch_last_edit() project_was_claimed = project.is_claimed project.refresh_is_claimed() if project_was_claimed is False and project.is_claimed is True: self.history_service.log_claimed_project(project, g.user_id) session_utils.session_commit() return team_members def __add_team_member(self, project_id: int, team_member: TeamMemberParams): self.__validate_user(team_member.userId) if self.projects_repository.is_users_projects_not_approver_exists(project_id, users_ids=[team_member.userId]): raise UnprocessableEntity("Project user already exists") self.__validate_role(team_member.roleId) role = UserProjectRoles(team_member.roleId) user_project = UserProject( project_id=project_id, user_id=team_member.userId, shared_by=g.user_id, role=team_member.roleId ) if role is UserProjectRoles.EDITOR: self.__assign_categories_to_editor(user_project, team_member.categoryIds) db.session.add(user_project) def __add_team_members(self, project_id: int, users_ids: List[int], role_id: int): self.__validate_users(users_ids) self.__validate_role(role_id) user_project_for_delete = self.projects_repository.get_user_projects_for_users_except_approver( project_id, users_ids ) [db.session.delete(up) for up in user_project_for_delete] session_utils.session_flush() users_projects = [ UserProject(project_id=project_id, user_id=user_id, shared_by=g.user_id, role=role_id) for user_id in users_ids ] db.session.add_all(users_projects) def __update_team_member(self, project_id: int, team_member: TeamMemberParams): self.__validate_user(team_member.userId) self.__validate_role(team_member.roleId) user_project = self.__get_project_user(project_id, team_member.userId) if user_project.shared_by != g.user_id or user_project.role != team_member.roleId: user_project.shared_by = g.user_id user_project.role = team_member.roleId campaign_type_groups = user_project.campaign_type_groups for campaign_type_group in campaign_type_groups: if campaign_type_group.id not in team_member.categoryIds: user_project.campaign_type_groups.remove(campaign_type_group) if team_member.roleId == UserProjectRoles.EDITOR.value: self.__assign_categories_to_editor(user_project, team_member.categoryIds) def __delete_team_member(self, project_id: int, team_member: TeamMemberParams): db.session.delete(self.__get_project_user(project_id, team_member.userId)) def __assign_categories_to_editor(self, user_project, category_ids): if not category_ids: raise BadRequest("Categories should be present for Editor") for category in self.campaigns_repository.get_campaign_type_groups_by_ids(category_ids): if category not in user_project.campaign_type_groups: user_project.campaign_type_groups.append(category) def __get_project_user_by_role_id(self, project_id: int, user_id: int, role_id: int) -> UserProject: user_project = self.projects_repository.get_user_project_by_role(project_id, user_id, role_id) if not user_project: raise NotFound("Project user does not exists") return user_project def __get_project_user(self, project_id: int, user_id: int) -> UserProject: user_project = self.projects_repository.get_user_project(project_id, user_id) if not user_project: raise NotFound("Project user does not exists") return user_project def __validate_update_project_team_members(self, params: TeamMembersUpdateRequestSchema, project_id: int): self.__validate_duplications(params) self.__validate_artist_team([t.userId for t in [*params.added]], project_id) def __validate_artist_team(self, added_users_ids: List[int], project_id: int): artist_team = self.artist_team_repository.get_artist_team_for_project(project_id) # if there is no artist team, there is nothing to validate if artist_team is None: return artist_team_ids = [t.user_id for t in artist_team.artist_team_users] # if two lists intersect if not set(artist_team_ids).isdisjoint(set(added_users_ids)): raise BadRequest("Artist team members can't be added to collaborators") def __validate_duplications(self, params: TeamMembersUpdateRequestSchema): ids = [t.userId for t in [*params.added, *params.updated, *params.deleted]] if len(set(ids)) != len(ids): raise BadRequest("Users should be unique") def __validate_user(self, user_id: int): if not self.users_repository.is_user_exists(user_id): raise NotFound(f"User with id {user_id} does not exists") def __validate_users(self, users_ids: List[int]): users = self.users_repository.get_users_by_ids(users_ids) users_diff = diff(users_ids, [u.id for u in users]) if users_diff: raise NotFound(f"User with id {users_diff} does not exists") def __validate_role(self, role: int): if not UserProjectRoles.has_value(role): raise NotFound(f"Role with id {role} does not exists") def __history_project_team_members(self, project: Project) -> List[HistoryTeamMemberSchema]: team_members = self.projects_repository.get_team_members_by_project(project) return [ HistoryTeamMemberSchema( user_id=team_member.id, email=team_member.email, name=team_member.name, roles=self.permissions_by_role_service.get_collaborator_roles_for_project( user_id=team_member.id, project=project, ), ) for team_member in team_members ] def __log_project_users_change( self, project_id: int, old_values: List[HistoryTeamMemberSchema], current_members: List[TeamMemberResponseSchema], ): old_values_ids = list(map(lambda x: x.id, old_values)) current_members_ids = list(map(lambda x: x.id, current_members)) removed_ids = set(old_values_ids) - set(current_members_ids) if removed_ids: removed_users = list(filter(lambda user: user.id in removed_ids, old_values)) self.history_service.log_project_team_members_updated(project_id, removed_users, [], g.user_id) new_ids = set(current_members_ids) - set(old_values_ids) if new_ids: new_user = list(filter(lambda user: user.id in new_ids, current_members)) new_values = [ HistoryTeamMemberSchema(user_id=member.id, name=member.name, email=member.email, roles=member.roles) for member in new_user ] self.history_service.log_project_team_members_updated(project_id, [], new_values, g.user_id) def get_project_team_members_for_sharing(self, project_id: int): user_projects = self.projects_repository.get_user_projects_by_project(project_id) artist_team_and_label_admins = self.artist_team_repository.get_artist_team_users_and_label_admins_for_project( project_id ) project_team = list(filter(lambda user: user.role != UserProjectRoles.APPROVER.value, user_projects)) approvers = list(filter(lambda user: user.role == UserProjectRoles.APPROVER.value, user_projects)) return ProjectSharingResponseSchema( projectTeam=[TeamMemberSchema(user) for user in project_team], approvers=[TeamMemberSchema(user) for user in approvers], artistTeamAndAdmins=[TeamMemberSchema(user) for user in artist_team_and_label_admins] ) def add_project_team_members(self, project_id: int, params: AddTeamMemberRequestSchema): project = self.projects_repository.get_project_by_id(project_id) self.__validate_artist_team(params.users_ids, project_id) old_values = self.__history_project_team_members(project) self.__add_team_members(project_id, params.users_ids, params.role_id) has_changes = has_session_changes() session_utils.session_flush() team_members = self.get_project_team_members(project_id) if has_changes: self.__log_project_users_change(project_id, old_values, team_members) project.touch_last_edit() project_was_claimed = project.is_claimed project.refresh_is_claimed() if project_was_claimed is False and project.is_claimed is True: self.history_service.log_claimed_project(project, g.user_id) session_utils.session_commit() def delete_user_project_role(self, project_id: int, params: UpdateDeleteTeamMemberRequestSchema): project = self.projects_repository.get_project_by_id(project_id) old_values = self.__history_project_team_members(project) user_project = self.projects_repository.get_user_project_by_role( project_id, user_id=params.user_id, role_id=params.role_id ) if user_project is None: raise NotFound("Project user does not exists") db.session.delete(user_project) has_changes = has_session_changes() session_utils.session_flush() team_members = self.get_project_team_members(project_id) if has_changes: self.__log_project_users_change(project_id, old_values, team_members) project.touch_last_edit() project_was_claimed = project.is_claimed project.refresh_is_claimed() if project_was_claimed is False and project.is_claimed is True: self.history_service.log_claimed_project(project, g.user_id) session_utils.session_commit() def update_project_team_member_role(self, project_id: int, params: UpdateDeleteTeamMemberRequestSchema): if params.role_id not in [UserProjectRoles.EDITOR.value, UserProjectRoles.VIEWER.value]: raise BadRequest(f"{params.role_id} not supportable for update") self.__validate_user(params.user_id) self.__validate_role(params.role_id) user_project = self.__get_project_user(project_id, params.user_id) if user_project.role != params.role_id: user_project.shared_by = g.user_id user_project.role = params.role_id session_utils.session_commit()