from app import db from datetime import datetime, timezone from typing import List, Optional from workers.base_worker import BaseWorker from campaigns.repositories.campaigns_repository import CampaignsRepository from projects.repositories.projects_repository import ProjectsRepository from purchase_orders.repositories.purchase_orders_repository import ProjectsRepository as PurchaseOrderRepository from models import ( Project, Campaign, ProjectCampaign, MarketingAccount, PRSPurchaseOrder, ProjectTargetItem, Artist, Image ) from models.projects import ProjectCampaignStatus from models.project_entity_type import ProjectEntityAddType, ProjectEntityType from utils.session_utils import session_flush, session_add, session_add_all, session_commit from purchase_orders.schemas import BlanketPurchaseOrdersListQueryParams from projects.services.project_history_service import ProjectHistoryService from campaigns.services.campaigns_history_service import CampaignHistoryService from constants.labels_constants import TEST_UK_LABEL_ID, TEST_US_LABEL_ID # Labels mapping, key is original label id, value is copy label id LABELS_IDS_MAPPING = { 47: TEST_UK_LABEL_ID, # UK label 6: TEST_US_LABEL_ID # US label } class LabelsCopyWorker(BaseWorker): projects_repository = ProjectsRepository() campaigns_repository = CampaignsRepository() purchase_orders_repository = PurchaseOrderRepository() project_history_service = ProjectHistoryService() campaign_history_service = CampaignHistoryService() def execute(self): self.logger.info("Labels data copy", "Started") for label_id, copy_label_id in LABELS_IDS_MAPPING.items(): self.process_data(label_id, copy_label_id) self.logger.info("Labels data copy", "Finished") def process_data(self, label_id: int, copy_label_id: int): last_worker_run_at = self.__get_last_worker_run_date(copy_label_id) original_projects = self.__get_all_projects_from_label_created_after_date(label_id, last_worker_run_at) for project in original_projects: copied_project = self.__create_copy_of_project_data(project, copy_label_id) assigned_campaigns = self.__get_assigned_campaigns_by_project_id(project.id) self.__create_copy_of_campaigns(assigned_campaigns, copied_project, copy_label_id) purchase_orders = self.__get_purchase_orders_by_project_id(project.id) self.__create_copy_of_purchase_orders(purchase_orders, copied_project) unassigned_campaigns = self.__get_unassigned_campaigns(label_id, last_worker_run_at) self.__create_copy_of_campaigns(unassigned_campaigns, None, copy_label_id) # This is one-time logic and it must be removed after running projects_with_test_artist = self.__get_projects_with_test_artist() for project in projects_with_test_artist: self.__set_original_artist(project) session_commit() def __get_all_projects_from_label_created_after_date(self, label_id: int, last_worker_run_at: Optional[str]): return self.projects_repository.get_projects_from_label_created_after_last_run_date( label_id, last_worker_run_at ) def __create_copy_of_project_data(self, original_project: Project, copy_label_id: int): copy_of_project = self.__create_project(original_project, copy_label_id) session_add(copy_of_project) session_flush() self.project_history_service.log_project_created(copy_of_project, copy_of_project.create_user_id) return copy_of_project def __create_project(self, project: Project, copy_label_id: int) -> Project: current_timestamp = datetime.now(timezone.utc) project_copy = Project() project_copy.name = project.name project_copy.label_id = copy_label_id project_copy.initial_start_date = project.initial_start_date project_copy.end_date = project.end_date project_copy.created_at = current_timestamp project_copy.updated_at = current_timestamp project_copy.last_edit_at = current_timestamp project_copy.updated_by_import_at = project.updated_by_import_at project_copy.allocation = 1000.00 # censor project allocation project_copy.budget = 1000.00 # censor project budget project_copy.is_confidential = project.is_confidential project_copy.gras_project_code = "test" + project.gras_project_code project_copy.ccp_title = project.ccp_title project_copy.ccp_project_code = "test" + project.ccp_project_code if project.ccp_project_code else None project_copy.prs_project_code = "test" + project.prs_project_code if project.prs_project_code else None project_copy.prs_title = project.prs_title project_copy.gras_title = project.gras_title project_copy.create_user_id = project.create_user_id project_copy.edit_user_id = project.edit_user_id project_copy.source = project.source project_copy.currency = project.currency project_copy.notes = project.notes project_copy.is_claimed = False artist = self.__get_artist(project.target_items[0].entity_id) project_copy.target_items = [self.__create_project_target_item(artist, project.target_items[0])] return project_copy def __get_assigned_campaigns_by_project_id(self, project_id: int): return self.campaigns_repository.get_digital_campaigns_asigned_to_project(project_id) def __create_copy_of_campaigns( self, original_campaigns, copied_project: Optional[Project], copy_label_id: int ): for campaign in original_campaigns: campaign_copy = self.__create_campaign(campaign, copied_project, copy_label_id) if copied_project: project_campaign = ProjectCampaign() project_campaign.project = copied_project project_campaign.status = ProjectCampaignStatus.APPROVED.value project_campaign.campaign = campaign_copy copied_project.project_campaigns.append(project_campaign) if copied_project: state = self.campaign_history_service.get_territories_state(copied_project.id) self.campaign_history_service.log_campaign_created(campaign_copy, state, campaign.create_user_id) def __get_purchase_orders_by_project_id(self, project_id: int): params = BlanketPurchaseOrdersListQueryParams(provider=None, status=None, search=None) purchase_orders_data, _ = self.purchase_orders_repository.get_purchase_orders_list(params, project_id) return purchase_orders_data def __create_campaign(self, campaign: Campaign, copied_project: Project, copy_label_id: int): marketing_account = self.__get_or_create_marketing_account( campaign.marketing_account, copy_label_id ) campaign_copy = Campaign() campaign_copy.project = copied_project campaign_copy.name = campaign.name campaign_copy.start_date = campaign.start_date campaign_copy.end_date = campaign.end_date campaign_copy.planned_budget = 1000.00 # censor campaign planned_budget campaign_copy.budget_spend = 1000.00 # censor campaign budget_spend campaign_copy.marketing_account = marketing_account campaign_copy.objective_id = campaign.objective_id campaign_copy.provider_id = campaign.provider_id campaign_copy.age_range = campaign.age_range campaign_copy.genders = campaign.genders campaign_copy.source = campaign.source campaign_copy.metainfo = campaign.metainfo campaign_copy.platforms = campaign.platforms campaign_copy.gratis = campaign.gratis campaign_copy.type = campaign.type campaign_copy.workflow_status = campaign.workflow_status session_add(campaign_copy) session_flush() territories = [territory.territory_id for territory in campaign.territories] self.campaigns_repository.update_campaign_territories( campaign_copy, territories, campaign.planned_budget ) return campaign_copy def __create_purchase_order( self, original_purchase_order: PRSPurchaseOrder, copied_project: Project, parent_po_num: str = None ) -> PRSPurchaseOrder: current_timestamp = datetime.now(timezone.utc) purchase_order = PRSPurchaseOrder() purchase_order.name = original_purchase_order.name purchase_order.po_number = "test" + original_purchase_order.po_number purchase_order.po_line_item = original_purchase_order.po_line_item purchase_order.project_id = copied_project.id purchase_order.group_id = original_purchase_order.group_id purchase_order.category_id = original_purchase_order.category_id purchase_order.status = original_purchase_order.status purchase_order.total_amount = 1000.00 # censor purchase order total_amount purchase_order.paid_amount = 1000.00 # censor purchase order paid_amount purchase_order.created_date = current_timestamp purchase_order.last_modified_date = current_timestamp purchase_order.provider_id = original_purchase_order.provider_id purchase_order.is_under_uncommited_blanket = original_purchase_order.is_under_uncommited_blanket purchase_order.total_amount_currency = original_purchase_order.total_amount_currency purchase_order.paid_amount_currency = original_purchase_order.paid_amount_currency purchase_order.parent_po_number = parent_po_num return purchase_order def __create_copy_of_purchase_orders(self, purchase_orders: List[PRSPurchaseOrder], copied_project: Project): purchase_orders_copies = [] for purchase_order in purchase_orders: purchase_order_copy = self.__create_purchase_order(purchase_order, copied_project) purchase_orders_copies.append(purchase_order_copy) for child in purchase_order.child_purchase_orders: child_copy = self.__create_purchase_order(child, copied_project, purchase_order_copy.po_number) purchase_orders_copies.append(child_copy) session_add_all(purchase_orders_copies) session_flush() def __get_or_create_marketing_account( self, original_marketing_account: MarketingAccount, copy_label_id: int ) -> MarketingAccount: marketing_account = self.campaigns_repository.get_marketing_account_by_marketing_account( original_marketing_account, copy_label_id ) if not marketing_account: marketing_account = MarketingAccount() marketing_account.label_id = copy_label_id marketing_account.name = original_marketing_account.name marketing_account.external_id = "test" + original_marketing_account.external_id marketing_account.provider_id = original_marketing_account.provider_id marketing_account.source = original_marketing_account.source marketing_account.currency = original_marketing_account.currency marketing_account.type = original_marketing_account.type session_add(marketing_account) session_flush() return marketing_account def __get_artist(self, artist_id: str): artist = db.session.query(Artist).filter(Artist.id == artist_id).one_or_none() return artist def __create_project_target_item(self, artist: Artist, project_target_item: ProjectTargetItem): project_target_item_copy = ProjectTargetItem( artist, ProjectEntityType(project_target_item.entity_type), ProjectEntityAddType(project_target_item.add_type), ) return project_target_item_copy def __get_unassigned_campaigns(self, label_id: int, last_worker_run_at: Optional[str]): return self.campaigns_repository.get_unassigned_campaigns_after_last_run_date_by_label_id( label_id, last_worker_run_at ) def __get_last_worker_run_date(self, label_id: int): return self.projects_repository.get_max_created_at_date_by_label_id(label_id) def __get_projects_with_test_artist(self): projects = ( db.session.query(Project) .join(ProjectTargetItem, ProjectTargetItem.project_id == Project.id) .join(Artist, ProjectTargetItem.entity_id == Artist.id) .filter(Artist.external_id == "GRAS_9999991", Project.label_id.in_(LABELS_IDS_MAPPING.values())) .all() ) return projects def __get_original_project_by_copied_project(self, copied_project): query = db.session.query(Project) if copied_project.ccp_project_code: query = query.filter( Project.ccp_project_code == self.__prepare_project_code(copied_project.ccp_project_code) ) else: query = query.filter( Project.prs_project_code == self.__prepare_project_code(copied_project.prs_project_code) ) return query.first() def __prepare_project_code(self, project_code: str): return project_code.replace('test', '') def __set_original_artist(self, project): original_project = self.__get_original_project_by_copied_project(project) artist = self.__get_artist(original_project.target_items[0].entity_id) project.target_items = [self.__create_project_target_item(artist, original_project.target_items[0])] session_flush()