"""Project to project type detail item reference model.""" from enum import Enum from datetime import datetime, timezone from typing import Optional, List, Type from sqlalchemy import func, not_, text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.hybrid import hybrid_property from models.campaign_approval import ApprovalRequestCampaign from models.gender import Gender from models.campaign_ad import CampaignAd from constants.project_statuses import ( ProjectStatuses, PROJECT_STATUSES_MAPPING, ) from handlers.campaign_status_handler import ( DraftTransition, LiveTransition, BookedTransition, PausedTransition, CompletedTransition, PendingApprovalTransition, ApprovedTransition, DisapprovedTransition ) from handlers.enums import CampaignWorkflowStatus from db import db, fsm from sqlalchemy.dialects import postgresql from models.mixins.searchable import SearchableMixin CAMPAIGN_TYPE_MAX_LEN = 100 class CampaignSourceType(Enum): GOOGLE = "google" FACEBOOK = "facebook" TIKTOK = "tiktok" PRS = "PRS" INTERNAL = "dec" @staticmethod def all_sources() -> List[str]: # noinspection PyTypeChecker return list(map(lambda c: c.value, CampaignSourceType)) @staticmethod def all_external_sources() -> List[str]: # noinspection PyTypeChecker return [CampaignSourceType.GOOGLE.value, CampaignSourceType.FACEBOOK.value, CampaignSourceType.TIKTOK.value] @staticmethod def all_internal_sources(): # noinspection PyTypeChecker return [CampaignSourceType.PRS.value, CampaignSourceType.INTERNAL.value] class Campaign(SearchableMixin, db.Model): """Project to project type detail item reference model.""" __tablename__ = "Campaign" __searchable__ = { "name": {"type": "text", "analyzer": "autocomplete", "search_analyzer": "standard"}, "external_id": {"type": "text"}, } id = db.Column(db.BigInteger, primary_key=True) external_id = db.Column(db.String, unique=True, nullable=True) name = db.Column(db.String, nullable=True) naming_convention = db.Column(db.String, nullable=True) naming_convention_manual = db.Column(db.String, nullable=True) external_name = db.Column(db.String, nullable=True) start_date = db.Column(db.Date, nullable=True) end_date = db.Column(db.Date, nullable=True) planned_budget = db.Column(db.Numeric(12, 2), nullable=True) budget_spend = db.Column(db.Numeric(12, 2), nullable=True) actual_spent = db.Column(db.Numeric(12, 2), nullable=True) creative_description = db.Column(db.String, nullable=True) headline = db.Column(db.String, nullable=True) call_to_action = db.Column(db.String, nullable=True) is_deleted = db.Column(db.Boolean, nullable=False, server_default="FALSE") notes = db.Column(db.String, nullable=True) age_range = db.Column(postgresql.INT4RANGE, nullable=True) ecpm = db.Column(db.DECIMAL, nullable=True) audience_notes = db.Column(db.String, nullable=True) # deprecated remove this column after CampaignCreativeLink table is on prod ad_creative_link = db.Column(db.String, nullable=True) ad_creative_notes = db.Column(db.String, nullable=True) objective_id = db.Column(db.Integer, db.ForeignKey("CampaignObjective.id"), nullable=True, index=True) project_id = db.Column(db.BigInteger, db.ForeignKey("Project.id", ondelete="CASCADE"), nullable=True, index=True) currency_id = db.Column(db.Integer, db.ForeignKey("Currency.id"), nullable=True) provider_id = db.Column(db.Integer, db.ForeignKey("CampaignProvider.id"), nullable=True, index=True) source = db.Column(db.String, nullable=True) kpi_metrics_field_id = db.Column(db.Integer, db.ForeignKey("MetricsField.id"), nullable=True) created_at = db.Column(db.TIMESTAMP(timezone=True), nullable=False, server_default=func.current_timestamp()) updated_at = db.Column(db.TIMESTAMP(timezone=True), nullable=False, server_default=func.current_timestamp()) create_user_id = db.Column(db.Integer, db.ForeignKey("User.id"), nullable=True) edit_user_id = db.Column(db.Integer, db.ForeignKey("User.id"), nullable=True) type_id = db.Column(db.Integer, db.ForeignKey("CampaignTypes.id"), nullable=True) gratis = db.Column(db.Boolean, nullable=True) # metainfo controls the possibility to edit columns in created campaign metainfo = db.Column(db.JSON, nullable=False, server_default="{}") style = db.Column(JSONB, nullable=True) external_marketing_account_id = db.Column( db.String, db.ForeignKey("MarketingAccount.external_id"), nullable=True, index=True ) purchase_order_id = db.Column(db.BigInteger, db.ForeignKey("PRSPurchaseOrder.id"), nullable=True, index=True) media_plan_phase_id = db.Column( db.Integer, db.ForeignKey("MediaPlanPhase.id", ondelete="RESTRICT"), nullable=True, index=True ) order_in_phase = db.Column(db.SmallInteger, nullable=True) uuid = db.Column( postgresql.UUID(as_uuid=True), nullable=False, unique=True, server_default=text("uuid_generate_v4()") ) workflow_status = db.Column(fsm.FSMField, nullable=False, server_default='Draft') create_user = db.relationship("User", foreign_keys=[create_user_id]) edit_user = db.relationship("User", foreign_keys=[edit_user_id]) project = db.relationship("Project", back_populates="campaigns") currency = db.relationship("Currency") territories = db.relationship("CampaignTerritory", back_populates="campaign") history = db.relationship("ProjectHistoryItem") objective = db.relationship("CampaignObjective") provider = db.relationship("CampaignProvider") genders = db.relationship("Gender", secondary="CampaignGender", order_by=Gender.id) links: List["CampaignLink"] = db.relationship("CampaignLink", cascade="all, delete-orphan") ad_set = db.relationship("CampaignAd", back_populates="campaign") project_campaigns = db.relationship("ProjectCampaign", back_populates="campaign") marketing_account = db.relationship("MarketingAccount") purchase_order = db.relationship("PRSPurchaseOrder", foreign_keys=[purchase_order_id]) linkfire_campaigns = db.relationship("LinkfireCampaign", back_populates="campaign") type = db.relationship("CampaignTypes") platforms = db.relationship("CampaignPlatforms", secondary="CampaignPlatformsLinks") placements = db.relationship("CampaignPlacements", secondary="CampaignPlacementsLinks") ad_creative_links: List["CampaignCreativeLink"] = db.relationship("CampaignCreativeLink", back_populates="campaign") phase = db.relationship("MediaPlanPhase", back_populates="campaigns") kpi_metrics_field = db.relationship("MetricsField") approval_request_campaigns: List[ApprovalRequestCampaign] = db.relationship( "ApprovalRequestCampaign", back_populates="campaign" ) draft_transition_handler = DraftTransition live_transition_handler = LiveTransition booked_transition_handler = BookedTransition paused_transition_handler = PausedTransition completed_transition_handler = CompletedTransition pending_approval_transition_handler = PendingApprovalTransition approved_transition_handler = ApprovedTransition disapproved_transition_handler = DisapprovedTransition __table_args__ = ( db.UniqueConstraint( "media_plan_phase_id", "order_in_phase", name="uq_Campaign_media_plan_phase_id_order_in_phase", deferrable=True, initially="deferred", ), db.UniqueConstraint( "uuid", name="uq_Campaign_uuid", ), db.CheckConstraint( sqltext="not ((source != 'dec' and source is not null) and media_plan_phase_id is not null)", name="check_only_internal_campaigns_have_phase" ) ) @classmethod def draft_campaign_filters(cls): except_list = [cls.id, cls.workflow_status, cls.project_id, cls.media_plan_phase_id, cls.order_in_phase, cls.source, cls.created_at, cls.updated_at, cls.edit_user_id, cls.create_user_id, cls.is_deleted, cls.metainfo, cls.uuid, cls.style] except_list_names = [field.name for field in except_list] relationship_list = [cls.links, cls.territories, cls.platforms, cls.placements, cls.ad_creative_links, cls.genders] filters = [Campaign.workflow_status == CampaignWorkflowStatus.DRAFT.value] for col in cls.__table__.columns: if col.name not in except_list_names: if col.nullable: filters.append(col.is_(None)) filters.extend([not_(rel.any()) for rel in relationship_list]) return filters def touch(self): self.updated_at = datetime.now(timezone.utc) @classmethod def es_query(cls: Type[db.Model]): return cls.query.filter_by(is_deleted=False) @property def is_editable(self) -> bool: return ( self.external_id is None or self.source is None or self.source in CampaignSourceType.all_internal_sources() ) @property def status_id(self) -> int: """Status ID of the campaign. Campaign status is determined by start and end dates. Possible values are the same as for projects: Scheduled - 1, In Progress - 2, Completed - 3. Returns: Campaign status ID. """ now = datetime.now(timezone.utc).date() if self.end_date is not None and now > self.end_date: return ProjectStatuses.COMPLETED.value if self.start_date is not None and now >= self.start_date: return ProjectStatuses.INPROGRESS.value return ProjectStatuses.SCHEDULED.value @property def status(self) -> str: """String representation of campaign status. Returns: Campaign status name. """ return PROJECT_STATUSES_MAPPING[self.status_id] @property def platform_value(self) -> Optional[List[str]]: return [platform.name for platform in self.platforms] @property def formatted_source(self) -> str: if not self.source or self.source == CampaignSourceType.INTERNAL.value: return "Manual" return self.source @property def created(self): """History record of campaign creation. Returns: ProjectHistory """ return min(self.history, key=lambda i: i.created_at) @property def last_update(self): """History record of the most recent campaign update. Returns: ProjectHistory """ return max(self.history, key=lambda i: i.created_at) @property def territories_names(self): return " ".join(list(map(lambda campaign_territory: campaign_territory.territory.name, self.territories))) @hybrid_property def ads_count(self): return len(self.ad_set) @ads_count.expression def ads_count(cls): return func.count(CampaignAd.id) @property def fallback_end_date(self): if self.project: return self.end_date or self.project.end_date return self.end_date @property def current_approval_request_campaign(self) -> Optional[ApprovalRequestCampaign]: def get_current_filter(x): return x.is_live current_approvals_list = list(filter(get_current_filter, self.approval_request_campaigns)) if current_approvals_list: return current_approvals_list[0] else: return None @property def label_id(self): if self.project: return self.project.label_id else: return self.marketing_account.label_id campaign_gender = db.Table( "CampaignGender", db.Model.metadata, db.Column("campaign_id", db.Integer, db.ForeignKey("Campaign.id", ondelete="CASCADE"), nullable=False), db.Column("gender_id", db.Integer, db.ForeignKey("Gender.id", ondelete="CASCADE"), nullable=False), ) class CampaignTypeGroup(db.Model): """Campaign type group.""" __tablename__ = "CampaignTypeGroup" id = db.Column(db.SmallInteger, primary_key=True) name = db.Column(db.String(50), nullable=False) is_hidden = db.Column(db.Boolean, nullable=False, server_default="false") user_projects = db.relationship("UserProject", secondary="UserProjectCampaignTypeGroup") artist_team_users = db.relationship("ArtistTeamUser", secondary="ArtistTeamUserCampaignTypeGroup") campaign_types = db.relationship("CampaignTypes") class CampaignLink(db.Model): """Campaign link.""" __tablename__ = "CampaignLink" id = db.Column(db.Integer, primary_key=True, autoincrement=True) campaign_id = db.Column(db.Integer, db.ForeignKey("Campaign.id", ondelete="CASCADE"), nullable=False) link = db.Column(db.String) campaign = db.relationship("Campaign") class CampaignTypes(db.Model): """Campaign Types""" __tablename__ = "CampaignTypes" id = db.Column(db.Integer, primary_key=True, autoincrement=True) name = db.Column(db.String, nullable=False, index=True) group_id = db.Column(db.Integer, db.ForeignKey("CampaignTypeGroup.id"), nullable=False) group = db.relationship("CampaignTypeGroup") campaign_platforms_placements = db.Table( "CampaignPlatformPlacements", db.Model.metadata, db.Column("platform_id", db.Integer, db.ForeignKey("CampaignPlatforms.id", ondelete="CASCADE"), nullable=False), db.Column("placement_id", db.Integer, db.ForeignKey("CampaignPlacements.id", ondelete="CASCADE"), nullable=False), ) class CampaignPlatforms(db.Model): """Campaign Platforms""" __tablename__ = "CampaignPlatforms" id = db.Column(db.Integer, primary_key=True, autoincrement=True) name = db.Column(db.String, nullable=False, index=True) label_id = db.Column(db.Integer, db.ForeignKey("Label.id", ondelete="CASCADE"), nullable=True) common = db.Column(db.Boolean, nullable=False, server_default="false") platform_group_id = db.Column( db.Integer, db.ForeignKey("CampaignPlatformGroups.id", ondelete="SET NULL"), nullable=True ) platform_group = db.relationship("CampaignPlatformGroups") placements = db.relationship("CampaignPlacements", secondary="CampaignPlatformPlacements") campaign = db.relationship("Campaign", secondary="CampaignPlatformsLinks") campaign_platforms_links = db.Table( "CampaignPlatformsLinks", db.Model.metadata, db.Column("campaign_id", db.Integer, db.ForeignKey("Campaign.id", ondelete="CASCADE"), nullable=False), db.Column("platform_id", db.Integer, db.ForeignKey("CampaignPlatforms.id", ondelete="CASCADE"), nullable=False), ) class CampaignPlatformGroups(db.Model): """Campaign Platform Groups""" __tablename__ = "CampaignPlatformGroups" id = db.Column(db.Integer, primary_key=True, autoincrement=True) name = db.Column(db.String(50), nullable=False) platforms: List["CampaignPlatforms"] = db.relationship("CampaignPlatforms", back_populates="platform_group") class CampaignPlacements(db.Model): """Campaign Placements""" __tablename__ = "CampaignPlacements" id = db.Column(db.Integer, primary_key=True, autoincrement=True) name = db.Column(db.String, nullable=False) label_id = db.Column(db.Integer, db.ForeignKey("Label.id"), nullable=True) campaign_placements_links = db.Table( "CampaignPlacementsLinks", db.Model.metadata, db.Column("campaign_id", db.Integer, db.ForeignKey("Campaign.id", ondelete="CASCADE"), nullable=False), db.Column("placement_id", db.Integer, db.ForeignKey("CampaignPlacements.id", ondelete="CASCADE"), nullable=False), ) class CampaignCreativeLink(db.Model): """Campaign Ad Creative Link model""" __tablename__ = "CampaignCreativeLink" id = db.Column(db.Integer, primary_key=True, autoincrement=True) campaign_id = db.Column(db.Integer, db.ForeignKey("Campaign.id", ondelete="CASCADE"), nullable=False) url = db.Column(db.String, nullable=False) campaign = db.relationship("Campaign", back_populates="ad_creative_links")