from dataclasses import asdict from typing import List from flask import g from psycopg2.extras import NumericRange from sqlalchemy.orm.attributes import flag_modified from sqlalchemy import null as sqlalchemy_null import utils.session_utils as session_utils from campaigns.services.campaigns_history_service import CampaignHistoryService from external_api.utils import remove_records_with_values, remove_none_values from media_plan.repositories.media_plan_repository import MediaPlanRepository from campaigns.repositories.campaigns_repository import CampaignsRepository from artists.repositories.artist_repository import ArtistRepository from models.media_plan_phases import DEFAULT_PHASE_NAME from notifications.services.email_service import EmailService from services.users_repository import UsersRepository from media_plan.campaigns_repository import MediaPlanCampaignsRepository from media_plan.phases_repository import MediaPlanPhasesRepository from media_plan.schemas import ( MediaPlanCampaignPartialUpdateSchema, MediaPlanCampaignPartialUpdateResponseSchema, MediaPlanCampaignsBulkActionSchema, MediaPlanCampaignsBulkActionResponseSchema, MediaPlanCampaignsBulkActionCampaignResponseSchema, MediaPlanCampaignsBulkActionUpdateCampaignSchema, MediaPlanCampaignsBulkActionCreateCampaignSchema, ApproverUserResponseSchema, ApprovalRequestSchema, ApprovalRequestResponseSchema, SubmitApprovalSchema, MediaPlanReviewPhaseSchema, MediaPlanReviewResponseSchema, MediaPlanMetricsResponseSchema, MediaPlanResponseSchema, CreateUpdateMediaPlanRequest ) from campaigns.exceptions import CampaignNotFound from campaigns.schemas import CampaignNameGenerateSchema from models.media_plans import MediaPlan, MEDIA_PLANS_MAX_NUMBER, DEFAULT_MEDIA_PLAN_NAME from models.campaign import Campaign from projects.repositories.projects_repository import ProjectsRepository from services.permissions.exceptions import ProjectNotFound, UserNotFound, MediaPlanNotFound, MediaPlanMaxNumberExceed from services.territory.territories_repository import TerritoriesRepository from media_plan.validators import CampaignsValidator from handlers.enums import CampaignWorkflowStatus from media_plan.exceptions import CampaignsForReviewNotFound, NoApprovalRequestForCampaign, LastMediaPlanDeletionError from services.labels_repository import LabelsRepository from services.media_plan_service import MediaPlanService as media_plan_calculate_service from projects.schemas import TerritorySchema from campaigns.services.campaigns_name_generator_service import CampaignsNameGeneratorService from media_plan.phases_service import PhasesService from media_plan.services.media_plan_history_service import MediaPlanHistoryService, CampaignApprovalHistorySchema class MediaPlanService: campaign_history_service = CampaignHistoryService() media_plan_repository = MediaPlanRepository() campaign_name_generator_service = CampaignsNameGeneratorService() email_service = EmailService() campaigns_validator = CampaignsValidator() media_plan_campaigns_repository = MediaPlanCampaignsRepository() campaigns_repository = CampaignsRepository() territories_repository = TerritoriesRepository() projects_repository = ProjectsRepository() users_repository = UsersRepository() artists_repository = ArtistRepository() phases_repository = MediaPlanPhasesRepository() labels_repository = LabelsRepository() calculate_service = media_plan_calculate_service() phases_service = PhasesService() media_plan_history_service = MediaPlanHistoryService() def get_media_plan(self, media_plan_id): media_plan = self.media_plan_repository.get_media_plan_with_all_data_by_id(media_plan_id) if media_plan is None: raise MediaPlanNotFound() return media_plan def get_media_plans_for_project(self, project_id: int) -> List[MediaPlan]: project = self.projects_repository.get_project_by_id(project_id) if project is None: raise ProjectNotFound() return self.media_plan_repository.get_media_plans_by_project_id(project_id) def get_approver_users(self, media_plan_id: int): project = self.media_plan_repository.get_project_by_media_plan_id(media_plan_id) if project is None: raise ProjectNotFound() users = self.users_repository.get_users_for_approval( project_id=project.id, label_id=project.label_id ) return [ ApproverUserResponseSchema( id=user.id, name=user.name, email=user.email, hasProjectAccess=user.has_project_access ) for user in users ] def update_media_plan_campaign( self, campaign_uuid: str, campaign_data: MediaPlanCampaignPartialUpdateSchema, media_plan_id: int, user_id: int, ): project = self.media_plan_repository.get_project_by_media_plan_id(media_plan_id) campaign = self.campaigns_repository.get_campaign_by_uuid(campaign_uuid) campaign = self._set_campaign_columns(campaign, campaign_data, project.id, media_plan_id, user_id) session_utils.session_commit() return MediaPlanCampaignPartialUpdateResponseSchema(namingConvention=campaign.naming_convention) def bulk_action(self, media_plan_id: int, params: MediaPlanCampaignsBulkActionSchema, user_id: int): media_plan = self.media_plan_repository.get_media_plan_by_id(media_plan_id) for campaign_uuid in params.delete: self._delete_campaign(media_plan_id, campaign_uuid, user_id, media_plan.project_id) created = [] for campaign_data in params.create: created.append(self._create_campaign(media_plan.project_id, media_plan_id, campaign_data, user_id)) updated = [] for campaign_data in params.update: updated.append(self._update_campaign(media_plan.project_id, media_plan_id, campaign_data, user_id)) self.projects_repository.touch_project_last_edit(media_plan.project_id) session_utils.session_commit() return MediaPlanCampaignsBulkActionResponseSchema( created=[ MediaPlanCampaignsBulkActionCampaignResponseSchema( id=cmp.id, namingConvention=cmp.naming_convention, uuid=cmp.uuid, ) for cmp in created ], updated=[ MediaPlanCampaignsBulkActionCampaignResponseSchema( id=cmp.id, namingConvention=cmp.naming_convention, uuid=cmp.uuid, ) for cmp in updated ], ) def _set_campaign_style(self, campaign: Campaign, style): if style: style = asdict(style) if campaign.style is None: campaign.style = {} campaign.style.update(remove_records_with_values(style, ["missing"])) campaign.style = remove_none_values(campaign.style) if not campaign.style or not style: campaign.style = sqlalchemy_null() # an ordinary None is treated as null json and not null value column flag_modified(campaign, "style") # flag_modified use because issue with is_dirty of json field that not tracked in moment of first update def _delete_campaign(self, media_plan_id: int, campaign_uuid: str, user_id: int, project_id: int): campaign = self.campaigns_repository.get_campaign_by_uuid(campaign_uuid) self.campaigns_validator.validate_delete_campaign(campaign) territories_state = self.campaign_history_service.get_territories_state(project_id) self.campaign_history_service.log_campaign_deleted(campaign, territories_state, user_id) media_plan_phase_id = campaign.media_plan_phase_id order_in_phase = campaign.order_in_phase self.media_plan_campaigns_repository.delete_media_plan_campaign(campaign, media_plan_id, user_id, project_id) self.media_plan_campaigns_repository.shift_campaigns_order( media_plan_phase_id, order_in_phase, is_deletion=True ) session_utils.session_flush() def _create_campaign( self, project_id: int, media_plan_id: int, campaign_data: MediaPlanCampaignsBulkActionCreateCampaignSchema, user_id: int, ) -> Campaign: self.media_plan_campaigns_repository.shift_campaigns_order( campaign_data.phase_id, campaign_data.order_in_phase, is_deletion=False, ) session_utils.session_flush() campaign = Campaign() territories_state = self.campaign_history_service.get_territories_state(project_id) project = self.projects_repository.get_project_by_id(project_id) campaign.project = project campaign.media_plan_phase_id = campaign_data.phase_id campaign.order_in_phase = campaign_data.order_in_phase campaign.uuid = campaign_data.campaign_uuid if campaign_data.style != "missing": self._set_campaign_style(campaign, campaign_data.style) session_utils.session_flush() self.campaign_history_service.log_campaign_created(campaign=campaign, state=territories_state, user_id=user_id) return self._set_campaign_columns( campaign, campaign_data.columns, project_id, media_plan_id, user_id, ) def _update_campaign( self, project_id: int, media_plan_id: int, campaign_data: MediaPlanCampaignsBulkActionUpdateCampaignSchema, user_id: int, ) -> Campaign: campaign = self.campaigns_repository.get_campaign_by_uuid(campaign_data.campaign_uuid) if not campaign or campaign.is_deleted: raise CampaignNotFound() if campaign_data.order_in_phase: campaign.order_in_phase = campaign_data.order_in_phase if campaign_data.phase_id: campaign.media_plan_phase_id = campaign_data.phase_id if campaign_data.style != "missing": self._set_campaign_style(campaign, campaign_data.style) return self._set_campaign_columns( campaign, campaign_data.columns, project_id, media_plan_id, user_id, ) def _set_campaign_columns( self, campaign: Campaign, campaign_data: MediaPlanCampaignPartialUpdateSchema, project_id: int, media_plan_id: int, updated_by: int, ) -> Campaign: campaign_mask = self.campaign_history_service.get_campaign_state(campaign) re_index = False campaign_data_dict = { key: value for key, value in asdict(campaign_data).items() if value != "missing" } self._set_campaign_name(campaign_data, campaign_data_dict, campaign, media_plan_id) if "audience_lower_age" in campaign_data_dict or "audience_higher_age" in campaign_data_dict: if (campaign_data_dict.get("audience_lower_age") is None and campaign_data_dict.get("audience_higher_age") is None): campaign_data_dict["age_range"] = None else: fallback_upper_bound = ( campaign.age_range.upper - 1 # fe operates in [] while be operates in [) if campaign.age_range is not None and campaign.age_range.upper is not None else None ) fallback_lower_bound = ( campaign.age_range.lower if campaign.age_range is not None and campaign.age_range.lower is not None else None ) lower = campaign_data_dict.pop("audience_lower_age", fallback_lower_bound) upper = campaign_data_dict.pop("audience_higher_age", fallback_upper_bound) if lower is not None and upper is not None and lower > upper: lower, upper = upper, lower campaign_data_dict["age_range"] = NumericRange(lower, upper, "[]") campaign_platforms_ids = campaign_data_dict.pop('platforms', "missing") campaign_placements_ids = campaign_data_dict.pop('placements', "missing") campaign_genders_ids = campaign_data_dict.pop('genders', "missing") campaign_territories_ids = campaign_data_dict.pop('territories', "missing") campaign_ad_creative_links = campaign_data_dict.pop('ad_creative_links', "missing") campaign_destination_links = campaign_data_dict.pop('destination_links', "missing") workflow_status = campaign_data_dict.pop('workflow_status', "missing") if campaign_platforms_ids != "missing": campaign_data_dict["platforms"] = self.campaigns_repository.get_campaign_platforms_by_ids( campaign_platforms_ids) if campaign_placements_ids != "missing": campaign_data_dict["placements"] = self.campaigns_repository.get_campaign_placements_by_id( campaign_placements_ids) if campaign_genders_ids != "missing": campaign_data_dict["genders"] = self.campaigns_repository.get_campaign_genders_by_ids(campaign_genders_ids) searchable_keys = campaign.__searchable__.keys() for key, value in campaign_data_dict.items(): setattr(campaign, key, value) if key in searchable_keys: re_index = True if updated_by is not None: campaign.edit_user_id = updated_by campaign.project.edit_user_id = updated_by if campaign_territories_ids != "missing": self.campaigns_repository.update_campaign_territories( campaign, campaign_territories_ids, 0 # todo get rid of territory budget logic ) if campaign_ad_creative_links != "missing": self.campaigns_repository.update_campaign_creative_links(campaign, campaign_ad_creative_links or []) if campaign_destination_links != "missing": self.campaigns_repository.update_campaign_links(campaign, campaign_destination_links or []) if workflow_status != "missing": self.campaigns_repository.update_workflow_status(campaign, workflow_status) if re_index: Campaign.index_model(campaign) self.campaign_history_service.log_campaign_changes(campaign, campaign_mask, updated_by) session_utils.session_flush() return campaign def _set_campaign_name(self, campaign_data, campaign_data_dict, campaign, media_plan_id): # todo rethink name autogenerate logic, possibly change FE too system_fields = [ "workflow_status", "naming_convention" ] campaign_reset = all( [field is None for key, field in asdict(campaign_data).items() if key not in system_fields]) new_campaign = all( [field == "missing" for key, field in asdict(campaign_data).items() if key not in system_fields]) if not (campaign_data.workflow_status == CampaignWorkflowStatus.DRAFT.value and (campaign_reset or new_campaign) ): campaign_data_dict['naming_convention'] = self.campaign_name_generator_service.generate_campaign_name_str( media_plan_id=media_plan_id, params=CampaignNameGenerateSchema( startDate=campaign_data_dict.get('start_date', campaign.start_date), placements=campaign_data_dict.get( 'placements', [placement.id for placement in campaign.placements] ), provider=campaign_data_dict.get('provider_id', campaign.provider_id), objective=campaign_data_dict.get('objective_id', campaign.objective_id), territories=campaign_data_dict.get('territories', [territory.territory_id for territory in campaign.territories]), platforms=campaign_data_dict.get('platforms', [platform.id for platform in campaign.platforms]), campaignType=campaign_data_dict.get('type_id', campaign.type_id), lowerAge=campaign_data_dict.get( 'audience_lower_age', campaign.age_range.lower if campaign.age_range else None), higherAge=campaign_data_dict.get( 'audience_higher_age', campaign.age_range.upper if campaign.age_range else None), creativeDescription=campaign_data_dict.get('creative_description', campaign.creative_description), genders=campaign_data_dict.get('genders', campaign.genders), audienceNotes=campaign_data_dict.get('audience_notes', campaign.audience_notes) ), ) else: campaign_data_dict['naming_convention'] = None def create_approval_request(self, media_plan_id, params: ApprovalRequestSchema): media_plan = self.media_plan_repository.get_media_plan_by_id(media_plan_id) if media_plan is None: raise MediaPlanNotFound() project = self.projects_repository.get_project_by_id(media_plan.project_id) if project is None: raise ProjectNotFound() reviewer = self.users_repository.get_user_by_id(params.reviewers[0]) if reviewer is None: raise UserNotFound() artist = self.artists_repository.get_primary_artists_for_project_id(project.id) request = self.media_plan_campaigns_repository.create_approve_request( media_plan_id=media_plan_id, reviewer_id=reviewer.id, message=params.message ) session_utils.session_flush() self.media_plan_campaigns_repository.delete_active_approval_request_campaigns_by_uuid(params.campaigns) request_campaigns = self.media_plan_campaigns_repository.create_approve_request_campaigns( request_id=request.id, campaigns_uuids=params.campaigns ) self.media_plan_campaigns_repository.add_user_approval_role( project_id=project.id, approver_id=reviewer.id, ) self.campaigns_repository.change_workflow_status_by_campaings_uuid( params.campaigns, CampaignWorkflowStatus.PENDING_APPROVAL.value ) requester = self.users_repository.get_user_by_id(g.user_id) label = self.labels_repository.get_label_by_id(project.label_id) email_uuid = self.email_service.send_request_approval_email( requester=requester, reviewer=reviewer, project=project, media_plan_id=media_plan_id, artist=artist[0], label=label, message=params.message, ) request.email_uuid = email_uuid self.projects_repository.touch_project_last_edit(project.id) session_utils.session_commit() self.media_plan_history_service.log_approval_request(request, g.user_id) return ApprovalRequestResponseSchema(request) def get_campaigns_for_review(self, media_plan_id: int) -> List[MediaPlanReviewResponseSchema]: requester_ids = self.media_plan_campaigns_repository.get_requester_ids( media_plan_id=media_plan_id, reviewer_id=g.user_id ) requester_ids = [requester_id[0] for requester_id in requester_ids] response = [] for requester_id in requester_ids: res_phases = [] phases = self.phases_repository.get_phases_for_approval_requester_id( requester_id=requester_id, media_plan_id=media_plan_id, reviewer_id=g.user_id ) for phase in phases: campaigns = self.media_plan_campaigns_repository.get_campaigns_for_approval_requester_id( requester_id=requester_id, phase_id=phase.id, reviewer_id=g.user_id ) res_phases.append( MediaPlanReviewPhaseSchema(phase, campaigns) ) response.append( MediaPlanReviewResponseSchema( phases=res_phases, requester=self.users_repository.get_user_by_id(requester_id) ) ) if not requester_ids: raise CampaignsForReviewNotFound() return response def submit_approval_request(self, media_plan_id, params: SubmitApprovalSchema): project = self.media_plan_repository.get_project_by_media_plan_id(media_plan_id) if project is None: raise ProjectNotFound() artists = self.artists_repository.get_primary_artists_for_project_id(project.id) campaign_uuids = [campaign.campaign_uuid for campaign in params.campaigns] campaigns = { str(campaign.uuid): campaign for campaign in self.media_plan_campaigns_repository.get_campaigns_by_uuids_with_approval_data(campaign_uuids, project.id) } submit_approval_history_list = [] approval_request_campaigns_ids = [] for campaign_schema in params.campaigns: campaign = campaigns[campaign_schema.campaign_uuid] if campaign is None: raise CampaignNotFound() campaign_approval_request = campaign.current_approval_request_campaign if campaign_approval_request is None: raise NoApprovalRequestForCampaign() submit_approval_history_list.append( CampaignApprovalHistorySchema( campaign_uuid=campaign.uuid, campaign_approval_request_id=campaign_approval_request.id, workflow_status=campaign_schema.workflow_status ) ) self.campaigns_repository.update_workflow_status(campaign, campaign_schema.workflow_status) approval_request_campaigns_ids.append(campaign_approval_request.id) session_utils.session_flush() users = self.media_plan_campaigns_repository.get_approval_campaign_counts_by_user( media_plan_id, campaign_uuids, approval_request_campaigns_ids, ) self.projects_repository.touch_project_last_edit(project.id) session_utils.session_commit() self.email_service.send_submit_approval_emails(users, project, media_plan_id, artists[0]) self.media_plan_history_service.log_submit_approval( project.id, media_plan_id, submit_approval_history_list, g.user_id ) def get_media_plan_metrics(self, media_plan_id): internal_campaigns = self.campaigns_repository.get_manual_not_draft_campaigns_by_media_plan_id(media_plan_id) internal_campaign_ids = [internal_campaign.id for internal_campaign in internal_campaigns] territories = self.campaigns_repository.get_campaigns_territories(internal_campaign_ids) # budget_ecpms = self.campaigns_repository.get_campaign_budget_ecpms(project_id=project_id) # TODO: rewrite do the same as upper campaigns_platforms_data = self.campaigns_repository.get_campaigns_platforms_count(internal_campaign_ids) project_metrics = MediaPlanMetricsResponseSchema( campaignsCount=len(internal_campaign_ids), platformsCount=campaigns_platforms_data.platforms_count, territories=[TerritorySchema(territory) for territory in territories], estimatedImpressions=self.calculate_service.calculate_estimated_impressions(internal_campaigns), plannedBudget=self.calculate_service.calculate_budget(internal_campaigns) ) return project_metrics def get_last_media_plan_order_number(self, project_id: int) -> int: last_media_plan = self.media_plan_repository.get_last_media_plan(project_id) if last_media_plan is None: return 0 return last_media_plan.order + 1 def create_media_plan(self, project_id: int, params: CreateUpdateMediaPlanRequest): media_plans_count = self.media_plan_repository.media_plan_count(project_id) if media_plans_count >= MEDIA_PLANS_MAX_NUMBER: raise MediaPlanMaxNumberExceed() order = self.get_last_media_plan_order_number(project_id) media_plan = self.media_plan_repository.add_new_media_plan( project_id, name=params.name if params.name else DEFAULT_MEDIA_PLAN_NAME, order=order, create_user_id=g.user_id ) session_utils.session_flush() phase = self.phases_repository.add_new_phase( name=DEFAULT_PHASE_NAME, media_plan_id=media_plan.id, order=0, ) phase.campaigns = [ Campaign( project_id=project_id, order_in_phase=0, workflow_status=CampaignWorkflowStatus.DRAFT.value, create_user_id=g.user_id, )] self.projects_repository.touch_project_last_edit(project_id) session_utils.session_commit() self.media_plan_history_service.log_media_plan_created(media_plan, g.user_id) return MediaPlanResponseSchema(media_plan) def delete_media_plan(self, project_id: int, media_plan_id: int): media_plan = self.media_plan_repository.get_media_plan_by_id(media_plan_id) media_plans_count = self.media_plan_repository.media_plan_count(project_id) if media_plans_count == 1: raise LastMediaPlanDeletionError() self.media_plan_history_service.log_media_plan_deleted(media_plan, g.user_id) self.media_plan_repository.shift_media_plan_order(media_plan_id, media_plan.order, is_deletion=True) self.media_plan_repository.delete_media_campaigns(media_plan_id) self.media_plan_repository.delete_media_plan_phases(media_plan_id) self.media_plan_repository.delete_media_plan(media_plan_id) self.projects_repository.touch_project_last_edit(project_id) session_utils.session_commit() def update_media_plan(self, project_id: int, media_plan_id: int, params: CreateUpdateMediaPlanRequest): media_plan = self.get_media_plan(media_plan_id) media_plan_mask = self.media_plan_history_service.get_media_plan_state(media_plan) self.media_plan_repository.update_media_plan(media_plan, params.name) self.media_plan_history_service.log_media_plan_updated(media_plan, media_plan_mask, g.user_id) self.projects_repository.touch_project_last_edit(project_id) session_utils.session_commit() return MediaPlanResponseSchema(media_plan)