from typing import List, Optional from sqlalchemy import func, String, cast, case from sqlalchemy.dialects.postgresql import array from models.images import Image from models.artists import Artist from models.labels import Label from projects.services.project_history_service import ProjectHistoryService from projects.constants import MAX_PRIMARY_ARTISTS, MAX_FEATURED_ARTISTS from db import db from datetime import datetime, timezone from models.project_entity_type import ProjectEntityType, ProjectEntityAddType from models.raw_prs_budgets import RawPRSBudgets from models.raw_prs_projects import RawPRSProjects from models.raw_prs_purchase_orders import RawPRSPurchaseOrders from models.projects import Project, ProjectTargetItem from workers.prs.repository import PRSImportRepository from workers.base_worker import BaseWorker from config import ARTIST_IMAGE_SERVICE_URL class Queries: SUPPORTABLE_LABEL_IDS = [1, 44, 6, 3] CMG_LEGACY_COMPANY_CODES = ["2433", "2436", "2460", "2470", "2480", "2490", "2438", "2432", "2407", "2403"] CMG_LEGACY_LABEL_ID = 44 @staticmethod def projects_query(): po_info = ( db.session.query( RawPRSPurchaseOrders.sub_project_cd.label("sub_project_cd"), func.max(RawPRSPurchaseOrders.po_created_on).label("end_date"), func.min(RawPRSPurchaseOrders.po_created_on).label("min_po_created_date"), ) .group_by(RawPRSPurchaseOrders.sub_project_cd) ).cte("po_info") return ( db.session.query( RawPRSProjects.recording_project_no.label("gras_id"), RawPRSProjects.sub_project_cd.label("prs_id"), RawPRSProjects.sub_project_description.label("prs_title"), RawPRSProjects.recording_project_title_tx.label("gras_title"), func.array_agg( func.distinct( array( ( cast(RawPRSProjects.recording_project_artist_num, String), func.coalesce(RawPRSProjects.recording_project_artist_tx, "Unknown"), ) ) ) ).label("artists"), func.array_agg(func.distinct(RawPRSProjects.admin_group_cd)).label("labels"), func.min(func.to_date(RawPRSBudgets.created_at, "MM/DD/YYYY")).label("start_date"), po_info.c.end_date.label("end_date"), po_info.c.min_po_created_date.label("min_po_created_date"), case( [ ( func.trim(RawPRSProjects.company_code).in_(Queries.CMG_LEGACY_COMPANY_CODES), Queries.CMG_LEGACY_LABEL_ID, ) ], else_=Label.id, ).label("label_id"), ) .select_from(RawPRSProjects) .outerjoin(RawPRSBudgets, RawPRSProjects.budgets) .outerjoin(po_info, po_info.c.sub_project_cd == RawPRSProjects.sub_project_cd) .join(Label, Label.child_rep_owner_keys.any(RawPRSProjects.rep_owner_key)) .filter(Label.id.in_(Queries.SUPPORTABLE_LABEL_IDS)) .group_by( RawPRSProjects.recording_project_no, RawPRSProjects.sub_project_cd, Label.id, po_info.c.end_date, po_info.c.min_po_created_date ) ) @staticmethod def get_gras_only_project_by_external_id(project_id: int): return ( db.session.query(Project) .filter(Project.gras_project_code == str(project_id)) .filter(Project.prs_project_code.is_(None)) ) @staticmethod def get_gras_sample_project_by_external_id(project_id: int): return ( db.session.query(Project) .filter(Project.gras_project_code == str(project_id)) .filter(Project.prs_project_code.is_(None)) ) @staticmethod def get_prs_project_by_external_id(prs_project_id: str): return db.session.query(Project).filter(Project.prs_project_code == prs_project_id) class Utils: @staticmethod def project_title(artists: List[ProjectTargetItem]): return ", ".join([i.entity.name for i in artists if i.is_primary_artist()]) + " | PRS" @staticmethod def project_target_item(artist: Artist): project_artist = ProjectTargetItem(artist, ProjectEntityType.PRIMARY_ARTIST) project_artist.add_type = ProjectEntityAddType.LOCKED.value return project_artist @staticmethod def get_artists(raw_artists: List[List[str]]) -> List[Artist]: artists = [] for artist_obj in raw_artists: artist_id = f"GRAS_{artist_obj[0]}" if not artist_obj or artist_obj[0] is None: continue artist = db.session.query(Artist).filter(Artist.external_id == artist_id).one_or_none() if artist: artists.append(artist) else: artist_name = artist_obj[1] artist = Artist(external_id=artist_id, name=artist_name) artist.images = [ Image(url=ARTIST_IMAGE_SERVICE_URL.format(id=artist_obj[0])) ] db.session.add(artist) db.session.flush() artists.append(artist) return artists @staticmethod def resolve_artists_for_project(project: Project, prs_project): def __convert_to_featured(target_item): target_item.entity_type = ProjectEntityType.FEATURED_ARTIST.value return target_item artists = Utils.get_artists(prs_project.artists[:MAX_PRIMARY_ARTISTS]) # get PRS and GRAS artists from current project project_target_items = project.target_items non_prs_project_artists = list(filter(lambda x: not x.is_locked(), project_target_items)) prs_project_artists = list(filter(lambda x: x.is_locked(), project_target_items)) # get featured and primary artists for current project project_primary_artists = list(filter(lambda x: x.is_primary_artist(), non_prs_project_artists)) project_featured_artists = list(filter(lambda x: x.is_featured_artist(), non_prs_project_artists)) # filter primary artists to not include artists from PRS project_primary_artists = list(filter(lambda x: x.entity not in artists, project_primary_artists)) # get GRAS artists that identical to the PRS artists to_delete_primary_artists = list(filter(lambda x: x.entity in artists, project_primary_artists)) to_delete_featured_artists = list(filter(lambda x: x.entity in artists, project_featured_artists)) prs_target_items = [Utils.project_target_item(artist) for artist in artists] template_project_primary_artists = prs_target_items + project_primary_artists new_primary_artists = template_project_primary_artists[:MAX_PRIMARY_ARTISTS] converted_primary = list(map(__convert_to_featured, template_project_primary_artists[MAX_PRIMARY_ARTISTS:])) new_featured_artists = converted_primary + project_featured_artists # remove existing PRS artists cause we re-adding them [db.session.delete(item) for item in prs_project_artists] # remove artists from GRAS but identical to the PRS artists [db.session.delete(item) for item in to_delete_primary_artists] [db.session.delete(item) for item in to_delete_featured_artists] # remove extra featured artists if needed extra_items_start_index = MAX_FEATURED_ARTISTS - len(converted_primary) [db.session.delete(item) for item in new_featured_artists[extra_items_start_index:MAX_FEATURED_ARTISTS]] return new_primary_artists + new_featured_artists[:MAX_FEATURED_ARTISTS] class PRSProjectsWorker(BaseWorker): assigned_project_history_service = ProjectHistoryService() repository = PRSImportRepository() worker_name = "PRSProjectsWorker" def should_log_exceptions(self): return True def execute(self): projects = Queries.projects_query().all() prs_user = self.repository.get_prs_user() self.logger.info("PRS Projects import", f"Importing {len(projects)} projects...") for prs_project in projects: existing_prs_project = Queries.get_prs_project_by_external_id(prs_project.prs_id).one_or_none() if existing_prs_project: self.__update_prs_project(prs_project, existing_prs_project) else: new_project = self.__create_prs_project(prs_project) if not new_project: continue db.session.add(new_project) db.session.flush() Project.index_model(new_project) self.assigned_project_history_service.log_project_created(new_project, prs_user.id) db.session.flush() self.logger.success("PRS Projects import", "Finished!") db.session.commit() db.session.close() def __create_prs_project(self, prs_project) -> Optional[Project]: prs_user = self.repository.get_prs_user() artists = Utils.get_artists(prs_project.artists[:MAX_PRIMARY_ARTISTS]) if not artists: return None target_items = [Utils.project_target_item(artist) for artist in artists] current_timestamp = datetime.now(timezone.utc) external_project = Project() external_project.prs_title = prs_project.prs_title or Utils.project_title(target_items) external_project.prs_project_code = prs_project.prs_id external_project.gras_title = prs_project.gras_title or external_project.prs_title external_project.gras_project_code = str(prs_project.gras_id) external_project.target_items = target_items external_project.label_id = prs_project.label_id external_project.created_at = current_timestamp external_project.updated_at = current_timestamp external_project.last_edit_at = current_timestamp external_project.initial_start_date = self.__least_project_start_date( prs_project.start_date, prs_project.end_date, prs_project.min_po_created_date ) external_project.end_date = prs_project.end_date external_project.updated_by_import_at = current_timestamp external_project.create_user_id = prs_user.id external_project.edit_user_id = prs_user.id return external_project def __update_prs_project(self, prs_project, project: Project): prs_user = self.repository.get_prs_user() state = self.assigned_project_history_service.get_project_state(project) current_timestamp = datetime.now(timezone.utc) project.updated_by_import_at = current_timestamp project.prs_project_code = prs_project.prs_id project.prs_title = prs_project.prs_title project.label_id = prs_project.label_id if db.session.is_modified(project): project.updated_at = current_timestamp project.last_edit_at = current_timestamp if not project.is_claimed: project.initial_start_date = self.__least_project_start_date( prs_project.start_date, prs_project.end_date, prs_project.min_po_created_date ) project.end_date = prs_project.end_date project.target_items = Utils.resolve_artists_for_project(project, prs_project) self.assigned_project_history_service.log_project_changes(project, state, prs_user.id) Project.index_model(project) def __least_project_start_date(self, budgets_min_date, po_max_date, po_min_date): return min((date for date in [budgets_min_date, po_max_date, po_min_date] if date is not None), default=None)