from dataclasses import dataclass, field from typing import Optional, List from sqlalchemy import func, and_, or_, text from sqlalchemy.orm import aliased, Query from models.user_project import UserProject from models.user import User, user_label from models.projects import Project, ProjectTargetItem from models.campaign import Campaign from models.projects_history import ProjectHistoryItem, HistoryActionType from models.project_entity_type import ProjectEntityType, ProjectEntityAddType from models import ArtistTeam, ArtistTeamUser from utils.elastic_search.query import escape_query from services.project_history.schemas import QueryActionTypes from db import db from shared.queries import Queries as SharedQueries NOTES_CHANGE_HISTORY_LENGTH = 2 PROJECTS_ACTIVITY_LENGTH = 5 ALLOWED_ACTION_TYPES = [ HistoryActionType.ADD_CAMPAIGN, HistoryActionType.REMOVE_CAMPAIGN, HistoryActionType.UPDATE_CAMPAIGN_TIMELINE, HistoryActionType.UPDATE_CAMPAIGN_BUDGET, HistoryActionType.CHANGE_TERRITORIES, HistoryActionType.ADD_PHASE, HistoryActionType.REMOVE_PHASE, HistoryActionType.UPDATE_PROJECT_BUDGET, HistoryActionType.UPDATE_ARTIST, HistoryActionType.CHANGE_PROJECT_TEAM, HistoryActionType.CHANGE_CAMPAIGN_PHASES, HistoryActionType.CHANGE_CAMPAIGN_PLATFORM, HistoryActionType.ADD_PROJECT, HistoryActionType.UPDATE_PROJECT_TARGET_TYPE, HistoryActionType.ADD_PRIMARY_ARTISTS_TO_PROJECT, HistoryActionType.REMOVE_PRIMARY_ARTISTS_FROM_PROJECT, HistoryActionType.ADD_PLAYLISTS_TO_PROJECT, HistoryActionType.REMOVE_PLAYLISTS_FROM_PROJECT, HistoryActionType.ADD_FEATURED_ARTISTS_TO_PROJECT, HistoryActionType.REMOVE_FEATURED_ARTISTS_FROM_PROJECT, HistoryActionType.CLAIM_PROJECT_OWNERSHIP, HistoryActionType.APPROVE_DSP_CAMPAIGN, HistoryActionType.REJECT_DSP_CAMPAIGN, HistoryActionType.ADD_MEDIA_PLAN, HistoryActionType.REMOVE_MEDIA_PLAN, HistoryActionType.UPDATE_MEDIA_PLAN_NAME, HistoryActionType.UPDATE_MEDIA_PLAN_ORDER, HistoryActionType.UPDATE_PHASE_NAME, HistoryActionType.APPROVAL_REQUEST, HistoryActionType.SUBMIT_APPROVAL_REQUEST, HistoryActionType.CHANGE_CAMPAIGN ] RECENT_CAMPAIGN_ALLOWED_ACTION_TYPES = [ HistoryActionType.CHANGE_TERRITORIES, HistoryActionType.ADD_PHASE, HistoryActionType.REMOVE_PHASE, HistoryActionType.UPDATE_PROJECT_BUDGET, HistoryActionType.CHANGE_PROJECT_TEAM, HistoryActionType.ADD_PROJECT, HistoryActionType.UPDATE_PROJECT_TARGET_TYPE, HistoryActionType.ADD_PRIMARY_ARTISTS_TO_PROJECT, HistoryActionType.ADD_FEATURED_ARTISTS_TO_PROJECT, HistoryActionType.REMOVE_PRIMARY_ARTISTS_FROM_PROJECT, HistoryActionType.REMOVE_FEATURED_ARTISTS_FROM_PROJECT, HistoryActionType.ADD_PLAYLISTS_TO_PROJECT, HistoryActionType.REMOVE_PLAYLISTS_FROM_PROJECT, HistoryActionType.CLAIM_PROJECT_OWNERSHIP, HistoryActionType.APPROVE_DSP_CAMPAIGN, HistoryActionType.REJECT_DSP_CAMPAIGN, HistoryActionType.ADD_MEDIA_PLAN, HistoryActionType.REMOVE_MEDIA_PLAN, HistoryActionType.UPDATE_MEDIA_PLAN_NAME, HistoryActionType.UPDATE_MEDIA_PLAN_ORDER, HistoryActionType.UPDATE_PHASE_NAME, HistoryActionType.APPROVAL_REQUEST, HistoryActionType.SUBMIT_APPROVAL_REQUEST, HistoryActionType.CHANGE_CAMPAIGN ] @dataclass class ProjectHistoryItemModel: action: HistoryActionType user_id: Optional[str] = None project_id: Optional[int] = None media_plan_id: Optional[int] = None campaign_id: Optional[int] = None old_value: Optional[dict] = None new_value: Optional[dict] = None keywords: Optional[str] = field(default="") @dataclass class ProjectHistoryFetchResult: item: ProjectHistoryItem role: Optional[int] = None class ProjectHistoryRepository: def store_project_history_item(self, model: ProjectHistoryItemModel): item = ProjectHistoryItem() item.user_id = model.user_id item.project_id = model.project_id item.media_plan_id = model.media_plan_id item.campaign_id = model.campaign_id item.action_id = model.action.value item.old_value = model.old_value item.new_value = model.new_value item.keywords = model.keywords db.session.add(item) db.session.commit() def get_project_history_items_count( self, project_id: int, search: Optional[str], action_type: Optional[int] ) -> int: query = self.__get_project_history_query(project_id) query = self.__action_type_query(query, action_type) if search: query = query.filter(self.__search_in_history_query(search)) return query.count() def get_project_history( self, project_id: int, offset: int, limit: int, search: Optional[str], action_type: Optional[int] ) -> List[ProjectHistoryFetchResult]: return self.__get_project_history(project_id, offset, limit, action_type, search) def get_recent_activity(self, user_id: int) -> List[ProjectHistoryItem]: return ( db.session.query(ProjectHistoryItem) .join(Project, and_(Project.id == ProjectHistoryItem.project_id, Project.is_deleted.is_(False))) .outerjoin(UserProject, UserProject.project_id == Project.id) .filter( UserProject.user_id == user_id, ProjectHistoryItem.user_id != user_id, ProjectHistoryItem.action_id.in_([item.value for item in RECENT_CAMPAIGN_ALLOWED_ACTION_TYPES]), ) .order_by(ProjectHistoryItem.created_at.desc()) .limit(PROJECTS_ACTIVITY_LENGTH) .all() ) def get_project_notes_changes(self, project_id: int) -> List[ProjectHistoryFetchResult]: results = ( self.__get_project_history_query(project_id) .filter(ProjectHistoryItem.action_id == HistoryActionType.UPDATE_PROJECT_NOTES.value) .order_by(ProjectHistoryItem.created_at.desc()) .limit(NOTES_CHANGE_HISTORY_LENGTH) .all() ) return self.__map_database_response(results) def __get_project_history( self, project_id: int, offset: int, limit: int, action_type: Optional[int], search: Optional[str] = None ) -> List[ProjectHistoryFetchResult]: query = self.__get_project_history_query(project_id) query = self.__action_type_query(query, action_type) if search: query = query.filter(self.__search_in_history_query(search)) results = ( query.order_by(ProjectHistoryItem.created_at.desc(), ProjectHistoryItem.id) .limit(limit) .offset(offset) .all() ) return self.__map_database_response(results) def __get_project_history_query(self, project_id: int): creator = aliased(User, name="creator") aliasedCampaign = aliased(Campaign, name="campaign") role_query = SharedQueries.role_for_project_query(creator.id) # We have history events like add/remove artist without owner(user_id is null) # so we need to use left join owner return ( db.session.query(ProjectHistoryItem, func.min(role_query)) .select_from(ProjectHistoryItem) .outerjoin(creator, ProjectHistoryItem.user_id == creator.id) .join(Project, and_(Project.id == project_id, ProjectHistoryItem.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, and_( ArtistTeam.artist_id == ProjectTargetItem.entity_id, ArtistTeam.label_id == Project.label_id, Project.is_confidential.is_(False), ), ) .outerjoin( ArtistTeamUser, and_( creator.is_admin.is_(False), ArtistTeam.id == ArtistTeamUser.artist_team_id, ArtistTeamUser.user_id == creator.id, ), ) .outerjoin(user_label, and_(user_label.c.user_id == creator.id, user_label.c.label_id == Project.label_id)) .outerjoin(aliasedCampaign, ProjectHistoryItem.campaign_id == aliasedCampaign.id) .outerjoin( UserProject, and_( ArtistTeamUser.role.is_(None), creator.is_admin.is_(False), UserProject.project_id == Project.id, UserProject.user_id == creator.id, ), ) .group_by(ProjectHistoryItem.id) ) def __map_database_response(self, results: List[tuple]) -> List[ProjectHistoryFetchResult]: return [ProjectHistoryFetchResult(result[0], result[1]) for result in results] def __search_in_history_query(self, search: str): escaped_items = ["'%{}%'".format(escape_query(s)) for s in search.split()] array_search_condition = "(array[{}])".format(", ".join(escaped_items)) return text( """ ( CASE WHEN "ProjectHistoryItem".action_id IN (1, 2, 3, 4, 12, 13) THEN creator.name ||' '|| campaign.name ||' '|| "ProjectHistoryItem".keywords ILIKE ALL {name} ELSE creator.name ||' '|| "ProjectHistoryItem".keywords ILIKE ALL {name} END ) """.format( name=array_search_condition ) ) def __action_type_query(self, query, action_type: Optional[int]) -> Query: if QueryActionTypes.ADDED.value == action_type: action_ids = [HistoryActionType.CHANGE_TERRITORIES.value, HistoryActionType.CHANGE_PROJECT_TEAM.value] return query.filter( or_( ProjectHistoryItem.action_id.in_( [ HistoryActionType.ADD_PRIMARY_ARTISTS_TO_PROJECT.value, HistoryActionType.ADD_FEATURED_ARTISTS_TO_PROJECT.value, HistoryActionType.ADD_PLAYLISTS_TO_PROJECT.value, ] ), and_( ProjectHistoryItem.action_id.in_(action_ids), func.json_array_length(ProjectHistoryItem.old_value) <= func.json_array_length(ProjectHistoryItem.new_value), ).self_group(), ) ) if QueryActionTypes.CREATED.value == action_type: action_ids = [ HistoryActionType.ADD_PROJECT.value, HistoryActionType.ADD_PHASE.value, HistoryActionType.ADD_CAMPAIGN.value, HistoryActionType.ADD_MEDIA_PLAN.value, ] return query.filter(ProjectHistoryItem.action_id.in_(action_ids)) if QueryActionTypes.EXTENDED.value == action_type: return query.filter( and_( ProjectHistoryItem.action_id == HistoryActionType.CHANGE_CAMPAIGN_PHASES.value, func.json_array_length(ProjectHistoryItem.old_value) < func.json_array_length(ProjectHistoryItem.new_value), ) ) if QueryActionTypes.MOVED.value == action_type: return query.filter( and_( ProjectHistoryItem.action_id == HistoryActionType.CHANGE_CAMPAIGN_PHASES.value, func.json_array_length(ProjectHistoryItem.old_value) == func.json_array_length(ProjectHistoryItem.new_value), ) ) if QueryActionTypes.REMOVED.value == action_type: return query.filter( or_( ProjectHistoryItem.action_id == HistoryActionType.REMOVE_PHASE.value, ProjectHistoryItem.action_id == HistoryActionType.REMOVE_CAMPAIGN.value, ProjectHistoryItem.action_id == HistoryActionType.REMOVE_PRIMARY_ARTISTS_FROM_PROJECT.value, ProjectHistoryItem.action_id == HistoryActionType.REMOVE_FEATURED_ARTISTS_FROM_PROJECT.value, ProjectHistoryItem.action_id == HistoryActionType.REMOVE_PLAYLISTS_FROM_PROJECT.value, ProjectHistoryItem.action_id == HistoryActionType.REMOVE_MEDIA_PLAN.value, and_( ProjectHistoryItem.action_id.in_([5, 11]), ( func.json_array_length(ProjectHistoryItem.old_value) > func.json_array_length(ProjectHistoryItem.new_value) ), ).self_group(), ) ) if QueryActionTypes.SHORTENED.value == action_type: return query.filter( and_( ProjectHistoryItem.action_id == HistoryActionType.CHANGE_CAMPAIGN_PHASES.value, func.json_array_length(ProjectHistoryItem.old_value) > func.json_array_length(ProjectHistoryItem.new_value), ) ) if QueryActionTypes.UPDATED.value == action_type: action_ids = [ HistoryActionType.CHANGE_CAMPAIGN_PLATFORM.value, HistoryActionType.UPDATE_CAMPAIGN_TIMELINE.value, HistoryActionType.UPDATE_CAMPAIGN_BUDGET.value, HistoryActionType.UPDATE_ARTIST.value, HistoryActionType.UPDATE_PROJECT_BUDGET.value, HistoryActionType.UPDATE_MEDIA_PLAN_NAME.value, HistoryActionType.UPDATE_MEDIA_PLAN_ORDER.value, HistoryActionType.UPDATE_PHASE_NAME.value, HistoryActionType.CHANGE_CAMPAIGN.value, ] return query.filter(ProjectHistoryItem.action_id.in_(action_ids)) if QueryActionTypes.REJECTED.value == action_type: return query.filter(ProjectHistoryItem.action_id == HistoryActionType.REJECT_DSP_CAMPAIGN.value) if QueryActionTypes.APPROVED.value == action_type: return query.filter(ProjectHistoryItem.action_id == HistoryActionType.APPROVE_DSP_CAMPAIGN.value) if QueryActionTypes.CLAIMED.value == action_type: return query.filter(ProjectHistoryItem.action_id == HistoryActionType.CLAIM_PROJECT_OWNERSHIP.value) return query.filter(ProjectHistoryItem.action_id.in_([item.value for item in ALLOWED_ACTION_TYPES]))