from typing import List, Optional from datetime import datetime, timezone from campaigns.services.campaign_platforms_mapper import CampaignPlatformMapper from db import db from flask import g from psycopg2.extras import NumericRange import config from campaigns.exceptions import NotEditableField from utils.json_encoder import DataclassDictionaryEncoder as Encoder from campaigns.services.campaigns_history_service import CampaignHistoryService from campaigns.services.campaigns_name_generator_service import CampaignsNameGeneratorService from services.project_history.history_service import HistoryActionType from campaigns.schemas import ( CampaignResponseSchema, CampaignCreateSchema, CampaignDetailsResponseSchema, CampaignListQueryParams, CampaignDetailsProviderResponseSchema, CampaignDetailsGenderResponseSchema, CampaignTypesResponseModel, CampaignMetadata, CampaignEditableFields, CampaignScheduleUpdateSchema, CampaignTabSchema, CampaignStatusSchema, CreatePlacementSchema, PlacementResponseModel, CampaignPlatformsNamesList, CampaignCategoryResponseModel, CampaignTypeResponseModel, CampaignPlacementResponseModel, CampaignTypesRequestSchema, ProjectCampaignPlatformSchema, CampaignNameGenerateSchema, ) from linkfire.schemas import LinkfireLinkSchema from campaigns.repositories.campaigns_repository import CampaignsRepository from campaigns.repositories.campaigns_list_repository import CampaignsListRepository from linkfire.linkfire_repository import LinkfireRepository from projects.repositories.projects_repository import ProjectsRepository from utils.exceptions import BadRequest, Forbidden, NotFound from models.campaign import Campaign, CampaignLink, CampaignPlacements, CampaignCreativeLink from campaigns.services.validators import CampaignsValidator from flask_fsm import state_map class CampaignsService: campaigns_repository = CampaignsRepository() campaign_history_service = CampaignHistoryService() project_repository = ProjectsRepository() campaigns_list_repository = CampaignsListRepository() linkfire_repository = LinkfireRepository() campaigns_validator = CampaignsValidator() campaign_name_generator_service = CampaignsNameGeneratorService() campaign_platforms_mapper = CampaignPlatformMapper() def get_campaign_detail(self, campaign_uuid: str, project_id: int) -> Optional[CampaignDetailsResponseSchema]: campaign = self.campaigns_repository.get_campaign_details(project_id, campaign_uuid) if not campaign: return None return self.__get_campaign_response(campaign) def get_unassigned_project_campaign_detail( self, campaign_uuid: str, project_id: int ) -> Optional[CampaignDetailsResponseSchema]: # Check that campaign is pending or approved and belongs to the project campaign_exists = self.campaigns_repository.get_project_campaign(project_id, campaign_uuid) is not None if not campaign_exists: return None campaign = self.campaigns_repository.get_campaign_details_by_uuid(campaign_uuid) if not campaign: return None return self.__get_campaign_response(campaign) def get_campaigns_by_project( self, project_id: int, params: CampaignListQueryParams ) -> List[CampaignResponseSchema]: if not params.isDigital: campaigns = self.campaigns_list_repository.get_filtered_campaigns_by_project(project_id, g.user_id, params) else: campaigns = self.campaigns_list_repository.get_filtered_digital_campaigns_by_project( project_id, g.user_id, params ) return self.__get_campaigns_response(campaigns, params.isDigital) def __get_campaigns_response(self, campaigns, is_digital): result = [] for campaign in campaigns: is_pending = False if is_digital: is_pending = campaign.is_pending campaign = campaign[0] currency = self.campaigns_repository.get_currency(campaign.uuid) result.append(CampaignResponseSchema(campaign, is_pending=is_pending, currency=currency)) return result def __get_campaign_response(self, campaign: Campaign) -> CampaignDetailsResponseSchema: is_pending_campaign = self.campaigns_repository.is_pending_campaign(campaign.id) currency = self.campaigns_repository.get_currency(campaign.uuid) return CampaignDetailsResponseSchema( campaign=campaign, phases=[], is_pending=is_pending_campaign, currency=currency ) """ Deprecated, needs to be rewritten to accommodate new MediaPlanPhase logic """ def create_campaign( self, project_id: int, created_by: int, campaign_data: CampaignCreateSchema ) -> CampaignDetailsResponseSchema: """Create new campaign. Creating a new campaign is considered as project update, that is why this function also updates project's updated_at and edit_user_id fields. Args: project_id (int): Project ID. campaign_data (CampaignCreateUpdateSchema): All campaign data created_by (int): ID of the user, who created the campaign. Returns: Campaign: New campaign instance. Raises: PhaseStartDateError: If the dates provided are invalid. """ # check if entities passed in types exist project = self.project_repository.get_project_by_id(project_id) self.campaigns_validator.validate_campaign_start_date(project_id=project_id, start_date=campaign_data.startDate) self.campaigns_validator.validate_campaign_end_date(project_id=project_id, end_date=campaign_data.endDate) campaign_platforms = self.campaigns_repository.get_campaign_platforms_by_ids(campaign_data.platforms) self.campaigns_validator.validate_campaign_platforms(campaign_platforms) placements = self.campaigns_repository.get_campaign_placements_by_id(campaign_data.placements) self.campaigns_validator.validate_campaign_placements(project.label_id, placements) new_campaign = Campaign() new_campaign.project_id = project_id new_campaign.name = campaign_data.name new_campaign.naming_convention = campaign_data.namingConvention new_campaign.start_date = campaign_data.startDate new_campaign.end_date = campaign_data.endDate new_campaign.planned_budget = campaign_data.plannedBudget new_campaign.budget_spend = campaign_data.budgetSpend new_campaign.create_user_id = created_by new_campaign.edit_user_id = created_by new_campaign.objective_id = campaign_data.objective new_campaign.notes = campaign_data.notes new_campaign.ecpm = campaign_data.ecpm new_campaign.provider_id = campaign_data.provider new_campaign.audience_notes = campaign_data.audienceNotes new_campaign.ad_creative_notes = campaign_data.adCreativeNotes new_campaign.kpi_metrics_field_id = campaign_data.kpiMetricsFieldId new_campaign.workflow_status = campaign_data.workflowStatus new_campaign.age_range = self.__age_range_from_params(campaign_data) new_campaign.genders = self.campaigns_repository.get_campaign_genders_by_ids(campaign_data.genders) new_campaign.links = [CampaignLink(link=link) for link in campaign_data.destinationLinks or []] new_campaign.metainfo = Encoder.convert_to_dict(CampaignMetadata(fields=CampaignEditableFields())) new_campaign.type_id = campaign_data.campaignType new_campaign.gratis = campaign_data.gratis new_campaign.ad_creative_links = [CampaignCreativeLink(url=url) for url in campaign_data.adCreativeLinks] new_campaign.placements = placements new_campaign.platforms = campaign_platforms territories_state = self.campaign_history_service.get_territories_state(project_id) self.campaigns_repository.update_campaign_territories( new_campaign, campaign_data.territories, int(campaign_data.plannedBudget) ) self.linkfire_repository.update_campaign_linkfire_links( new_campaign, [ LinkfireLinkSchema(id=link.id, link_id=link.linkId, link_url=link.linkUrl, deletable=link.deletable) for link in campaign_data.linkfireLinks ], ) self.campaigns_repository.save_campaign_changes(new_campaign, created_by) new_campaign.naming_convention = self.campaign_name_generator_service.generate_campaign_name_str( project_id, params=CampaignNameGenerateSchema( startDate=campaign_data.startDate, placements=campaign_data.placements, provider=campaign_data.provider, objective=campaign_data.objective, territories=campaign_data.territories, platforms=campaign_data.platforms, campaignType=campaign_data.campaignType, audienceNotes=campaign_data.audienceNotes, lowerAge=campaign_data.audienceLowerAge, higherAge=campaign_data.audienceHigherAge, creativeDescription=None, genders=campaign_data.genders ), ) self.campaign_history_service.log_campaign_created(new_campaign, territories_state, g.user_id) currency = self.project_repository.get_project_currency(project_id) return CampaignDetailsResponseSchema( campaign=new_campaign, phases=[], # TODO: upd schema to not include phases is_pending=False, currency=currency ) """ Deprecated, needs to be rewritten to accommodate new MediaPlanPhase logic """ def update_campaign( self, campaign_id: int, updated_by: int, campaign_data: CampaignCreateSchema, project_id: int ) -> CampaignDetailsResponseSchema: """Update campaign. Args: campaign_id: int Id of the campaign updated updated_by: int id of user updating the campaign campaign_data (CampaignCreateSchema): Campaign data that must be updated. project_id: int id of the project the campaign belongs to Returns: Campaign: Updated campaign instance. Raises: CampaignTypeError: If provided campaign type hierarchy is invalid. """ campaign = self.campaigns_repository.get_campaign_by_id(campaign_id) self.campaigns_validator.validate_campaign_start_date(campaign.project_id, campaign_data.startDate) self.campaigns_validator.validate_campaign_end_date(campaign.project_id, campaign_data.endDate) validation_errors = self.campaigns_validator.validate_campaign_edit(campaign, campaign_data) if validation_errors: raise NotEditableField(detail="\n".join(validation_errors)) campaign_mask = self.campaign_history_service.get_campaign_state(campaign) campaign_platforms = self.campaigns_repository.get_campaign_platforms_by_ids(campaign_data.platforms) self.campaigns_validator.validate_campaign_platforms(campaign_platforms) placements = self.campaigns_repository.get_campaign_placements_by_id(campaign_data.placements) self.campaigns_validator.validate_campaign_placements(campaign.project.label_id, placements) campaign.naming_convention = self.campaign_name_generator_service.generate_campaign_name_str( project_id=project_id, params=CampaignNameGenerateSchema( startDate=campaign_data.startDate, placements=campaign_data.placements, provider=campaign_data.provider, objective=campaign_data.objective, territories=campaign_data.territories, platforms=campaign_data.platforms, campaignType=campaign_data.campaignType, audienceNotes=campaign_data.audienceNotes, lowerAge=campaign_data.audienceLowerAge, higherAge=campaign_data.audienceHigherAge, creativeDescription=None, genders=campaign_data.genders ), ) campaign.name = campaign_data.name campaign.naming_convention = campaign_data.namingConvention campaign.start_date = campaign_data.startDate campaign.end_date = campaign_data.endDate campaign.planned_budget = campaign_data.plannedBudget campaign.budget_spend = campaign_data.budgetSpend campaign.objective_id = campaign_data.objective campaign.notes = campaign_data.notes campaign.provider_id = campaign_data.provider campaign.audience_notes = campaign_data.audienceNotes campaign.ad_creative_notes = campaign_data.adCreativeNotes campaign.kpi_metrics_field_id = campaign_data.kpiMetricsFieldId campaign.workflow_status = campaign_data.workflowStatus campaign.ecpm = campaign_data.ecpm campaign.age_range = self.__age_range_from_params(campaign_data) campaign.genders = self.campaigns_repository.get_campaign_genders_by_ids(campaign_data.genders) campaign.type_id = campaign_data.campaignType campaign.gratis = campaign_data.gratis campaign.platforms = campaign_platforms campaign.placements = placements campaign.touch() if updated_by is not None: campaign.edit_user_id = updated_by campaign.project.edit_user_id = updated_by self.linkfire_repository.update_campaign_linkfire_links( campaign, [ LinkfireLinkSchema(id=link.id, link_id=link.linkId, link_url=link.linkUrl, deletable=link.deletable) for link in campaign_data.linkfireLinks ], ) self.campaigns_repository.update_campaign_territories( campaign, campaign_data.territories, int(campaign_data.plannedBudget) ) self.campaigns_repository.update_campaign_creative_links(campaign, campaign_data.adCreativeLinks or []) self.campaigns_repository.update_campaign_links(campaign, campaign_data.destinationLinks or []) self.campaigns_repository.save_campaign_changes(campaign) self.campaign_history_service.log_campaign_changes(campaign, campaign_mask, g.user_id) return self.__get_campaign_response(campaign) def update_campaign_schedule( self, campaign_uuid: str, updated_by: int, params: CampaignScheduleUpdateSchema ) -> CampaignDetailsResponseSchema: campaign = self.campaigns_repository.get_campaign_by_uuid(campaign_uuid) self.campaigns_validator.validate_campaign_start_date(campaign.project_id, params.startDate) validation_errors = self.campaigns_validator.validate_campaign_schedule_update( campaign.id, params.startDate, params.endDate ) if validation_errors: raise NotEditableField(detail="\n".join(validation_errors)) campaign_mask = self.campaign_history_service.get_campaign_state(campaign) campaign.start_date = params.startDate campaign.end_date = params.endDate campaign.touch() if updated_by is not None: campaign.edit_user_id = updated_by campaign.project.edit_user_id = updated_by self.campaigns_repository.save_campaign_changes(campaign, updated_by) self.campaign_history_service.log_campaign_changes(campaign, campaign_mask, updated_by) return self.__get_campaign_response(campaign) def delete_campaign(self, campaign_uuid: str, project_id: int, user_id: int): campaign = self.campaigns_repository.get_campaign_by_uuid(campaign_uuid) self.campaigns_validator.validate_manual_campaign(campaign) territories_state = self.campaign_history_service.get_territories_state(project_id) self.campaigns_repository.delete_campaign(campaign, project_id, user_id) self.campaign_history_service.log_campaign_deleted(campaign, territories_state, g.user_id) def undo_delete_campaign(self, campaign_uuid: str, user_id: int): campaign = self.campaigns_repository.get_campaign_by_uuid(campaign_uuid) if not campaign: raise NotFound("Campaign not found.") if not campaign.is_deleted: return history_item = self.campaigns_repository.get_history_item_for_campaign( campaign_id=campaign.id, action_id=HistoryActionType.REMOVE_CAMPAIGN.value ) if history_item.user_id != user_id: raise Forbidden("You don't have permission to revert the action.") if (datetime.now(timezone.utc) - history_item.created_at).total_seconds() > config.UNDO_GRACE_PERIOD: raise BadRequest("Deletion cannot be reverted.") self.campaigns_repository.undo_delete_campaign(campaign=campaign, history_item=history_item) def get_campaign_types(self, params: CampaignTypesRequestSchema) -> CampaignTypesResponseModel: groups = self.campaigns_repository.get_campaign_groups() platforms = self.campaigns_repository.get_campaign_platforms_for_labels(params.labelIds) placements = self.campaigns_repository.get_available_placements_for_labels(params.labelIds) return CampaignTypesResponseModel( categories=[ CampaignCategoryResponseModel( id=group.id, name=group.name, campaignTypes=[ CampaignTypeResponseModel(id=campaign_type.id, name=campaign_type.name) for campaign_type in group.campaign_types ], ) for group in groups ], platforms=self.campaign_platforms_mapper.build(platforms), placements=[ CampaignPlacementResponseModel(id=placement.id, name=placement.name, labelId=placement.label_id) for placement in placements ], ) def get_campaign_providers(self, only_existing: bool) -> List[CampaignDetailsProviderResponseSchema]: providers = self.campaigns_repository.get_campaign_providers(only_existing) return [CampaignDetailsProviderResponseSchema(provider.id, provider.name) for provider in providers] def get_campaign_providers_for_campaign(self, campaign_uuid: str): providers = self.campaigns_repository.get_campaign_providers_for_campaign(campaign_uuid) return [CampaignDetailsProviderResponseSchema(provider.id, provider.name) for provider in providers] def get_campaign_genders(self) -> List[CampaignDetailsGenderResponseSchema]: genders = self.campaigns_repository.get_campaign_genders() return [CampaignDetailsGenderResponseSchema(gender.id, gender.name) for gender in genders] def get_available_platforms( self, project_id: int, params: ProjectCampaignPlatformSchema ) -> CampaignPlatformsNamesList: platforms = self.campaigns_repository.get_available_platforms_for_project( project_id, with_pending=params.withPending, ) return CampaignPlatformsNamesList(items=[platform[0] for platform in platforms]) def get_available_dsp_platforms(self, project_id: int) -> CampaignPlatformsNamesList: platforms = self.campaigns_repository.get_available_platforms_for_project( project_id, with_pending=False, only_external=True ) return CampaignPlatformsNamesList(items=[platform[0] for platform in platforms]) def get_all_statuses(self) -> List[CampaignStatusSchema]: campaign_statuses = [] for status in state_map.get_all_states(): campaign_statuses.append( CampaignStatusSchema( status=status, available_transitions=state_map.get_target(status) ) ) return campaign_statuses def __age_range_from_params(self, params: CampaignCreateSchema): return NumericRange(params.audienceLowerAge, params.audienceHigherAge, "[]") def map_campaigns(self, campaigns, currency): items = [] for res in campaigns: items.append(CampaignTabSchema(campaign=res[0], is_pending=res.is_pending, currency=currency)) return items def create_placement(self, label_id: int, params: CreatePlacementSchema) -> PlacementResponseModel: self.campaigns_validator.validate_create_placement(params, label_id) new_placement = CampaignPlacements(name=params.name, label_id=label_id) db.session.add(new_placement) db.session.commit() return PlacementResponseModel(id=new_placement.id, name=new_placement.name, label_id=new_placement.label_id)