import decimal from datetime import datetime from typing import TYPE_CHECKING, Optional from urllib.parse import urljoin import sqlalchemy as sa from audience_common import datetimeutc from audience_common.auth.account import Account from more_itertools import first_true from sqlalchemy.dialects.postgresql import ARRAY from sqlalchemy.orm import Mapped, mapped_column, relationship from campaigns.campaigns.enums import CampaignStatus from campaigns.connectors.db.orm import Base from campaigns.connectors.db.sqltypes import ChoiceType from campaigns.meta import constants from campaigns.meta.enums import ( CampaignCallToAction, CampaignContentType, CampaignLinkType, CampaignObjective, CampaignPlatform, CampaignPlatformSelector, ) from campaigns.meta.exceptions import ( CampaignImageDoesNotExist, CampaignVideoDoesNotExist, ) from campaigns.utils.uuid import uuid_string if TYPE_CHECKING: from .campaign_image import CampaignImage from .campaign_video import CampaignVideo from .engagement_audience import EngagementAudience from .facebook_page import FacebookPage from .targeting_audience import TargetingAudience class Campaign(Base): __tablename__ = "meta_campaign" id: Mapped[str] = mapped_column(primary_key=True, default=uuid_string) name: Mapped[str] vendor_id: Mapped[int] subaccount_id: Mapped[int] global_participant_id: Mapped[str] product_id: Mapped[int | None] product_upc: Mapped[str | None] objective: Mapped[CampaignObjective] = mapped_column(ChoiceType(CampaignObjective)) external_id: Mapped[str | None] = mapped_column(unique=True) external_status: Mapped[str | None] status: Mapped[CampaignStatus] = mapped_column( ChoiceType(CampaignStatus), default=CampaignStatus.DRAFT, ) facebook_page_id: Mapped[str] = mapped_column(sa.ForeignKey("facebook_page.id")) facebook_post_external_id: Mapped[str | None] instagram_account_external_id: Mapped[str | None] instagram_user_external_id: Mapped[str | None] instagram_post_external_id: Mapped[str | None] start_time: Mapped[datetime | None] end_time: Mapped[datetime | None] lifetime_budget: Mapped[decimal.Decimal | None] platforms: Mapped[list[CampaignPlatform] | None] = mapped_column( ARRAY(ChoiceType(CampaignPlatform)) ) content_type: Mapped[CampaignContentType] = mapped_column( ChoiceType(CampaignContentType), default=CampaignContentType.IMAGE, ) primary_text: Mapped[str | None] headline: Mapped[str | None] description: Mapped[str | None] link_type: Mapped[CampaignLinkType | None] = mapped_column( ChoiceType(CampaignLinkType) ) website_url: Mapped[str | None] landing_page_path: Mapped[str | None] landing_page_product_id: Mapped[int | None] call_to_action: Mapped[CampaignCallToAction | None] = mapped_column( ChoiceType(CampaignCallToAction) ) targeting_audience_id: Mapped[str | None] = mapped_column( sa.ForeignKey("meta_targeting_audience.id") ) engagement_audience_id: Mapped[str | None] = mapped_column( sa.ForeignKey("meta_engagement_audience.id") ) created_at: Mapped[datetime] = mapped_column(default=datetimeutc.now) created_by: Mapped[str] updated_at: Mapped[datetime] = mapped_column( default=datetimeutc.now, onupdate=datetimeutc.now ) updated_by: Mapped[str] # relations facebook_page: Mapped["FacebookPage"] = relationship(lazy="raise") targeting_audience: Mapped[Optional["TargetingAudience"]] = relationship( lazy="raise" ) engagement_audience: Mapped[Optional["EngagementAudience"]] = relationship( lazy="raise" ) images: Mapped[list["CampaignImage"]] = relationship( back_populates="campaign", lazy="raise" ) videos: Mapped[list["CampaignVideo"]] = relationship( back_populates="campaign", lazy="raise" ) def __str__(self) -> str: return self.name @property def account(self) -> Account: return Account(vendor_id=self.vendor_id, subaccount_id=self.subaccount_id) @property def is_draft(self) -> bool: return self.status == CampaignStatus.DRAFT @property def platform(self) -> CampaignPlatformSelector | None: if not self.platforms: return None return CampaignPlatformSelector.from_list(self.platforms) @property def post_external_id(self) -> str | None: return self.instagram_post_external_id or self.facebook_post_external_id def get_image(self, image_id: str) -> "CampaignImage": image = first_true(self.images, default=None, pred=lambda i: i.id == image_id) if not image: raise CampaignImageDoesNotExist return image def get_video(self, video_id: str) -> "CampaignVideo": video = first_true(self.videos, default=None, pred=lambda v: v.id == video_id) if not video: raise CampaignVideoDoesNotExist return video @property def has_valid_images(self) -> bool: return any(image.is_valid for image in self.images) @property def has_valid_videos(self) -> bool: return any(video.is_valid for video in self.videos) @property def landing_page_url(self) -> str | None: if self.landing_page_path: return urljoin(constants.LANDING_PAGE_BASE_URL, self.landing_page_path) return None @property def call_to_action_url(self) -> str | None: return ( self.website_url if self.link_type == CampaignLinkType.WEBSITE else self.landing_page_url )