import enum from typing import Literal, Self from pydantic import BaseModel, ConfigDict, Field, model_validator from dmp.audiences.enums import AudienceTarget from dmp.audiences.types import FanDataListId from dmp.fandata.enums import ( BirthMonth, DigitalServiceProvider, FanSegment, Genre, HeavyRotationFilter, LastEngagement, ) class Operator(enum.StrEnum): INCLUDE = "INCLUDE" EXCLUDE = "EXCLUDE" @property def as_sql(self) -> str: if self == self.EXCLUDE: return "NOT" return "" class EmailCampaignAction(enum.StrEnum): DELIVERED = "DELIVERED" OPENED = "OPENED" CLICKED = "CLICKED" class OperatorFilter[T](BaseModel): value: T operator: Operator class DigitalServiceProviderInFilter(OperatorFilter[list[DigitalServiceProvider]]): pass class LastEngagementFilter(OperatorFilter[LastEngagement]): pass class FanSegmentInFilter(OperatorFilter[list[FanSegment]]): value: list[FanSegment] class SongwhipPresavePageInFilter(OperatorFilter[list[str]]): value: list[str] class SongwhipCampaignInFilter(OperatorFilter[list[str]]): value: list[str] class CrmCampaignFilter(OperatorFilter[list[str]]): value: list[str] class CrmSublistFilter(OperatorFilter[list[str]]): value: list[str] class EmailCampaignsFilter(OperatorFilter[list[str]]): value: list[str] action: EmailCampaignAction class HeavyRotationV2InFilter(OperatorFilter[list[HeavyRotationFilter]]): value: list[HeavyRotationFilter] class FanGenreInFilter(OperatorFilter[list[Genre]]): pass class FanLanguageInFilter(OperatorFilter[list[str]]): pass class BirthMonthInFilter(OperatorFilter[list[BirthMonth]]): value: list[BirthMonth] class RegionFilter(BaseModel): place_id: str = Field(alias="placeId") title: str country_code: str = Field(alias="countryCode") region_code: str | None = Field(None, alias="regionCode") model_config = ConfigDict(populate_by_name=True) @property def stored_value(self) -> str: return f"{self.title}, {self.country_code}" class CityFilter(BaseModel): place_id: str = Field(alias="placeId") title: str | None = None country_iso2: str | None = Field(None, alias="countryIso2") city_id: str | None = Field(None, alias="cityId") postal_code: str | None = Field(None, alias="postalCode") latitude: float | None = None longitude: float | None = None bounding_radius_in_km: float | None = Field(None, alias="boundingRadiusInKm") model_config = ConfigDict(populate_by_name=True) @property def is_clean(self) -> bool: return self.country_iso2 is not None and self.city_id is not None class AudienceFilters(BaseModel): countries: list[str] | None = Field(None, examples=[["GB"]]) exclude_countries: bool | None = Field(None, alias="excludeCountries") cities: list[CityFilter] | None = None radius: float | None = Field(None, alias="radius") radius_unit: Literal["MI", "KM"] | None = Field(None, alias="radiusUnit") spend_min: float | None = Field(None, alias="spendMin") spend_max: float | None = Field(None, alias="spendMax") heavy_rotation_v2: HeavyRotationV2InFilter | None = Field( None, alias="heavyRotationV2" ) dsp: DigitalServiceProviderInFilter | None = None last_engagement: LastEngagementFilter | None = Field(None, alias="lastEngagement") fan_segments: FanSegmentInFilter | None = Field(None, alias="fanSegments") include_secondary_fans: bool | None = Field(None, alias="includeSecondaryFans") songwhip_presave_page_ids: SongwhipPresavePageInFilter | None = Field( None, alias="songwhipPresavePageIds" ) songwhip_campaign_ids: SongwhipCampaignInFilter | None = Field( None, alias="songwhipCampaignIds" ) crm_campaigns: CrmCampaignFilter | None = Field(None, alias="crmCampaigns") crm_sublists: CrmSublistFilter | None = Field(None, alias="crmSublists") email_campaigns: EmailCampaignsFilter | None = Field(None, alias="emailCampaigns") genres: FanGenreInFilter | None = None fan_languages: FanLanguageInFilter | None = Field(None, alias="fanLanguages") birth_months: BirthMonthInFilter | None = Field(None, alias="birthMonths") regions: list[RegionFilter] | None = None excluded_fan_data_lists: list[FanDataListId] | None = Field( None, alias="excludedFanDataLists" ) model_config = ConfigDict(populate_by_name=True) @property def radius_in_km(self) -> float | None: if self.radius is None: return None if self.radius_unit == "MI": return self.radius * 1.60934 return self.radius @property def city_ids(self) -> list[str]: return [ city_filter.city_id for city_filter in self.cities or [] if city_filter.city_id is not None ] @property def city_countries(self) -> list[str]: return [ city_filter.country_iso2 for city_filter in self.cities or [] if city_filter.country_iso2 is not None ] def set_include_secondary_fans(self, target: AudienceTarget) -> None: if self.include_secondary_fans is not None: return # Do not set include_secondary_fans for audience targets other than ads if target != AudienceTarget.ADS: return # Default to include secondary fans for ads only if not self.fan_segments or not self.fan_segments.value: self.include_secondary_fans = True return include_secondary_fans = FanSegment.SECONDARY_FANS in self.fan_segments.value if self.fan_segments.operator == Operator.INCLUDE: self.include_secondary_fans = include_secondary_fans if self.fan_segments.value == [FanSegment.SECONDARY_FANS]: self.fan_segments.operator = Operator.EXCLUDE self.fan_segments.value = [ segment for segment in FanSegment if segment not in {FanSegment.SECONDARY_FANS, FanSegment.UNSEGMENTED_FANS} ] else: # EXCLUDE self.include_secondary_fans = not include_secondary_fans @model_validator(mode="after") def backfill_songwhip_campaign_ids(self) -> Self: if ( self.songwhip_campaign_ids is None and self.songwhip_presave_page_ids is not None ): self.songwhip_campaign_ids = SongwhipCampaignInFilter( value=self.songwhip_presave_page_ids.value, operator=self.songwhip_presave_page_ids.operator, ) return self @property def artists_to_exclude(self) -> list[str]: if not self.excluded_fan_data_lists: return [] return [ item.id for item in self.excluded_fan_data_lists if item.type == "ARTIST" ] @property def custom_lists_to_exclude(self) -> list[str]: if not self.excluded_fan_data_lists: return [] return [ item.id for item in self.excluded_fan_data_lists if item.type == "CUSTOM_LIST" ]