from datetime import date from copy import deepcopy from dataclasses import dataclass from typing import List, Optional from utils.json_encoder import DataclassDictionaryEncoder as Encoder from utils import list_utils from services.project_history.history_service import HistoryService from utils.date_utils import DateRange from models.campaign import Campaign from models.projects import Project from campaigns.repositories.campaigns_repository import CampaignsRepository from services.project_history.schemas import ( HistoryTerritory, HistoryObjective, HistoryCampaignType, HistoryCampaignChanged ) from shared.schemas import CampaignLinkfireLinkListMapper, LinkfireLinkResponseSchema from campaigns.schemas import CampaignPlacementSchema @dataclass class CampaignHistoryState: name: str workflow_status: str territories: List[HistoryTerritory] budget: float schedule: DateRange objective: HistoryObjective linkfire_links: List[LinkfireLinkResponseSchema] platforms: List[str] type: Optional[HistoryCampaignType] gratis: Optional[bool] notes: Optional[str] naming_convention: Optional[str] naming_convention_manual: Optional[str] start_date: Optional[date] end_date: Optional[date] creative_description: Optional[str] ecpm: Optional[float] audience_notes: Optional[str] ad_creative_link: Optional[str] ad_creative_notes: Optional[str] headline: Optional[str] call_to_action: Optional[str] order_in_phase: Optional[int] placements: Optional[CampaignPlacementSchema] kpi_metrics_field_id: Optional[int] audience_lower_age: Optional[int] audience_higher_age: Optional[int] genders: Optional[List[int]] ad_creative_links: Optional[List[str]] links: Optional[List[str]] def __init__(self, campaign): self.name = campaign.name self.territories = self.get_territories(campaign) self.budget = campaign.planned_budget self.schedule = DateRange(campaign.start_date, campaign.fallback_end_date) self.objective = HistoryObjective(campaign) if campaign.objective else None self.notes = campaign.notes self.linkfire_links = CampaignLinkfireLinkListMapper.map(campaign.linkfire_campaigns) self.platforms = deepcopy(campaign.platform_value) self.type = HistoryCampaignType(campaign) if campaign.type else None self.gratis = campaign.gratis self.naming_convention = campaign.naming_convention self.naming_convention_manual = campaign.naming_convention_manual self.start_date = campaign.start_date self.end_date = campaign.end_date self.creative_description = campaign.creative_description self.ecpm = campaign.ecpm self.audience_notes = campaign.audience_notes self.ad_creative_link = campaign.ad_creative_link self.ad_creative_notes = campaign.ad_creative_notes self.order_in_phase = campaign.order_in_phase self.workflow_status = campaign.workflow_status self.headline = campaign.headline self.call_to_action = campaign.call_to_action self.placements = [ CampaignPlacementSchema(id=p.id, name=p.name, labelId=p.label_id) for p in campaign.placements ] self.kpi_metrics_field_id = campaign.kpi_metrics_field_id self.audience_lower_age = getattr(campaign.age_range, "lower", None) self.audience_higher_age = getattr(campaign.age_range, "upper", None) self.genders = [g.id for g in campaign.genders] self.ad_creative_links = [link.id for link in campaign.ad_creative_links] if campaign.ad_creative_links else [] self.links = [link.id for link in campaign.links] if campaign.links else [] @staticmethod def get_territories(campaign: Campaign): campaigns_repository = CampaignsRepository() return [ HistoryTerritory(model) for model in campaigns_repository.get_campaign_project_territories(campaign.project_id) ] @staticmethod def exclude_fields(): return ["schedule", "platforms", "linkfire_links", "territories", "type", "objective", "budget", "placements", "audience_lower_age", "audience_higher_age", "genders", "ad_creative_links", "links"] @dataclass class CampaignTerritoriesHistoryState: territories: List[HistoryTerritory] def __init__(self, territories: List[HistoryTerritory]): self.territories = territories class CampaignHistoryService: campaigns_repository = CampaignsRepository() history_service = HistoryService() @staticmethod def get_campaign_state(campaign: Campaign) -> CampaignHistoryState: return CampaignHistoryState(campaign) def get_territories_state(self, project_id: Optional[int]) -> CampaignTerritoriesHistoryState: if not project_id: return CampaignTerritoriesHistoryState([]) repository = self.campaigns_repository territories = [HistoryTerritory(model) for model in repository.get_campaign_project_territories(project_id)] return CampaignTerritoriesHistoryState(territories) def log_campaign_changes(self, campaign: Campaign, campaign_mask: CampaignHistoryState, user_id: Optional[int]): self.__log_campaign_changed_if_needed(campaign=campaign, campaign_mask=campaign_mask, user_id=user_id) if campaign.project_id: new_territories = [ HistoryTerritory(model) for model in self.campaigns_repository.get_campaign_project_territories(campaign.project_id) ] self.__log_project_territories_changed_if_needed( campaign, campaign.project_id, campaign_mask.territories, new_territories, user_id ) def log_campaign_created( self, campaign: Campaign, state: Optional[CampaignTerritoriesHistoryState], user_id: Optional[int] ): self.history_service.log_campaign_created(campaign, user_id) project_id = campaign.project_id if project_id: new_territories = [ HistoryTerritory(model) for model in self.campaigns_repository.get_campaign_project_territories(project_id) ] self.__log_project_territories_changed_if_needed( campaign, campaign.project_id, state.territories, new_territories, user_id ) self.__log_campaign_linkfire_links_changed_if_needed(campaign, [], campaign.linkfire_campaigns, user_id) def log_campaign_deleted( self, campaign: Campaign, state: CampaignTerritoriesHistoryState, user_id: Optional[int] ): project_id = campaign.project_id if project_id: new_territories = [ HistoryTerritory(model) for model in self.campaigns_repository.get_campaign_project_territories(project_id) ] self.__log_project_territories_changed_if_needed( campaign, project_id, state.territories, new_territories, user_id ) self.history_service.log_campaign_deleted(campaign, user_id) def log_campaign_approved( self, campaign: Campaign, old_territories: CampaignTerritoriesHistoryState, user_id: Optional[int] ): self.history_service.log_campaign_approved(campaign, user_id) project_id = campaign.project_id new_territories = [ HistoryTerritory(model) for model in self.campaigns_repository.get_campaign_project_territories(project_id) ] self.__log_project_territories_changed_if_needed( campaign, campaign.project_id, old_territories.territories, new_territories, user_id ) self.campaigns_repository.save_campaign_changes(campaign) def log_campaign_rejected(self, campaign: Campaign, project: Project, user_id: Optional[int]): self.history_service.log_campaign_rejected(campaign, project, user_id) def __log_project_territories_changed_if_needed( self, campaign: Campaign, project_id: int, old_territories: List[HistoryTerritory], new_territories: List[HistoryTerritory], user_id: Optional[int], ): old_territories_ids = [territory.id for territory in old_territories] new_territories_ids = [territory.id for territory in new_territories] if list_utils.has_changes(old_territories_ids, new_territories_ids): removed_ids = set(old_territories_ids) - set(new_territories_ids) if removed_ids: removed_territories = list(filter(lambda t: t.id in removed_ids, old_territories)) self.history_service.log_campaign_countries_updated( campaign, project_id, removed_territories, [], user_id ) new_ids = set(new_territories_ids) - set(old_territories_ids) if new_ids: added_territories = list(filter(lambda t: t.id in new_ids, new_territories)) self.history_service.log_campaign_countries_updated( campaign, project_id, [], added_territories, user_id ) def __log_campaign_budget_changed_if_needed( self, campaign: Campaign, old_budget: float, new_budget: float, user_id: Optional[int] ): if old_budget != new_budget: self.history_service.log_campaign_budget_updated(campaign, old_budget, new_budget, user_id) def __log_campaign_schedule_changed_if_needed( self, campaign: Campaign, old_dates: DateRange, new_dates: DateRange, user_id: Optional[int] ): if old_dates.start != new_dates.start or old_dates.end != new_dates.end: self.history_service.log_campaign_schedule_updated(campaign, old_dates, new_dates, user_id) def __log_campaign_platforms_changed_if_needed( self, campaign: Campaign, old_platforms: Optional[List[str]], user_id: Optional[int] ): if sorted(old_platforms or []) != sorted(campaign.platform_value or []): self.history_service.log_campaign_platform_updated( campaign, old_platforms, campaign.platform_value, user_id ) def __log_campaign_objective_changed_if_needed( self, campaign: Campaign, old_value: HistoryObjective, new_value: HistoryObjective, user_id: Optional[int] ): if old_value != new_value: self.history_service.log_campaign_objective_updated(campaign, old_value, new_value, user_id) def __log_campaign_linkfire_links_changed_if_needed( self, campaign: Campaign, old_value: List[LinkfireLinkResponseSchema], new_value: List[LinkfireLinkResponseSchema], user_id: Optional[int], ): if old_value != new_value: self.history_service.log_campaign_linkfire_changes(campaign, old_value, new_value, user_id) def __log_campaign_changed_if_needed( self, campaign: Campaign, campaign_mask: CampaignHistoryState, user_id: Optional[int] ): old_values = [] new_values = [] campaign_mask_dict = Encoder.convert_to_dict(campaign_mask) expected_keys = campaign_mask_dict.keys() exclude_fields = [field for field in campaign_mask.exclude_fields()] old_values_list = [(key, getattr(campaign_mask, key, None)) for key in expected_keys if key not in exclude_fields] new_values_list = [(key, getattr(campaign, key, None)) for key in expected_keys if key not in exclude_fields] if list_utils.has_changes(old_values_list, new_values_list): old_values_set = set(old_values_list) - set(new_values_list) new_values_set = set(new_values_list) - set(old_values_list) old_values = [HistoryCampaignChanged(value[0], value[1]) for value in old_values_set] new_values = [HistoryCampaignChanged(value[0], value[1]) for value in new_values_set] old_placements = [p.id for p in campaign_mask.placements] new_placements = [p.id for p in campaign.placements] if list_utils.has_changes(old_placements, new_placements): old_values.append( HistoryCampaignChanged(Campaign.placements.key, old_placements) ) new_values.append( HistoryCampaignChanged(Campaign.placements.key, new_placements) ) if ( getattr(campaign.age_range, "lower", None) != campaign_mask.audience_lower_age or getattr(campaign.age_range, "upper", None) != campaign_mask.audience_higher_age ): old_values.append( HistoryCampaignChanged( key=Campaign.age_range.name, value=f"{campaign_mask.audience_lower_age} - {campaign_mask.audience_higher_age}" ) ) new_values.append( HistoryCampaignChanged( key=Campaign.age_range.name, value=f'{getattr(campaign.age_range, "lower", None)} - {getattr(campaign.age_range, "upper", None)}' ) ) new_genders = [g.id for g in campaign.genders] if list_utils.has_changes(campaign_mask.genders, new_genders): old_values.append( HistoryCampaignChanged( key=Campaign.genders.key, value=campaign_mask.genders ) ) new_values.append( HistoryCampaignChanged( key=Campaign.genders.key, value=new_genders ) ) new_links = [link.id for link in campaign.ad_creative_links] if list_utils.has_changes(campaign_mask.ad_creative_links, new_links): old_values.append( HistoryCampaignChanged( key=Campaign.ad_creative_links.key, value=campaign_mask.ad_creative_links ) ) new_values.append( HistoryCampaignChanged( key=Campaign.ad_creative_links.key, value=new_links ) ) new_destination_links = [link.id for link in campaign.links] if list_utils.has_changes(campaign_mask.links, new_destination_links): old_values.append( HistoryCampaignChanged( key=Campaign.links.key, value=campaign_mask.links ) ) new_values.append( HistoryCampaignChanged( key=Campaign.links.key, value=new_destination_links ) ) if new_values: self.history_service.log_campaign_updated(campaign, old_values, new_values, user_id) self.__log_campaign_platforms_changed_if_needed(campaign, campaign_mask.platforms, user_id) new_schedule = DateRange(campaign.start_date, campaign.fallback_end_date) self.__log_campaign_schedule_changed_if_needed(campaign, campaign_mask.schedule, new_schedule, user_id) self.__log_campaign_budget_changed_if_needed(campaign, campaign_mask.budget, campaign.planned_budget, user_id) current_objective = HistoryObjective(campaign) if campaign.objective else None self.__log_campaign_objective_changed_if_needed(campaign, campaign_mask.objective, current_objective, user_id) self.__log_campaign_linkfire_links_changed_if_needed( campaign, campaign_mask.linkfire_links, campaign.linkfire_campaigns, user_id )