from typing import List import datetime from datetime import date from flask import g from artists.repositories.artist_team_repository import ArtistTeamRepository from db import db from labels.schemas import ( TeamMemberPermissionsResponse, LabelPermissionsCounterResponse, LabelArtistResponse, TeamMemberPermissionsUserResponse, LabelPermissionsArtistResponse ) from labels.repositories.labels_permissions_repository import LabelPermissionsRepository from migrations.fixtures.delphi_tables.v7.triggers.queries.user_projects_queries import UserProjectsQueries from models import Project from projects.repositories.projects_repository import ProjectsRepository from projects.services.project_history_service import ProjectHistoryService from projects.services.projects_service import ProjectsService from utils import session_utils from utils.trigger_utils import disable_trigger class LabelPermissionsService: labels_perm_repository = LabelPermissionsRepository() artist_team_repository = ArtistTeamRepository() projects_repository = ProjectsRepository() project_service = ProjectsService() project_history_service = ProjectHistoryService() def get_label_permissions_counters(self, label_id): team_members_counter = self.labels_perm_repository.count_team_members(label_id) artists_counter = self.labels_perm_repository.count_artists(label_id) return LabelPermissionsCounterResponse(team_members=team_members_counter, artists=artists_counter) def claim_projects(self, projects: List[Project]): self.projects_repository.set_projects_claimed([project.id for project in projects]) session_utils.session_flush() for project in projects: project_state = self.project_history_service.get_project_state(project) project.name = project.prs_title or project.ccp_title start_date = ( project.earliest_campaign_start_date or project.earliest_purchase_order_date or project.initial_start_date or date.today() ) end_date = ( project.latest_campaign_end_date or project.latest_purchase_order_date or start_date + datetime.timedelta(days=1) ) project.end_date = end_date if project.initial_start_date is None: project.initial_start_date = start_date self.project_service.set_project_default_media_plan(project=project) project.touch_last_edit() session_utils.session_add(project) session_utils.session_flush() self.projects_repository.refresh_project(project) self.project_history_service.log_project_claimed(project, project_state, g.user_id) Project.index_models(projects) def get_team_members_permissions(self, user_id, label_id): team_members = self.labels_perm_repository.get_team_member_permissions(label_id) return [TeamMemberPermissionsResponse( team_member.id, team_member.name, team_member.email, team_member.role, team_member.artists) for team_member in team_members ] @disable_trigger(table="ArtistTeamUser", trigger="log_artist_team_user_changes_insert_trigger") def add_user_to_artist_teams(self, label_id: int, user_id: int, artists_ids: List[str]): def __add_delphi_user_project_for_claimed_projects(): projects_data = UserProjectsQueries.get_user_projects( where_clause=f"project.is_deleted IS FALSE AND dec_user.id = {user_id}") db.session.execute(f""" DELETE FROM user_projects where user_id = {user_id}; INSERT INTO user_projects(project_id, user_id) {projects_data}; """) self.artist_team_repository.assign_user_to_multiple_teams_in_label_by_external_id(label_id, user_id, artists_ids) session_utils.session_flush() self.projects_repository.delete_users_from_artists_projects([user_id], artists_ids) projects = self.projects_repository.get_projects_for_artists_from_label(artists_ids, label_id) self.claim_projects(projects) __add_delphi_user_project_for_claimed_projects() session_utils.session_commit() @disable_trigger(table="ArtistTeamUser", trigger="log_artist_team_user_changes_delete_trigger") def delete_user_from_artist_team(self, label_id: int, user_id: int, artists_ids: List[str]): def __delete_delphi_user_projects_for_unclaimed_projects(): projects_data = UserProjectsQueries.get_user_projects( where_clause=f"project.is_deleted IS FALSE AND dec_user.id = {user_id}") db.session.execute(f""" DELETE FROM user_projects where user_id = {user_id}; INSERT INTO user_projects(project_id, user_id) {projects_data}; """) self.artist_team_repository.delete_user_from_artist_teams(user_id, artists_ids, label_id) session_utils.session_flush() self.artist_team_repository.remove_empty_artist_teams(artists_ids, label_id) session_utils.session_flush() projects = self.projects_repository.get_projects_without_project_teams_by_artists_external_ids( artists_ids, label_id ) self.projects_repository.unclaim_projects([project.id for project in projects]) __delete_delphi_user_projects_for_unclaimed_projects() session_utils.session_commit() @disable_trigger(table="ArtistTeamUser", trigger="log_artist_team_user_changes_delete_trigger") def delete_users_from_artist_teams(self, label_id: int, users_ids: List[int], artist_id: str): def __delete_delphi_user_projects_for_unclaimed_projects(): users_ids_str = ','.join(map(str, users_ids)) projects_data = UserProjectsQueries.get_user_projects( where_clause=f"project.is_deleted IS FALSE AND dec_user.id IN ({users_ids_str})") db.session.execute(f""" DELETE FROM user_projects where user_id IN ({users_ids_str}); INSERT INTO user_projects(project_id, user_id) {projects_data}; """) self.artist_team_repository.delete_multiple_users_from_artist_teams(users_ids, artist_id, label_id) session_utils.session_flush() self.artist_team_repository.remove_empty_artist_teams([artist_id], label_id) session_utils.session_flush() projects = self.projects_repository.get_projects_without_project_teams_by_artists_external_ids( [artist_id], label_id ) self.projects_repository.unclaim_projects([project.id for project in projects]) __delete_delphi_user_projects_for_unclaimed_projects() session_utils.session_commit() @disable_trigger(table="ArtistTeamUser", trigger="log_artist_team_user_changes_insert_trigger") def add_users_to_artist_teams(self, label_id: int, users_ids: List[int], artist_id: str): def __add_delphi_user_project_for_claimed_projects(): users_ids_str = ','.join(map(str, users_ids)) projects_data = UserProjectsQueries.get_user_projects( where_clause=f"project.is_deleted IS FALSE AND dec_user.id IN ({users_ids_str})") db.session.execute(f""" DELETE FROM user_projects where user_id IN ({users_ids_str}); INSERT INTO user_projects(project_id, user_id) {projects_data}; """) self.artist_team_repository.assign_multiple_users_to_artist_team_in_label_by_external_id( label_id, users_ids, artist_id) session_utils.session_flush() self.projects_repository.delete_users_from_artists_projects(users_ids, [artist_id]) projects = self.projects_repository.get_projects_for_artists_from_label([artist_id], label_id) self.claim_projects(projects) __add_delphi_user_project_for_claimed_projects() session_utils.session_commit() def get_label_permissions_artists(self, label_id: int): artists = self.labels_perm_repository.get_label_artists(label_id) artists_response = [] for artist, artist_team_members in artists: team_members = [ TeamMemberPermissionsUserResponse( team_member["user_id"], team_member["name"], team_member["email"], None, team_member["role"] ) for team_member in self.sort_artist_team_members_by_role(artist_team_members) if team_member ] artists_response.append( LabelPermissionsArtistResponse( artist=LabelArtistResponse(artist), teamMembers=team_members, ) ) return artists_response def sort_artist_team_members_by_role(self, artist_team_members): return sorted(artist_team_members, key=lambda x: x.get('role', []))